From 9d21d1c143f7231b3c257125dcf794846dc77550 Mon Sep 17 00:00:00 2001 From: Aleksei Chetaev Date: Wed, 15 Feb 2023 19:24:14 +0100 Subject: [PATCH] Implement test analitics export into TMS systems Signed-off-by: Aleksei Chetaev --- requirements.txt | 2 + src/frostfs_testlib/analytics/__init__.py | 4 + src/frostfs_testlib/analytics/test_case.py | 82 ++++++++ .../analytics/test_collector.py | 190 ++++++++++++++++++ .../analytics/test_exporter.py | 71 +++++++ .../analytics/testrail_exporter.py | 178 ++++++++++++++++ 6 files changed, 527 insertions(+) create mode 100644 src/frostfs_testlib/analytics/__init__.py create mode 100644 src/frostfs_testlib/analytics/test_case.py create mode 100644 src/frostfs_testlib/analytics/test_collector.py create mode 100644 src/frostfs_testlib/analytics/test_exporter.py create mode 100644 src/frostfs_testlib/analytics/testrail_exporter.py diff --git a/requirements.txt b/requirements.txt index a75b94f..eee5a85 100644 --- a/requirements.txt +++ b/requirements.txt @@ -5,6 +5,8 @@ neo-mamba==1.0.0 paramiko==2.10.3 pexpect==4.8.0 requests==2.28.1 +docstring_parser==0.15 +testrail-api==1.12.0 # Dev dependencies black==22.8.0 diff --git a/src/frostfs_testlib/analytics/__init__.py b/src/frostfs_testlib/analytics/__init__.py new file mode 100644 index 0000000..1eb8a74 --- /dev/null +++ b/src/frostfs_testlib/analytics/__init__.py @@ -0,0 +1,4 @@ +from test_case import id, suite_name, suite_section, title +from test_collector import TestCase, TestCaseCollector +from testrail_exporter import TestrailExporter +from testrail_exporter import TestrailExporter \ No newline at end of file diff --git a/src/frostfs_testlib/analytics/test_case.py b/src/frostfs_testlib/analytics/test_case.py new file mode 100644 index 0000000..c6e7ff5 --- /dev/null +++ b/src/frostfs_testlib/analytics/test_case.py @@ -0,0 +1,82 @@ +import allure + +from enum import Enum +from types import FunctionType + +class TestCasePriority(Enum): + HIGHEST = 0 + HIGH = 1 + MEDIUM = 2 + LOW = 3 + +def __set_label__(name: str, value: str, allure_decorator: FunctionType = None): + """ + Generic function for do not duplicate set label code in each decorator. + We get decorated function as an object and set needed argument inside. + + Args: + name: argument name to set into the function object + value: argument value to set into the function object + allure_decorator: allure decorator to decorate function and do not duplicate decorators with same value + """ + def wrapper(decorated_func): + if allure_decorator: + decorated_func = allure_decorator(value)(decorated_func) + setattr(decorated_func, name, value) + return decorated_func + + return wrapper + + +def id(uuid: str): + """ + Decorator for set test case ID which can be used as unique value due export into TMS. + + We prefer to use UUID4 format string for ID. + ID have to be generated manually for each new test. + + Args: + uuid: id to set as test_case_id into test function + """ + return __set_label__("__test_case_id__", uuid) + + +def title(title: str): + """ + Decorator for set test case title / name / summary / short description what we do. + + Args: + title: string with title to set into test function + """ + + return __set_label__("__test_case_title__", title, allure.title) + +def priority(priority: str): + """ + Decorator for set test case title / name / summary / short description what we do. + + Args: + priority: string with priority to set into test function + """ + + return __set_label__("__test_case_priority__", priority) + + +def suite_name(name: str): + """ + Decorator for set test case suite name. + Suite name is usually using in TMS for create structure of test cases. + + Args: + name: string with test suite name for set into test function + """ + + return __set_label__("__test_case_suite_name__", name, allure.story) + + +def suite_section(name: str): + """ + Decorator for set test case suite section. + Suite section is usually using in TMS for create deep test cases structure. + """ + return __set_label__("__test_case_suite_section__", name) diff --git a/src/frostfs_testlib/analytics/test_collector.py b/src/frostfs_testlib/analytics/test_collector.py new file mode 100644 index 0000000..0f5398e --- /dev/null +++ b/src/frostfs_testlib/analytics/test_collector.py @@ -0,0 +1,190 @@ +import re + +from docstring_parser import parse +from docstring_parser.common import DocstringStyle +from docstring_parser.google import DEFAULT_SECTIONS, Section, SectionType + +DEFAULT_SECTIONS.append(Section("Steps", "steps", SectionType.MULTIPLE)) + +class TestCase: + """ + Test case object implementation for use in collector and exporters + """ + + def __init__( + self, + uuid_id: str, + title: str, + description: str, + priority: int, + steps: dict, + params: str, + suite_name: str, + suite_section_name: str, + ): + """ + Base constructor for TestCase object + + Args: + uuid_id: uuid from id decorator + title: test case title from title decorator + priority: test case priority value (0-3) + steps: list of test case steps read from function __doc__ + params: string with test case param read from pytest Function(test) object + suite_name: test case suite name from test_suite decorator + suite_section_name: test case suite section from test_suite_section decorator + """ + + # It can confuse, but we rewrite id to "id [params]" string + # We do it in case that one functions can return a lot of tests if we use test params + if params: + self.id = f"{uuid_id} [{params}]" + else: + self.id: str = uuid_id + self.title: str = title + self.description: str = description + self.priority: int = priority + self.steps: dict = steps + self.params: str = params + self.suite_name: str = suite_name + self.suite_section_name: str = suite_section_name + + +class TestCaseCollector: + """ + Collector working like a plugin for pytest and can be used in collect-only call to get tests list from pytest + Additionally, we have several function to filter tests that can be exported. + """ + + pytest_tests = [] + + def __format_string_with_params__(self, source_string: str, test_params: dict) -> str: + """ + Helper function for format test case string arguments using test params. + Params name can be deep like a.b.c, so we will get the value from tests params. + Additionally, we check is the next object dict or real object to use right call for get next argument. + + Args: + source_string: string for format by using test params (if needed) + test_params: dictionary with test params got from pytest test object + Returns: + (str): formatted string with replaced params name by params value + """ + + target_string: str = source_string + for match in re.findall(r"\{(.*?)}", source_string): + nestings_attrs = match.split(".") + param = None + for nesting_attr in nestings_attrs: + if not param: + param = test_params.get(nesting_attr) + else: + if isinstance(param, dict): + param = param.get(nesting_attr) + else: + param = getattr(param, nesting_attr) + target_string = target_string.replace(f"{{{match}}}", str(param)) + return target_string + + def __get_test_case_from_pytest_test__(self, test) -> TestCase: + """ + Parce test meta and return test case if there is enough information for that. + + Args: + test: pytest Function object + Returns: + (TestCase): return tests cases if there is enough information for that and None if not + """ + + # Default values for use behind + suite_name: str = None + suite_section_name: str = None + test_case_steps = dict() + test_case_params: str = "" + test_case_description: str = "" + + # Read test_case suite and section name from test class if possible and get test function from class + if test.cls: + suite_name = test.cls.__dict__.get("__test_case_suite_name__", suite_name) + suite_section_name = test.cls.__dict__.get("__test_case_suite_section__", suite_section_name) + test_function = test.cls.__dict__[test.originalname] + else: + # If no test class, read test function from module + test_function = test.module.__dict__[test.originalname] + + # Read base values from test function arguments + test_case_id = test_function.__dict__.get("__test_case_id__", None) + test_case_title = test_function.__dict__.get("__test_case_title__", None) + test_case_priority = test_function.__dict__.get("__test_case_priority__", None) + suite_name = test_function.__dict__.get("__test_case_suite_name__", suite_name) + suite_section_name = test_function.__dict__.get("__test_case_suite_section__", suite_section_name) + + # Parce test_steps if they define in __doc__ + doc_string = parse(test_function.__doc__, style=DocstringStyle.GOOGLE) + + if doc_string.short_description: + test_case_description = doc_string.short_description + if doc_string.long_description: + test_case_description = f"{doc_string.short_description}\r\n{doc_string.long_description}" + + if doc_string.meta: + for meta in doc_string.meta: + if meta.args[0] == "steps": + test_case_steps[meta.args[1]] = meta.description + + # Read params from tests function if its exist + test_case_call_spec = getattr(test, "callspec", "") + + if test_case_call_spec: + # Set test cases params string value + test_case_params = test_case_call_spec.id + # Format title with params + if test_case_title: + test_case_title = self.__format_string_with_params__(test_case_title,test_case_call_spec.params) + # Format steps with params + if test_case_steps: + for key, value in test_case_steps.items(): + value = self.__format_string_with_params__(value,test_case_call_spec.params) + test_case_steps[key] = value + + # If there is set basic test case attributes create TestCase and return + if test_case_id and test_case_title and suite_name and suite_name: + test_case = TestCase( + id=test_case_id, + title=test_case_title, + description=test_case_description, + priority=test_case_priority, + steps=test_case_steps, + params=test_case_params, + suite_name=suite_name, + suite_section_name=suite_section_name, + ) + return test_case + # Return None if there is no enough information for return test case + return None + + def pytest_report_collectionfinish(self, pytest_tests: list) -> None: + """ + !!! DO NOT CHANGE THE NANE IT IS NOT A MISTAKE + Implement specific function with specific name + Pytest will be call this function when he uses plugin in collect-only call + + Args: + pytest_tests: list of pytest tests + """ + self.pytest_tests.extend(pytest_tests) + + def collect_test_cases(self) -> list[TestCase]: + """ + We're collecting test cases from the pytest tests list and return them in test case representation. + + Returns: + (list[TestCase]): list of test cases that we found in the pytest tests code + """ + test_cases = [] + + for test in self.pytest_tests: + test_case = self.__get_test_case_from_pytest_test__(test) + if test_case: + test_cases.append(test_case) + return test_cases \ No newline at end of file diff --git a/src/frostfs_testlib/analytics/test_exporter.py b/src/frostfs_testlib/analytics/test_exporter.py new file mode 100644 index 0000000..dd57478 --- /dev/null +++ b/src/frostfs_testlib/analytics/test_exporter.py @@ -0,0 +1,71 @@ +from abc import ABC, abstractmethod + +from test_collector import TestCase + +class TestExporter(ABC): + test_cases_cache = [] + test_suites_cache = [] + test_case_id_field_name = "" + + @abstractmethod + def fill_suite_cache(self) -> None: + """ + Fill test_suite_cache by all tests cases in TMS + It's help do not call TMS each time then we search test suite + """ + + @abstractmethod + def fill_cases_cache(self) -> None: + """ + Fill test_cases_cache by all tests cases in TMS + It's help do not call TMS each time then we search test case + """ + + @abstractmethod + def search_test_case_id(self, test_case_id: str) -> object: + """ + Find test cases in TMS by ID + """ + + @abstractmethod + def get_or_create_test_suite(self, test_suite_name) -> object: + """ + Get suite name with exact name or create if not exist + """ + + @abstractmethod + def get_or_create_suite_section(self, test_rail_suite, section_name) -> object: + """ + Get suite section with exact name or create new one if not exist + """ + + @abstractmethod + def create_test_case(self, test_case: TestCase, test_suite, test_suite_section) -> None: + """ + Create test case in TMS + """ + + @abstractmethod + def update_test_case(self, test_case: TestCase, test_case_in_tms, test_suite, test_suite_section) -> None: + """ + Update test case in TMS + """ + + def export_test_cases(self, test_cases: list[TestCase]): + # Fill caches before starting imports + self.fill_suite_cache() + self.fill_cases_cache() + + for test_case in test_cases: + test_suite = self.get_or_create_test_suite(test_case.suite_name) + test_section = self.get_or_create_suite_section(test_suite, test_case.suite_section_name) + test_case_in_tms = self.search_test_case_id(test_case.id) + steps = [ + {"content": value, "expected": " "} + for key, value in test_case.steps.items() + ] + + if test_case: + self.update_test_case(test_case, test_case_in_tms) + else: + self.create_test_case(test_case) \ No newline at end of file diff --git a/src/frostfs_testlib/analytics/testrail_exporter.py b/src/frostfs_testlib/analytics/testrail_exporter.py new file mode 100644 index 0000000..cefbf2e --- /dev/null +++ b/src/frostfs_testlib/analytics/testrail_exporter.py @@ -0,0 +1,178 @@ +from testrail_api import TestRailAPI + +from test_collector import TestCase +from test_exporter import TestExporter + + +class TestrailExporter(TestExporter): + def __init__( + self, + tr_url: str, + tr_username: str, + tr_password: str, + tr_project_id: int, + tr_template_id_without_steps: int, + tr_template_id_with_steps: int, + tr_priority_map: dict, + tr_id_field: str, + tr_description_fields: str, + tr_steps_field: str, + ): + """ + Redefine init for base exporter for get test rail credentials and project on create exporter + + Args: + tr_url: api url for create TestRailAPI object. See lib docs for details + tr_username: Testrail user login for api authentication + tr_password: Testrail user password for api authentication + tr_template_id_with_steps: id of test case template with steps + tr_template_id_without_steps: id of test case template without steps + tr_priority_map: mapping of TestCasePriority to priority ids in Testrail + """ + + self.api: TestRailAPI = TestRailAPI(tr_url, tr_username, tr_password) + self.tr_project_id: int = tr_project_id + self.tr_template_id_without_steps = tr_template_id_without_steps + self.tr_template_id_with_steps = tr_template_id_with_steps + self.tr_priority_map = tr_priority_map + self.tr_id_field = tr_id_field + self.tr_description_fields = tr_description_fields + self.tr_steps_field = tr_steps_field + + def fill_suite_cache(self) -> None: + """ + Fill test_suite_cache by all tests cases in TestRail + It's help do not call TMS each time then we search test suite + """ + project_suites = self.api.suites.get_suites(project_id=self.tr_project_id) + + for test_suite in project_suites: + test_suite_sections = self.api.sections.get_sections( + project_id=self.tr_project_id, + suite_id=test_suite["id"], + ) + test_suite["sections"] = test_suite_sections + + self.test_suites_cache.append(test_suite) + + def fill_cases_cache(self) -> None: + """ + Fill test_cases_cache by all tests cases in TestRail + It's help do not call TMS each time then we search test case + """ + for test_suite in self.test_suites_cache: + self.test_cases_cache.extend( + self.api.cases.get_cases(self.tr_project_id, suite_id=test_suite["id"]) + ) + + def search_test_case_id(self, test_case_id: str) -> object: + """ + Find test cases in TestRail (cache) by ID + """ + test_cases = [ + test_case + for test_case in self.test_cases_cache + if test_case["custom_autotest_name"] == test_case_id + ] + + if len(test_cases) > 1: + raise RuntimeError(f"Too many results found in test rail for id {test_case_id}") + elif len(test_cases) == 1: + return test_cases.pop() + else: + return None + + def get_or_create_test_suite(self, test_suite_name) -> object: + """ + Get suite name with exact name from Testrail or create if not exist + """ + test_rail_suites = [ + suite for suite in self.test_suites_cache if suite["name"] == test_suite_name + ] + + if not test_rail_suites: + test_rail_suite = self.api.suites.add_suite( + project_id=self.tr_project_id, + name=test_suite_name, + ) + test_rail_suite["sections"] = list() + self.test_suites_cache.append(test_rail_suite) + return test_rail_suite + elif len(test_rail_suites) == 1: + return test_rail_suites.pop() + else: + raise RuntimeError(f"Too many results found in test rail for suite name {test_suite_name}") + + def get_or_create_suite_section(self, test_rail_suite, section_name) -> object: + """ + Get suite section with exact name from Testrail or create new one if not exist + """ + test_rail_sections = [ + section for section in test_rail_suite["sections"] if section["name"] == section_name + ] + + if not test_rail_sections: + test_rail_section = self.api.sections.add_section( + project_id=self.tr_project_id, + suite_id=test_rail_suite["id"], + name=section_name, + ) + # !!!!!! BAD !!!!!! Do we really change object from cache or copy of suite object???? + # !!!!!! WE have to update object in cache + # !!!!! In opposite we will try to create section twice and get error from API + test_rail_suite["sections"].append(test_rail_section) + return test_rail_section + elif len(test_rail_sections) == 1: + return test_rail_sections.pop() + else: + raise RuntimeError( + f"Too many results found in test rail for section name {section_name}" + ) + + def prepare_request_body(self, test_case: TestCase, test_suite, test_suite_section) -> dict: + """ + Helper to prepare request body for add or update tests case from TestCase object + """ + request_body = { + "title": test_case.title, + "section_id": test_suite_section["id"], + self.test_case_id_field_name: test_case.id, + + } + + if test_case.priority: + request_body["priority_id"] = self.tr_priority_map.get(test_case.priority) + + if test_case.steps: + steps = [ + {"content": value, "expected": " "} + for key, value in test_case.steps.items() + ] + request_body[self.tr_steps_field] = steps + request_body["template_id"]=self.tr_template_id_with_steps + else: + request_body["template_id"] = self.tr_template_id_without_steps + if test_case.description: + request_body[self.tr_description_fields] = self.tr_description_fields + + return request_body + + + def create_test_case(self, test_case: TestCase, test_suite, test_suite_section) -> None: + """ + Create test case in Testrail + """ + request_body = self.prepare_request_body(test_case, test_suite, test_suite_section) + + self.api.cases.add_case(**request_body) + + + def update_test_case(self, test_case: TestCase, test_case_in_tms, test_suite, test_suite_section) -> None: + """ + Update test case in Testrail + """ + request_body = self.prepare_request_body(test_case, test_suite, test_suite_section) + + self.api.cases.update_case(case_id=test_case_in_tms["id"], **request_body) + +