How to use assert_sets_equal method in Testify

Best Python code snippet using Testify_python

pre_commit_check_provider_yaml_files.py

Source:pre_commit_check_provider_yaml_files.py Github

copy

Full Screen

...83 "Please delete duplicates."84 )85 print(tabulate(duplicates, headers=["Integration name", "Number of occurrences"]))86 sys.exit(3)87def assert_sets_equal(set1, set2):88 try:89 difference1 = set1.difference(set2)90 except TypeError as e:91 raise AssertionError(f'invalid type when attempting set difference: {e}')92 except AttributeError as e:93 raise AssertionError(f'first argument does not support set difference: {e}')94 try:95 difference2 = set2.difference(set1)96 except TypeError as e:97 raise AssertionError(f'invalid type when attempting set difference: {e}')98 except AttributeError as e:99 raise AssertionError(f'second argument does not support set difference: {e}')100 if not (difference1 or difference2):101 return102 lines = []103 if difference1:104 lines.append(' -- Items in the left set but not the right:')105 for item in sorted(difference1):106 lines.append(f' {item!r}')107 if difference2:108 lines.append(' -- Items in the right set but not the left:')109 for item in sorted(difference2):110 lines.append(f' {item!r}')111 standard_msg = '\n'.join(lines)112 raise AssertionError(standard_msg)113def check_if_objects_belongs_to_package(114 object_names: Set[str], provider_package: str, yaml_file_path: str, resource_type: str115):116 for object_name in object_names:117 if not object_name.startswith(provider_package):118 errors.append(119 f"The `{object_name}` object in {resource_type} list in {yaml_file_path} does not start"120 f" with the expected {provider_package}."121 )122def parse_module_data(provider_data, resource_type, yaml_file_path):123 package_dir = ROOT_DIR + "/" + os.path.dirname(yaml_file_path)124 provider_package = os.path.dirname(yaml_file_path).replace(os.sep, ".")125 py_files = chain(126 glob(f"{package_dir}/**/{resource_type}/*.py"),127 glob(f"{package_dir}/{resource_type}/*.py"),128 glob(f"{package_dir}/**/{resource_type}/**/*.py"),129 glob(f"{package_dir}/{resource_type}/**/*.py"),130 )131 expected_modules = {_filepath_to_module(f) for f in py_files if not f.endswith("/__init__.py")}132 resource_data = provider_data.get(resource_type, [])133 return expected_modules, provider_package, resource_data134def check_completeness_of_list_of_hooks_sensors_hooks(yaml_files: Dict[str, Dict]):135 print("Checking completeness of list of {sensors, hooks, operators}")136 print(" -- {sensors, hooks, operators} - Expected modules(Left): Current Modules(Right)")137 for (yaml_file_path, provider_data), resource_type in product(138 yaml_files.items(), ["sensors", "operators", "hooks"]139 ):140 expected_modules, provider_package, resource_data = parse_module_data(141 provider_data, resource_type, yaml_file_path142 )143 current_modules = {str(i) for r in resource_data for i in r.get('python-modules', [])}144 check_if_objects_belongs_to_package(current_modules, provider_package, yaml_file_path, resource_type)145 try:146 assert_sets_equal(set(expected_modules), set(current_modules))147 except AssertionError as ex:148 nested_error = textwrap.indent(str(ex), ' ')149 errors.append(150 f"Incorrect content of key '{resource_type}/python-modules' "151 f"in file: {yaml_file_path}\n{nested_error}"152 )153def check_duplicates_in_integrations_names_of_hooks_sensors_operators(yaml_files: Dict[str, Dict]):154 print("Checking for duplicates in list of {sensors, hooks, operators}")155 for (yaml_file_path, provider_data), resource_type in product(156 yaml_files.items(), ["sensors", "operators", "hooks"]157 ):158 resource_data = provider_data.get(resource_type, [])159 current_integrations = [r.get("integration-name", "") for r in resource_data]160 if len(current_integrations) != len(set(current_integrations)):161 for integration in current_integrations:162 if current_integrations.count(integration) > 1:163 errors.append(164 f"Duplicated content of '{resource_type}/integration-name/{integration}' "165 f"in file: {yaml_file_path}"166 )167def check_completeness_of_list_of_transfers(yaml_files: Dict[str, Dict]):168 print("Checking completeness of list of transfers")169 resource_type = 'transfers'170 print(" -- Expected transfers modules(Left): Current transfers Modules(Right)")171 for yaml_file_path, provider_data in yaml_files.items():172 expected_modules, provider_package, resource_data = parse_module_data(173 provider_data, resource_type, yaml_file_path174 )175 current_modules = {r.get('python-module') for r in resource_data}176 check_if_objects_belongs_to_package(current_modules, provider_package, yaml_file_path, resource_type)177 try:178 assert_sets_equal(set(expected_modules), set(current_modules))179 except AssertionError as ex:180 nested_error = textwrap.indent(str(ex), ' ')181 errors.append(182 f"Incorrect content of key '{resource_type}/python-module' "183 f"in file: {yaml_file_path}\n{nested_error}"184 )185def check_hook_classes(yaml_files: Dict[str, Dict]):186 print("Checking connection classes belong to package")187 resource_type = 'hook-class-names'188 for yaml_file_path, provider_data in yaml_files.items():189 provider_package = os.path.dirname(yaml_file_path).replace(os.sep, ".")190 hook_class_names = provider_data.get(resource_type)191 if hook_class_names:192 check_if_objects_belongs_to_package(193 hook_class_names, provider_package, yaml_file_path, resource_type194 )195def check_duplicates_in_list_of_transfers(yaml_files: Dict[str, Dict]):196 print("Checking for duplicates in list of transfers")197 errors = []198 resource_type = "transfers"199 for yaml_file_path, provider_data in yaml_files.items():200 resource_data = provider_data.get(resource_type, [])201 source_target_integrations = [202 (r.get("source-integration-name", ""), r.get("target-integration-name", ""))203 for r in resource_data204 ]205 if len(source_target_integrations) != len(set(source_target_integrations)):206 for integration_couple in source_target_integrations:207 if source_target_integrations.count(integration_couple) > 1:208 errors.append(209 f"Duplicated content of \n"210 f" '{resource_type}/source-integration-name/{integration_couple[0]}' "211 f" '{resource_type}/target-integration-name/{integration_couple[1]}' "212 f"in file: {yaml_file_path}"213 )214def check_invalid_integration(yaml_files: Dict[str, Dict]):215 print("Detect unregistered integrations")216 all_integration_names = set(get_all_integration_names(yaml_files))217 for (yaml_file_path, provider_data), resource_type in product(218 yaml_files.items(), ["sensors", "operators", "hooks"]219 ):220 resource_data = provider_data.get(resource_type, [])221 current_names = {r['integration-name'] for r in resource_data}222 invalid_names = current_names - all_integration_names223 if invalid_names:224 errors.append(225 f"Incorrect content of key '{resource_type}/integration-name' in file: {yaml_file_path}. "226 f"Invalid values: {invalid_names}"227 )228 for (yaml_file_path, provider_data), key in product(229 yaml_files.items(), ['source-integration-name', 'target-integration-name']230 ):231 resource_data = provider_data.get('transfers', [])232 current_names = {r[key] for r in resource_data}233 invalid_names = current_names - all_integration_names234 if invalid_names:235 errors.append(236 f"Incorrect content of key 'transfers/{key}' in file: {yaml_file_path}. "237 f"Invalid values: {invalid_names}"238 )239def check_doc_files(yaml_files: Dict[str, Dict]):240 print("Checking doc files")241 current_doc_urls: List[str] = []242 current_logo_urls: List[str] = []243 for provider in yaml_files.values():244 if 'integrations' in provider:245 current_doc_urls.extend(246 guide247 for guides in provider['integrations']248 if 'how-to-guide' in guides249 for guide in guides['how-to-guide']250 )251 current_logo_urls.extend(252 integration['logo'] for integration in provider['integrations'] if 'logo' in integration253 )254 if 'transfers' in provider:255 current_doc_urls.extend(256 op['how-to-guide'] for op in provider['transfers'] if 'how-to-guide' in op257 )258 expected_doc_urls = {259 "/docs/" + os.path.relpath(f, start=DOCS_DIR)260 for f in glob(f"{DOCS_DIR}/apache-airflow-providers-*/operators/**/*.rst", recursive=True)261 if not f.endswith("/index.rst") and '/_partials' not in f262 }263 expected_doc_urls |= {264 "/docs/" + os.path.relpath(f, start=DOCS_DIR)265 for f in glob(f"{DOCS_DIR}/apache-airflow-providers-*/operators.rst", recursive=True)266 }267 expected_logo_urls = {268 "/" + os.path.relpath(f, start=DOCS_DIR)269 for f in glob(f"{DOCS_DIR}/integration-logos/**/*", recursive=True)270 if os.path.isfile(f)271 }272 try:273 print(" -- Checking document urls: expected(left), current(right)")274 assert_sets_equal(set(expected_doc_urls), set(current_doc_urls))275 print(" -- Checking logo urls: expected(left), current(right)")276 assert_sets_equal(set(expected_logo_urls), set(current_logo_urls))277 except AssertionError as ex:278 print(ex)279 sys.exit(1)280def check_unique_provider_name(yaml_files: Dict[str, Dict]):281 provider_names = [d['name'] for d in yaml_files.values()]282 duplicates = {x for x in provider_names if provider_names.count(x) > 1}283 if duplicates:284 errors.append(f"Provider name must be unique. Duplicates: {duplicates}")285def check_providers_are_mentioned_in_issue_template(yaml_files: Dict[str, Dict]):286 prefix_len = len("apache-airflow-providers-")287 short_provider_names = [d['package-name'][prefix_len:] for d in yaml_files.values()]288 jsonpath_expr = parse('$.body[?(@.attributes.label == "Apache Airflow Provider(s)")]..options[*]')289 with open(PROVIDER_ISSUE_TEMPLATE_PATH) as issue_file:290 issue_template = yaml.safe_load(issue_file)291 all_mentioned_providers = [match.value for match in jsonpath_expr.find(issue_template)]292 try:293 print(294 f" -- Checking providers: present in code(left), "295 f"mentioned in {PROVIDER_ISSUE_TEMPLATE_PATH} (right)"296 )297 assert_sets_equal(set(short_provider_names), set(all_mentioned_providers))298 except AssertionError as ex:299 print(ex)300 sys.exit(1)301def check_providers_have_all_documentation_files(yaml_files: Dict[str, Dict]):302 expected_files = ["commits.rst", "index.rst", "installing-providers-from-sources.rst"]303 for package_info in yaml_files.values():304 package_name = package_info['package-name']305 provider_dir = os.path.join(DOCS_DIR, package_name)306 for file in expected_files:307 if not os.path.isfile(os.path.join(provider_dir, file)):308 errors.append(309 f"The provider {package_name} misses `{file}` in documentation. "310 f"Please add the file to {provider_dir}"311 )...

