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

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

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)

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

src/frostfs_testlib/analytics/test_exporter.py

@@ -0,0 +1,70 @@
+from abc import ABC, abstractmethod
+
+from test_collector import TestCase
+
+class TestExporter(ABC):
+    test_cases_cache = []
+    test_suites_cache = []
+
+    @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: str) -> 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: str) -> 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)

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