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

...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                )...

varbindings_test.py

Source:varbindings_test.py

...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,...

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.