Full Screen

Full Screen

varbindings_test.py

Source:varbindings_test.py Github

copy

Full Screen

...16# 02110-1301, USA.17import compiler18import unittest19import varbindings20def assert_sets_equal(actual, expected):21 if sorted(actual) != sorted(expected):22 missing = [x for x in expected if x not in actual]23 excess = [x for x in actual if x not in expected]24 raise AssertionError("Sets don't match:\nexpected: %r\nactual: %r\n\n"25 "missing: %r\nexcess: %r" %26 (expected, actual, missing, excess))27parse_statement = compiler.parse28def find_expected_bindings(source_text):29 for index, line in enumerate(source_text.split("\n")):30 line_vars = {}31 if "VAR:" in line:32 bindings = line.split("VAR:", 1)[1]33 for binding in bindings.split(","):34 var_name, var_id = binding.strip().split(":", 1)35 assert var_name not in line_vars36 line_vars[var_name] = var_id37 yield line_vars38# Could be replaced with use of the bindings list returned by annotate().39def find_actual_bindings(tree):40 for node in varbindings.iter_nodes(tree):41 if hasattr(node, "bindings"):42 for binding in node.bindings:43 yield (binding, node.lineno)44class LintTest(unittest.TestCase):45 def check_find_assigned(self, expected_assigned_vars, code_text):46 tree = parse_statement(code_text)47 assert_sets_equal(varbindings.find_assigned(tree),48 set(expected_assigned_vars))49 global_vars, bindings = varbindings.annotate(tree)50 assigned_vars = set(var_name51 for var_name, binding in global_vars.iteritems()52 if binding.is_assigned)53 assert_sets_equal(assigned_vars, set(expected_assigned_vars))54 def test_find_assigned(self):55 self.check_find_assigned(["x", "y", "a", "b", "a2", "b2", "c", "d", "e",56 "func", "Class", "element", "exn"], """57x = foo158y = foo259a, b = foo360[a2, b2] = foo361c += foo362d = e = foo463def func(arg):64 f = foo565class Class(object):66 method = foo667for element in [1, 2, 3]:68 print element69try:70 pass71except Exception, exn:72 pass73""")74 self.check_find_assigned(["x", "y"], """75del x, y76""")77 self.check_find_assigned([], """78lambda x: x + 179""")80 self.check_find_assigned(["foo1", "foo2", "foo3",81 "a1", "a2", "a3", "a4",82 "name1", "name2", "name3", "name4"], """83import foo184import foo2.bar85import bar.baz as foo386# These can also be combined into a single import statement.87import blah1 as a1, blah2 as a2, a3.bar, a488from bar.baz import name1, name289from bar.baz import qux as name3, quux as name490from other_module import *91""")92 self.check_find_assigned(["x", "y", "z1", "z2"], """93# Assigned variables in list comprehensions escape, for no good reason94# that I can see. There's a good reason for the assigned variable to95# escape from a "for" loop, but not for a list comprehension.96# Why can't the variable be bound, as with a lambda?97# The Python Reference Manual says "this behavior is deprecated, and98# relying on it will not work once this bug is fixed in a future release".99# (http://docs.python.org/ref/lists.html)100[x+1 for x, y in enumerate(range(100))]101[None for z1 in range(10)102 for z2 in range(10) if z1 % 2 == 0]103""")104 # Generators have different binding rules: they introduce new105 # scopes.106 self.check_find_assigned([], """107list(x+1 for x, y in enumerate(range(100)))108list(None for z1 in range(10)109 for z2 in range(10) if z1 % 2 == 0)110""")111 def test_read_variables(self):112 source = """113print read1114w1 = 1115readwrite += 1116del w2117def w3():118 return read2119class w4:120 pass121import w5122from xx import w6123"""124 global_vars, bindings = varbindings.annotate(parse_statement(source))125 assert_sets_equal(sorted(global_vars.iterkeys()),126 ["read1", "read2", "readwrite",127 "w1", "w2", "w3", "w4", "w5", "w6"])128 assert_sets_equal([var for var, binding in global_vars.iteritems()129 if binding.is_assigned],130 ["readwrite", "w1", "w2", "w3", "w4", "w5", "w6"])131 assert_sets_equal([var for var, binding in global_vars.iteritems()132 if binding.is_read],133 ["readwrite", "read1", "read2"])134 def test_find_globals(self):135 tree = parse_statement("""136global a137global b, c138if False:139 global d140while False:141 global e142for x in y:143 global f144def func(arg):145 global hidden1146class C:147 global hidden2148""")149 self.assertEquals(varbindings.find_globals(tree),150 set(["a", "b", "c", "d", "e", "f"]))151 # Check that find_globals() corresponds to Python's scoping behaviour.152 eval(compiler.compile("""153x = 1154def func():155 x = "not seen"156func()157assert x == 1, x158def func():159 x = 2160 global x161func()162assert x == 2, x163x = 1164def func():165 def func2():166 # This declaration does not escape to affect func().167 global x168 x = "not seen"169func()170assert x == 1, x171def func():172 global x173 def func2():174 x = 3175 func2()176func()177assert x == 1, x178def func():179 # This declaration does not leak into func2().180 global x181 def func2():182 x = 3183 func2()184func()185assert x == 1, x186def func(x):187 def func2():188 global x189 def func3():190 assert x == 1191 func3()192 func2()193func(100)194assert x == 1, x195def func():196 global x197 def func2(x):198 assert x == 100199 func2(100)200func()201""", "filename", "exec"), {})202 def test_free_vars(self):203 def free_vars(text):204 global_vars, bindings = varbindings.annotate(parse_statement(text))205 return set(global_vars.iterkeys())206 text = """207f(a.attr)208def func(v):209 f(v)210 v2 = 1211 f(v2)212 f(b)213 global c214 c = 1215d = 2216"""217 self.assertEquals(free_vars(text),218 set(["f", "a", "b", "c", "d", "func"]))219 text = """220lambda x: x221lambda x: freevar222"""223 self.assertEquals(free_vars(text), set(["freevar"]))224 text = """225[x for x in (1,2)] # Variable escapes226(y for y in (1,2)) # Variable does not escape227"""228 self.assertEquals(free_vars(text), set(["x"]))229 text = """230class C(object):231 x = 0232 def f(self):233 return x # global var234 y = 0235 z = y # from class scope236 k = k # from global scope into class scope237"""238 # TODO: k should be considered a FV239 self.assertEquals(free_vars(text), set(["object", "x", "C"]))240 def check_isomorphism(self, relation):241 mapping1 = {}242 mapping2 = {}243 for x, y in relation:244 mapping1.setdefault(x, set()).add(y)245 mapping2.setdefault(y, set()).add(x)246 for x, ys in mapping1.iteritems():247 assert len(ys) == 1, (x, ys)248 for y, xs in mapping2.iteritems():249 assert len(xs) == 1, (y, xs)250 def match_up_bindings(self, source):251 tree = parse_statement(source)252 varbindings.annotate(tree)253 got_vars = list(find_actual_bindings(tree))254 expected = list(find_expected_bindings(source))255 # Check that lines refer to the expected variable names.256 expected_var_lines = [(var_name, line_index + 1)257 for line_index, var_map in enumerate(expected)258 for var_name in var_map.keys()]259 actual_var_lines = [(binding.name, lineno)260 for binding, lineno in got_vars]261 assert_sets_equal(sorted(set(actual_var_lines)),262 sorted(set(expected_var_lines)))263 # Check 1-1 mapping.264 relation = []265 for binding, lineno in got_vars:266 var_id = expected[lineno - 1][binding.name]267 relation.append((var_id, binding))268 self.check_isomorphism(relation)269 def test_scoping(self):270 bad_source = """271# Annotations wrongly say they refer to different variables.272x = 1 # VAR: x:one_var273x + x # VAR: x:another_var274"""275 self.assertRaises(AssertionError,...

Full Screen

Full Screen

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.

LambdaTest Learning Hubs:

YouTube

You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.

Run Testify automation tests on LambdaTest cloud grid

Perform automation testing on 3000+ real desktop and mobile devices online.

Try LambdaTest Now !!

Get 100 minutes of automation test minutes FREE!!

Next-Gen App & Browser Testing Cloud

Was this article helpful?

Helpful

NotHelpful