def _collect_test_definition_models()

in tools/scripts/codegen/protocol_tests_gen.py [0:0]

• 50 lines of code
• 17 McCabe index (conditional complexity)

    def _collect_test_definition_models(self) -> dict:
        all_test_clients_md = self._get_client_models_metadata()

        test_models = dict()  # ex: "{input: {ec2: ProtocolTestModel}, output: {ec2: ProtocolTestModel}}"

        for test_def_group in PROTOCOL_TESTS_DEFINITION_SETS:
            model_files = os.listdir(f"{self.test_definitions_dir}/{test_def_group}")
            for filename in model_files:
                if not os.path.isfile(f"{self.test_definitions_dir}/{test_def_group}/{filename}"):
                    continue
                match = TEST_DEFINITION_FILENAME_PATTERN.match(filename)

                test_def_name = match.group("name")
                if test_def_name in UNSUPPORTED_TESTS:
                    print(f"Skipping protocol tests generation: {test_def_group}/{filename}")
                    continue

                test_def_path = str(pathlib.Path(f"{self.test_definitions_dir}/{test_def_group}/{filename}").resolve())

                def _get_corresponding_test_client(test_clients_md: list, test_path: str) -> list:
                    # Get c2j client models matching the test suite
                    # more than 1 is possible (ex: xml and xml with namespace clients for a single test suite)
                    result = list()
                    with open(test_path, 'r') as file_content:
                        try:
                            proto_test_model = json.load(file_content)
                            proto_test_md = proto_test_model[0].get("metadata")
                            for c2j_md in test_clients_md:
                                for field_to_match in ["apiVersion", "protocols", "jsonVersion", "targetPrefix"]:
                                    if proto_test_md.get(field_to_match, None) != c2j_md.md.get(field_to_match, None):
                                        break
                                else:
                                    result.append(c2j_md)
                        except Exception as exc:
                            print(f"ERROR: unexpected file content in protocol tests {test_def_path}. "
                                  f"Expected c2j protocol test, but json metadata kew is missing: {exc}")
                    return result

                test_clients_for_suite = _get_corresponding_test_client(all_test_clients_md, test_def_path)

                if test_clients_for_suite is None or len(test_clients_for_suite) == 0:
                    raise Exception(f"ERROR: Unable to find C2J client model for the test suite: {test_def_path}")

                for index, client_md in enumerate(test_clients_for_suite):
                    if index == 0:
                        test_def_key = test_def_name
                    else:
                        test_def_key = f"{test_def_name}-{index}"
                    assert test_models.get(test_def_group, dict()).get(test_def_key, None) is None, \
                        f"This test suite {test_def_group}/{test_def_key} already exists: {test_models}"
                    if self.debug:
                        print("Protocol test generation task:\t"
                              f"{test_def_path.split('/')[-1]} with {client_md.model_path.split('/')[-1]}")
                    if test_def_group not in test_models:
                        test_models[test_def_group] = dict()
                    test_models[test_def_group][test_def_key] = self.ProtocolTestModel(test_type=test_def_group,
                                                                                       test_name=test_def_key,
                                                                                       c2j_test_model=test_def_path,
                                                                                       c2j_client_md=client_md)
        return test_models