16521578080

Committed 25 Jul 2025 12:08PM UTC coverage: 36.444% (+0.07%) from 36.372%

Build # 16521578080

Build Type

push

github

Committed by

web-flow

Commit Message

feat: TMS-34041: support the api-client without pydantic. (#196)

Co-authored-by: pavel.butuzov <pavel.butuzov@testit.software>

Run Details

11 of 39 new or added lines in 2 files covered. (28.21%)

2 existing lines in 1 file now uncovered.

1191 of 3268 relevant lines covered (36.44%)

0.73 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

30.04

/testit-python-commons/src/testit_python_commons/client/api_client.py

import logging
import os
import typing
from datetime import datetime

from testit_api_client import ApiClient, Configuration
from testit_api_client.apis import AttachmentsApi, AutoTestsApi, TestRunsApi, TestResultsApi, WorkItemsApi
from testit_api_client.models import (
    AutoTestApiResult,
    AutoTestPostModel,
    AutoTestPutModel,
    AttachmentPutModel,
    AutoTestResultsForTestRunModel,
    TestResultResponse,
    WorkItemIdModel,
    WorkItemIdentifierModel
)

from testit_python_commons.client.client_configuration import ClientConfiguration
from testit_python_commons.client.converter import Converter
from testit_python_commons.models.test_result import TestResult
from testit_python_commons.services.logger import adapter_logger
from testit_python_commons.services.retry import retry
from testit_python_commons.utils.html_escape_utils import HtmlEscapeUtils


class ApiClientWorker:
    __max_tests_for_write = 100

    def __init__(self, config: ClientConfiguration):
        api_client_config = self.__get_api_client_configuration(
            url=config.get_url(),
            verify_ssl=config.get_cert_validation() != 'false',
            proxy=config.get_proxy())
        api_client = self.__get_api_client(api_client_config, config.get_private_token())

        self.__test_run_api = TestRunsApi(api_client=api_client)
        self.__autotest_api = AutoTestsApi(api_client=api_client)
        self.__attachments_api = AttachmentsApi(api_client=api_client)
        self.__test_results_api = TestResultsApi(api_client=api_client)
        self.__work_items_api = WorkItemsApi(api_client=api_client)
        self.__config = config

    @staticmethod
    @adapter_logger
    def __get_api_client_configuration(url: str, verify_ssl: bool = True, proxy: str = None) -> Configuration:
        api_client_configuration = Configuration(host=url)
        api_client_configuration.verify_ssl = verify_ssl
        api_client_configuration.proxy = proxy

        return api_client_configuration

    @staticmethod
    @adapter_logger
    def __get_api_client(api_client_config: Configuration, token: str) -> ApiClient:
        return ApiClient(
            configuration=api_client_config,
            header_name='Authorization',
            header_value='PrivateToken ' + token)

    @staticmethod
    def _escape_html_in_model(model):
        """Apply HTML escaping to all models before sending to API"""
        return HtmlEscapeUtils.escape_html_in_object(model)

    @adapter_logger
    def create_test_run(self):
        test_run_name = f'TestRun_{datetime.today().strftime("%Y-%m-%dT%H:%M:%S")}' if \
            not self.__config.get_test_run_name() else self.__config.get_test_run_name()
        model = Converter.test_run_to_test_run_short_model(
            self.__config.get_project_id(),
            test_run_name
        )
        model = self._escape_html_in_model(model)

        response = self.__test_run_api.create_empty(create_empty_request=model)

        return Converter.get_id_from_create_test_run_response(response)

    @adapter_logger
    def set_test_run_id(self, test_run_id: str):
        self.__config.set_test_run_id(test_run_id)

    @adapter_logger
    def get_autotests_by_test_run_id(self):
        response = self.__test_run_api.get_test_run_by_id(self.__config.get_test_run_id())

        return Converter.get_resolved_autotests_from_get_test_run_response(
            response,
            self.__config.get_configuration_id())

    @adapter_logger
    def write_test(self, test_result: TestResult):
        model = Converter.project_id_and_external_id_to_auto_tests_search_post_request(
            self.__config.get_project_id(),
            test_result.get_external_id())

        autotests = self.__autotest_api.api_v2_auto_tests_search_post(api_v2_auto_tests_search_post_request=model)

        if autotests:
            self.__update_test(test_result, autotests[0])
        else:
            self.__create_test(test_result)

        return self.__load_test_result(test_result)

    @adapter_logger
    def write_tests(self, test_results: typing.List[TestResult], fixture_containers: dict):
        autotests_for_create = []
        autotests_for_update = []
        results_for_autotests_being_created = []
        results_for_autotests_being_updated = []

        for test_result in test_results:
            test_result = self.__add_fixtures_to_test_result(test_result, fixture_containers)

            model = Converter.project_id_and_external_id_to_auto_tests_search_post_request(
                self.__config.get_project_id(),
                test_result.get_external_id())

            test_result_model = Converter.test_result_to_testrun_result_post_model(
                test_result,
                self.__config.get_configuration_id())

            autotests = self.__autotest_api.api_v2_auto_tests_search_post(api_v2_auto_tests_search_post_request=model)

            if autotests:
                autotest_for_update = self.__prepare_to_mass_update_autotest(test_result, autotests[0])

                autotests_for_update.append(autotest_for_update)
                results_for_autotests_being_updated.append(test_result_model)

                if len(autotests_for_update) >= self.__max_tests_for_write:
                    self.__update_tests(autotests_for_update)
                    self.__load_test_results(results_for_autotests_being_updated)

                    autotests_for_update.clear()
                    results_for_autotests_being_updated.clear()
            else:
                autotest_for_create = self.__prepare_to_mass_create_autotest(test_result)

                autotests_for_create.append(autotest_for_create)
                results_for_autotests_being_created.append(test_result_model)

                if len(autotests_for_create) >= self.__max_tests_for_write:
                    self.__create_tests(autotests_for_create)
                    self.__load_test_results(results_for_autotests_being_created)

                    autotests_for_create.clear()
                    results_for_autotests_being_created.clear()

        if len(autotests_for_update) > 0:
            self.__update_tests(autotests_for_update)
            self.__load_test_results(results_for_autotests_being_updated)

            autotests_for_update.clear()
            results_for_autotests_being_updated.clear()

        if len(autotests_for_create) > 0:
            self.__create_tests(autotests_for_create)
            self.__load_test_results(results_for_autotests_being_created)

            autotests_for_create.clear()
            results_for_autotests_being_created.clear()

    @adapter_logger
    def __prepare_to_create_autotest(self, test_result: TestResult) -> AutoTestPostModel:
        logging.debug('Preparing to create the auto test ' + test_result.get_external_id())

        model = Converter.test_result_to_create_autotest_request(
            test_result,
            self.__config.get_project_id())
        model.work_item_ids_for_link_with_auto_test = self.__get_work_item_uuids_for_link_with_auto_test(
            test_result.get_work_item_ids())

        return model

    @adapter_logger
    def __prepare_to_mass_create_autotest(self, test_result: TestResult) -> AutoTestPostModel:
        logging.debug('Preparing to create the auto test ' + test_result.get_external_id())

        model = Converter.test_result_to_autotest_post_model(
            test_result,
            self.__config.get_project_id())
        model.work_item_ids_for_link_with_auto_test = self.__get_work_item_uuids_for_link_with_auto_test(
            test_result.get_work_item_ids())

        return model

    @adapter_logger
    def __prepare_to_update_autotest(
            self,
            test_result: TestResult,
            autotest: AutoTestApiResult) -> AutoTestPutModel:
        logging.debug('Preparing to update the auto test ' + test_result.get_external_id())

        model = Converter.test_result_to_update_autotest_request(
            test_result,
            self.__config.get_project_id())
        model.is_flaky = autotest.is_flaky
        model.work_item_ids_for_link_with_auto_test = self.__get_work_item_uuids_for_link_with_auto_test(
            test_result.get_work_item_ids(),
            str(autotest.global_id))

        return model

    @adapter_logger
    def __prepare_to_mass_update_autotest(
            self,
            test_result: TestResult,
            autotest: AutoTestApiResult) -> AutoTestPutModel:
        logging.debug('Preparing to update the auto test ' + test_result.get_external_id())

        model = Converter.test_result_to_autotest_put_model(
            test_result,
            self.__config.get_project_id())
        model.is_flaky = autotest.is_flaky
        model.work_item_ids_for_link_with_auto_test = self.__get_work_item_uuids_for_link_with_auto_test(
            test_result.get_work_item_ids(),
            str(autotest.global_id))

        return model

    @staticmethod
    @adapter_logger
    def __add_fixtures_to_test_result(
            test_result: TestResult,
            fixtures_containers: dict) -> TestResult:
        setup_results = []
        teardown_results = []

        for uuid, fixtures_container in fixtures_containers.items():
            if test_result.get_external_id() in fixtures_container.external_ids:
                if fixtures_container.befores:
                    setup_results += fixtures_container.befores[0].steps

                if fixtures_container.afters:
                    teardown_results = fixtures_container.afters[0].steps + teardown_results

        test_result.set_setup_results(setup_results)
        test_result.set_teardown_results(teardown_results)

        return test_result

    @adapter_logger
    def __get_work_item_uuids_for_link_with_auto_test(
            self,
            work_item_ids: list,
            autotest_global_id: str = None) -> list:
        linked_work_items = []

        if autotest_global_id:
            linked_work_items = self.__get_work_items_linked_to_autotest(autotest_global_id)

        work_item_uuids = self.__prepare_list_of_work_item_uuids(linked_work_items, work_item_ids)

        return work_item_uuids

    @adapter_logger
    def __prepare_list_of_work_item_uuids(
            self,
            linked_work_items: list,
            work_item_ids: list) -> list:
        work_item_uuids = []

        for linked_work_item in linked_work_items:
            linked_work_item_id = str(linked_work_item.global_id)
            linked_work_item_uuid = linked_work_item.id

            if linked_work_item_id in work_item_ids:
                work_item_ids.remove(linked_work_item_id)
                work_item_uuids.append(linked_work_item_uuid)

                continue

            if self.__config.get_automatic_updation_links_to_test_cases() != 'true':
                work_item_uuids.append(linked_work_item_uuid)

        for work_item_id in work_item_ids:
            work_item_uuid = self.__get_work_item_uuid_by_work_item_id(work_item_id)

            if work_item_uuid:
                work_item_uuids.append(work_item_uuid)

        return work_item_uuids

    @adapter_logger
    def __get_work_item_uuid_by_work_item_id(self, work_item_id: str) -> str or None:
        logging.debug('Getting workitem by id ' + work_item_id)

        try:
            work_item = self.__work_items_api.get_work_item_by_id(id=work_item_id)

            logging.debug(f'Got workitem {work_item}')

            return work_item.id
        except Exception as exc:
            logging.error(f'Getting workitem by id {work_item_id} status: {exc}')

    @adapter_logger
    def __get_work_items_linked_to_autotest(self, autotest_global_id: str) -> typing.List[WorkItemIdentifierModel]:
        return self.__autotest_api.get_work_items_linked_to_auto_test(id=autotest_global_id)

    @adapter_logger
    def __update_autotest_link_from_work_items(self, autotest_global_id: str, work_item_ids: list):
        linked_work_items = self.__get_work_items_linked_to_autotest(autotest_global_id)

        for linked_work_item in linked_work_items:
            linked_work_item_id = str(linked_work_item.global_id)

            if linked_work_item_id in work_item_ids:
                work_item_ids.remove(linked_work_item_id)

                continue

            if self.__config.get_automatic_updation_links_to_test_cases() != 'false':
                self.__unlink_test_to_work_item(autotest_global_id, linked_work_item_id)

        for work_item_id in work_item_ids:
            self.__link_test_to_work_item(autotest_global_id, work_item_id)

    @adapter_logger
    def __create_test(self, test_result: TestResult) -> str:
        logging.debug(f'Autotest "{test_result.get_autotest_name()}" was not found')

        model = self.__prepare_to_create_autotest(test_result)
        model = self._escape_html_in_model(model)

        autotest_response = self.__autotest_api.create_auto_test(create_auto_test_request=model)

        logging.debug(f'Autotest "{test_result.get_autotest_name()}" was created')

        return autotest_response.id

    @adapter_logger
    def __create_tests(self, autotests_for_create: typing.List[AutoTestPostModel]):
        logging.debug(f'Creating autotests: "{autotests_for_create}')

        autotests_for_create = self._escape_html_in_model(autotests_for_create)
        self.__autotest_api.create_multiple(auto_test_post_model=autotests_for_create)

        logging.debug(f'Autotests were created')

    @adapter_logger
    def __update_test(self, test_result: TestResult, autotest: AutoTestApiResult):
        logging.debug(f'Autotest "{test_result.get_autotest_name()}" was found')

        model = self.__prepare_to_update_autotest(test_result, autotest)
        model = self._escape_html_in_model(model)

        self.__autotest_api.update_auto_test(update_auto_test_request=model)

        logging.debug(f'Autotest "{test_result.get_autotest_name()}" was updated')

    @adapter_logger
    def __update_tests(self, autotests_for_update: typing.List[AutoTestPutModel]):
        logging.debug(f'Updating autotests: {autotests_for_update}')

        autotests_for_update = self._escape_html_in_model(autotests_for_update)
        self.__autotest_api.update_multiple(auto_test_put_model=autotests_for_update)

        logging.debug(f'Autotests were updated')

    @adapter_logger
    @retry
    def __unlink_test_to_work_item(self, autotest_global_id: str, work_item_id: str):
        self.__autotest_api.delete_auto_test_link_from_work_item(
            id=autotest_global_id,
            work_item_id=work_item_id)

        logging.debug(f'Autotest was unlinked with workItem "{work_item_id}" by global id "{autotest_global_id}')

    @adapter_logger
    @retry
    def __link_test_to_work_item(self, autotest_global_id: str, work_item_id: str):
        self.__autotest_api.link_auto_test_to_work_item(
            autotest_global_id,
            work_item_id_model=WorkItemIdModel(id=work_item_id))

        logging.debug(f'Autotest was linked with workItem "{work_item_id}" by global id "{autotest_global_id}')

    @adapter_logger
    def __load_test_result(self, test_result: TestResult) -> str:
        model = Converter.test_result_to_testrun_result_post_model(
            test_result,
            self.__config.get_configuration_id())
        model = self._escape_html_in_model(model)

        response = self.__test_run_api.set_auto_test_results_for_test_run(
            id=self.__config.get_test_run_id(),
            auto_test_results_for_test_run_model=[model])

        logging.debug(f'Result of the autotest "{test_result.get_autotest_name()}" was set '
                      f'in the test run "{self.__config.get_test_run_id()}"')

        return Converter.get_test_result_id_from_testrun_result_post_response(response)

    @adapter_logger
    def __load_test_results(self, test_results: typing.List[AutoTestResultsForTestRunModel]):
        logging.debug(f'Loading test results: {test_results}')

        test_results = self._escape_html_in_model(test_results)
        self.__test_run_api.set_auto_test_results_for_test_run(
            id=self.__config.get_test_run_id(),
            auto_test_results_for_test_run_model=test_results)

    @adapter_logger
    def get_test_result_by_id(self, test_result_id: str) -> TestResultResponse:
        return self.__test_results_api.api_v2_test_results_id_get(id=test_result_id)

    @adapter_logger
    def update_test_results(self, fixtures_containers: dict, test_result_ids: dict):
        test_results = Converter.fixtures_containers_to_test_results_with_all_fixture_step_results(
            fixtures_containers, test_result_ids)

        for test_result in test_results:
            model = Converter.convert_test_result_model_to_test_results_id_put_request(
                self.get_test_result_by_id(test_result.get_test_result_id()))

            model.setup_results = Converter.step_results_to_auto_test_step_result_update_request(
                    test_result.get_setup_results())
            model.teardown_results = Converter.step_results_to_auto_test_step_result_update_request(
                    test_result.get_teardown_results())
            
            model = self._escape_html_in_model(model)

            try:
                self.__test_results_api.api_v2_test_results_id_put(
                    id=test_result.get_test_result_id(),
                    api_v2_test_results_id_put_request=model)
            except Exception as exc:
                logging.error(f'Cannot update test result with id "{test_result.get_test_result_id()}" status: {exc}')

    @adapter_logger
    def load_attachments(self, attach_paths: list or tuple):
        attachments = []

        for path in attach_paths:
            if os.path.isfile(path):
                try:
                    attachment_response = self.__attachments_api.api_v2_attachments_post(file=open(path, "rb"))

                    attachments.append(AttachmentPutModel(attachment_response['id']))

                    logging.debug(f'Attachment "{path}" was uploaded')
                except Exception as exc:
                    logging.error(f'Upload attachment "{path}" status: {exc}')
            else:
                logging.error(f'File "{path}" was not found!')
        return attachments

1	import logging	2✔
2	import os	2✔
3	import typing	2✔
4	from datetime import datetime	2✔
5
6	from testit_api_client import ApiClient, Configuration	2✔
7	from testit_api_client.apis import AttachmentsApi, AutoTestsApi, TestRunsApi, TestResultsApi, WorkItemsApi	2✔
8	from testit_api_client.models import (	2✔
9	AutoTestApiResult,
10	AutoTestPostModel,
11	AutoTestPutModel,
12	AttachmentPutModel,
13	AutoTestResultsForTestRunModel,
14	TestResultResponse,
15	WorkItemIdModel,
16	WorkItemIdentifierModel
17	)
18
19	from testit_python_commons.client.client_configuration import ClientConfiguration	2✔
20	from testit_python_commons.client.converter import Converter	2✔
21	from testit_python_commons.models.test_result import TestResult	2✔
22	from testit_python_commons.services.logger import adapter_logger	2✔
23	from testit_python_commons.services.retry import retry	2✔
24	from testit_python_commons.utils.html_escape_utils import HtmlEscapeUtils	2✔
25
26
27	class ApiClientWorker:	2✔
28	__max_tests_for_write = 100	2✔
29
30	def __init__(self, config: ClientConfiguration):	2✔
31	api_client_config = self.__get_api_client_configuration(	×
32	url=config.get_url(),
33	verify_ssl=config.get_cert_validation() != 'false',
34	proxy=config.get_proxy())
35	api_client = self.__get_api_client(api_client_config, config.get_private_token())	×
36
37	self.__test_run_api = TestRunsApi(api_client=api_client)	×
38	self.__autotest_api = AutoTestsApi(api_client=api_client)	×
39	self.__attachments_api = AttachmentsApi(api_client=api_client)	×
40	self.__test_results_api = TestResultsApi(api_client=api_client)	×
41	self.__work_items_api = WorkItemsApi(api_client=api_client)	×
42	self.__config = config	×
43
44	@staticmethod	2✔
45	@adapter_logger	2✔
46	def __get_api_client_configuration(url: str, verify_ssl: bool = True, proxy: str = None) -> Configuration:	2✔
47	api_client_configuration = Configuration(host=url)	×
48	api_client_configuration.verify_ssl = verify_ssl	×
49	api_client_configuration.proxy = proxy	×
50
51	return api_client_configuration	×
52
53	@staticmethod	2✔
54	@adapter_logger	2✔
55	def __get_api_client(api_client_config: Configuration, token: str) -> ApiClient:	2✔
56	return ApiClient(	×
57	configuration=api_client_config,
58	header_name='Authorization',
59	header_value='PrivateToken ' + token)
60
61	@staticmethod	2✔
62	def _escape_html_in_model(model):	2✔
63	"""Apply HTML escaping to all models before sending to API"""
64	return HtmlEscapeUtils.escape_html_in_object(model)	×
65
66	@adapter_logger	2✔
67	def create_test_run(self):	2✔
68	test_run_name = f'TestRun_{datetime.today().strftime("%Y-%m-%dT%H:%M:%S")}' if \	×
69	not self.__config.get_test_run_name() else self.__config.get_test_run_name()
70	model = Converter.test_run_to_test_run_short_model(	×
71	self.__config.get_project_id(),
72	test_run_name
73	)
74	model = self._escape_html_in_model(model)	×
75
NEW 76	response = self.__test_run_api.create_empty(create_empty_request=model)	×
77
78	return Converter.get_id_from_create_test_run_response(response)	×
79
80	@adapter_logger	2✔
81	def set_test_run_id(self, test_run_id: str):	2✔
82	self.__config.set_test_run_id(test_run_id)	×
83
84	@adapter_logger	2✔
85	def get_autotests_by_test_run_id(self):	2✔
86	response = self.__test_run_api.get_test_run_by_id(self.__config.get_test_run_id())	×
87
88	return Converter.get_resolved_autotests_from_get_test_run_response(	×
89	response,
90	self.__config.get_configuration_id())
91
92	@adapter_logger	2✔
93	def write_test(self, test_result: TestResult):	2✔
94	model = Converter.project_id_and_external_id_to_auto_tests_search_post_request(	×
95	self.__config.get_project_id(),
96	test_result.get_external_id())
97
NEW 98	autotests = self.__autotest_api.api_v2_auto_tests_search_post(api_v2_auto_tests_search_post_request=model)	×
99
100	if autotests:	×
101	self.__update_test(test_result, autotests[0])	×
102	else:
103	self.__create_test(test_result)	×
104
105	return self.__load_test_result(test_result)	×
106
107	@adapter_logger	2✔
108	def write_tests(self, test_results: typing.List[TestResult], fixture_containers: dict):	2✔
109	autotests_for_create = []	×
110	autotests_for_update = []	×
111	results_for_autotests_being_created = []	×
112	results_for_autotests_being_updated = []	×
113
114	for test_result in test_results:	×
115	test_result = self.__add_fixtures_to_test_result(test_result, fixture_containers)	×
116
117	model = Converter.project_id_and_external_id_to_auto_tests_search_post_request(	×
118	self.__config.get_project_id(),
119	test_result.get_external_id())
120
121	test_result_model = Converter.test_result_to_testrun_result_post_model(	×
122	test_result,
123	self.__config.get_configuration_id())
124
NEW 125	autotests = self.__autotest_api.api_v2_auto_tests_search_post(api_v2_auto_tests_search_post_request=model)	×
126
127	if autotests:	×
NEW 128	autotest_for_update = self.__prepare_to_mass_update_autotest(test_result, autotests[0])	×
129
130	autotests_for_update.append(autotest_for_update)	×
131	results_for_autotests_being_updated.append(test_result_model)	×
132
133	if len(autotests_for_update) >= self.__max_tests_for_write:	×
134	self.__update_tests(autotests_for_update)	×
135	self.__load_test_results(results_for_autotests_being_updated)	×
136
137	autotests_for_update.clear()	×
138	results_for_autotests_being_updated.clear()	×
139	else:
NEW 140	autotest_for_create = self.__prepare_to_mass_create_autotest(test_result)	×
141
142	autotests_for_create.append(autotest_for_create)	×
143	results_for_autotests_being_created.append(test_result_model)	×
144
145	if len(autotests_for_create) >= self.__max_tests_for_write:	×
146	self.__create_tests(autotests_for_create)	×
147	self.__load_test_results(results_for_autotests_being_created)	×
148
149	autotests_for_create.clear()	×
150	results_for_autotests_being_created.clear()	×
151
152	if len(autotests_for_update) > 0:	×
153	self.__update_tests(autotests_for_update)	×
154	self.__load_test_results(results_for_autotests_being_updated)	×
155
156	autotests_for_update.clear()	×
157	results_for_autotests_being_updated.clear()	×
158
159	if len(autotests_for_create) > 0:	×
160	self.__create_tests(autotests_for_create)	×
161	self.__load_test_results(results_for_autotests_being_created)	×
162
163	autotests_for_create.clear()	×
164	results_for_autotests_being_created.clear()	×
165
166	@adapter_logger	2✔
167	def __prepare_to_create_autotest(self, test_result: TestResult) -> AutoTestPostModel:	2✔
168	logging.debug('Preparing to create the auto test ' + test_result.get_external_id())	×
169
NEW 170	model = Converter.test_result_to_create_autotest_request(	×
171	test_result,
172	self.__config.get_project_id())
NEW 173	model.work_item_ids_for_link_with_auto_test = self.__get_work_item_uuids_for_link_with_auto_test(	×
174	test_result.get_work_item_ids())
175
NEW 176	return model	×
177
178	@adapter_logger	2✔
179	def __prepare_to_mass_create_autotest(self, test_result: TestResult) -> AutoTestPostModel:	2✔
NEW 180	logging.debug('Preparing to create the auto test ' + test_result.get_external_id())	×
181
UNCOV 182	model = Converter.test_result_to_autotest_post_model(	×
183	test_result,
184	self.__config.get_project_id())
185	model.work_item_ids_for_link_with_auto_test = self.__get_work_item_uuids_for_link_with_auto_test(	×
186	test_result.get_work_item_ids())
187
188	return model	×
189
190	@adapter_logger	2✔
191	def __prepare_to_update_autotest(	2✔
192	self,
193	test_result: TestResult,
194	autotest: AutoTestApiResult) -> AutoTestPutModel:
195	logging.debug('Preparing to update the auto test ' + test_result.get_external_id())	×
196
NEW 197	model = Converter.test_result_to_update_autotest_request(	×
198	test_result,
199	self.__config.get_project_id())
NEW 200	model.is_flaky = autotest.is_flaky	×
NEW 201	model.work_item_ids_for_link_with_auto_test = self.__get_work_item_uuids_for_link_with_auto_test(	×
202	test_result.get_work_item_ids(),
203	str(autotest.global_id))
204
NEW 205	return model	×
206
207	@adapter_logger	2✔
208	def __prepare_to_mass_update_autotest(	2✔
209	self,
210	test_result: TestResult,
211	autotest: AutoTestApiResult) -> AutoTestPutModel:
NEW 212	logging.debug('Preparing to update the auto test ' + test_result.get_external_id())	×
213
UNCOV 214	model = Converter.test_result_to_autotest_put_model(	×
215	test_result,
216	self.__config.get_project_id())
217	model.is_flaky = autotest.is_flaky	×
218	model.work_item_ids_for_link_with_auto_test = self.__get_work_item_uuids_for_link_with_auto_test(	×
219	test_result.get_work_item_ids(),
220	str(autotest.global_id))
221
222	return model	×
223
224	@staticmethod	2✔
225	@adapter_logger	2✔
226	def __add_fixtures_to_test_result(	2✔
227	test_result: TestResult,
228	fixtures_containers: dict) -> TestResult:
229	setup_results = []	×
230	teardown_results = []	×
231
232	for uuid, fixtures_container in fixtures_containers.items():	×
233	if test_result.get_external_id() in fixtures_container.external_ids:	×
234	if fixtures_container.befores:	×
235	setup_results += fixtures_container.befores[0].steps	×
236
237	if fixtures_container.afters:	×
238	teardown_results = fixtures_container.afters[0].steps + teardown_results	×
239
240	test_result.set_setup_results(setup_results)	×
241	test_result.set_teardown_results(teardown_results)	×
242
243	return test_result	×
244
245	@adapter_logger	2✔
246	def __get_work_item_uuids_for_link_with_auto_test(	2✔
247	self,
248	work_item_ids: list,
249	autotest_global_id: str = None) -> list:
250	linked_work_items = []	×
251
252	if autotest_global_id:	×
253	linked_work_items = self.__get_work_items_linked_to_autotest(autotest_global_id)	×
254
255	work_item_uuids = self.__prepare_list_of_work_item_uuids(linked_work_items, work_item_ids)	×
256
257	return work_item_uuids	×
258
259	@adapter_logger	2✔
260	def __prepare_list_of_work_item_uuids(	2✔
261	self,
262	linked_work_items: list,
263	work_item_ids: list) -> list:
264	work_item_uuids = []	×
265
266	for linked_work_item in linked_work_items:	×
267	linked_work_item_id = str(linked_work_item.global_id)	×
268	linked_work_item_uuid = linked_work_item.id	×
269
270	if linked_work_item_id in work_item_ids:	×
271	work_item_ids.remove(linked_work_item_id)	×
272	work_item_uuids.append(linked_work_item_uuid)	×
273
274	continue	×
275
276	if self.__config.get_automatic_updation_links_to_test_cases() != 'true':	×
277	work_item_uuids.append(linked_work_item_uuid)	×
278
279	for work_item_id in work_item_ids:	×
280	work_item_uuid = self.__get_work_item_uuid_by_work_item_id(work_item_id)	×
281
282	if work_item_uuid:	×
283	work_item_uuids.append(work_item_uuid)	×
284
285	return work_item_uuids	×
286
287	@adapter_logger	2✔
288	def __get_work_item_uuid_by_work_item_id(self, work_item_id: str) -> str or None:	2✔
289	logging.debug('Getting workitem by id ' + work_item_id)	×
290
291	try:	×
292	work_item = self.__work_items_api.get_work_item_by_id(id=work_item_id)	×
293
294	logging.debug(f'Got workitem {work_item}')	×
295
296	return work_item.id	×
297	except Exception as exc:	×
298	logging.error(f'Getting workitem by id {work_item_id} status: {exc}')	×
299
300	@adapter_logger	2✔
301	def __get_work_items_linked_to_autotest(self, autotest_global_id: str) -> typing.List[WorkItemIdentifierModel]:	2✔
302	return self.__autotest_api.get_work_items_linked_to_auto_test(id=autotest_global_id)	×
303
304	@adapter_logger	2✔
305	def __update_autotest_link_from_work_items(self, autotest_global_id: str, work_item_ids: list):	2✔
306	linked_work_items = self.__get_work_items_linked_to_autotest(autotest_global_id)	×
307
308	for linked_work_item in linked_work_items:	×
309	linked_work_item_id = str(linked_work_item.global_id)	×
310
311	if linked_work_item_id in work_item_ids:	×
312	work_item_ids.remove(linked_work_item_id)	×
313
314	continue	×
315
316	if self.__config.get_automatic_updation_links_to_test_cases() != 'false':	×
317	self.__unlink_test_to_work_item(autotest_global_id, linked_work_item_id)	×
318
319	for work_item_id in work_item_ids:	×
320	self.__link_test_to_work_item(autotest_global_id, work_item_id)	×
321
322	@adapter_logger	2✔
323	def __create_test(self, test_result: TestResult) -> str:	2✔
324	logging.debug(f'Autotest "{test_result.get_autotest_name()}" was not found')	×
325
326	model = self.__prepare_to_create_autotest(test_result)	×
327	model = self._escape_html_in_model(model)	×
328
NEW 329	autotest_response = self.__autotest_api.create_auto_test(create_auto_test_request=model)	×
330
331	logging.debug(f'Autotest "{test_result.get_autotest_name()}" was created')	×
332
333	return autotest_response.id	×
334
335	@adapter_logger	2✔
336	def __create_tests(self, autotests_for_create: typing.List[AutoTestPostModel]):	2✔
337	logging.debug(f'Creating autotests: "{autotests_for_create}')	×
338
339	autotests_for_create = self._escape_html_in_model(autotests_for_create)	×
340	self.__autotest_api.create_multiple(auto_test_post_model=autotests_for_create)	×
341
342	logging.debug(f'Autotests were created')	×
343
344	@adapter_logger	2✔
345	def __update_test(self, test_result: TestResult, autotest: AutoTestApiResult):	2✔
346	logging.debug(f'Autotest "{test_result.get_autotest_name()}" was found')	×
347
348	model = self.__prepare_to_update_autotest(test_result, autotest)	×
349	model = self._escape_html_in_model(model)	×
350
NEW 351	self.__autotest_api.update_auto_test(update_auto_test_request=model)	×
352
353	logging.debug(f'Autotest "{test_result.get_autotest_name()}" was updated')	×
354
355	@adapter_logger	2✔
356	def __update_tests(self, autotests_for_update: typing.List[AutoTestPutModel]):	2✔
357	logging.debug(f'Updating autotests: {autotests_for_update}')	×
358
359	autotests_for_update = self._escape_html_in_model(autotests_for_update)	×
360	self.__autotest_api.update_multiple(auto_test_put_model=autotests_for_update)	×
361
362	logging.debug(f'Autotests were updated')	×
363
364	@adapter_logger	2✔
365	@retry	2✔
366	def __unlink_test_to_work_item(self, autotest_global_id: str, work_item_id: str):	2✔
367	self.__autotest_api.delete_auto_test_link_from_work_item(	×
368	id=autotest_global_id,
369	work_item_id=work_item_id)
370
371	logging.debug(f'Autotest was unlinked with workItem "{work_item_id}" by global id "{autotest_global_id}')	×
372
373	@adapter_logger	2✔
374	@retry	2✔
375	def __link_test_to_work_item(self, autotest_global_id: str, work_item_id: str):	2✔
376	self.__autotest_api.link_auto_test_to_work_item(	×
377	autotest_global_id,
378	work_item_id_model=WorkItemIdModel(id=work_item_id))
379
380	logging.debug(f'Autotest was linked with workItem "{work_item_id}" by global id "{autotest_global_id}')	×
381
382	@adapter_logger	2✔
383	def __load_test_result(self, test_result: TestResult) -> str:	2✔
384	model = Converter.test_result_to_testrun_result_post_model(	×
385	test_result,
386	self.__config.get_configuration_id())
387	model = self._escape_html_in_model(model)	×
388
389	response = self.__test_run_api.set_auto_test_results_for_test_run(	×
390	id=self.__config.get_test_run_id(),
391	auto_test_results_for_test_run_model=[model])
392
393	logging.debug(f'Result of the autotest "{test_result.get_autotest_name()}" was set '	×
394	f'in the test run "{self.__config.get_test_run_id()}"')
395
396	return Converter.get_test_result_id_from_testrun_result_post_response(response)	×
397
398	@adapter_logger	2✔
399	def __load_test_results(self, test_results: typing.List[AutoTestResultsForTestRunModel]):	2✔
400	logging.debug(f'Loading test results: {test_results}')	×
401
402	test_results = self._escape_html_in_model(test_results)	×
403	self.__test_run_api.set_auto_test_results_for_test_run(	×
404	id=self.__config.get_test_run_id(),
405	auto_test_results_for_test_run_model=test_results)
406
407	@adapter_logger	2✔
408	def get_test_result_by_id(self, test_result_id: str) -> TestResultResponse:	2✔
409	return self.__test_results_api.api_v2_test_results_id_get(id=test_result_id)	×
410
411	@adapter_logger	2✔
412	def update_test_results(self, fixtures_containers: dict, test_result_ids: dict):	2✔
413	test_results = Converter.fixtures_containers_to_test_results_with_all_fixture_step_results(	×
414	fixtures_containers, test_result_ids)
415
416	for test_result in test_results:	×
417	model = Converter.convert_test_result_model_to_test_results_id_put_request(	×
418	self.get_test_result_by_id(test_result.get_test_result_id()))
419
420	model.setup_results = Converter.step_results_to_auto_test_step_result_update_request(	×
421	test_result.get_setup_results())
422	model.teardown_results = Converter.step_results_to_auto_test_step_result_update_request(	×
423	test_result.get_teardown_results())
424
425	model = self._escape_html_in_model(model)	×
426
427	try:	×
428	self.__test_results_api.api_v2_test_results_id_put(	×
429	id=test_result.get_test_result_id(),
430	api_v2_test_results_id_put_request=model)
431	except Exception as exc:	×
432	logging.error(f'Cannot update test result with id "{test_result.get_test_result_id()}" status: {exc}')	×
433
434	@adapter_logger	2✔
435	def load_attachments(self, attach_paths: list or tuple):	2✔
436	attachments = []	×
437
438	for path in attach_paths:	×
439	if os.path.isfile(path):	×
440	try:	×
NEW 441	attachment_response = self.__attachments_api.api_v2_attachments_post(file=open(path, "rb"))	×
442
NEW 443	attachments.append(AttachmentPutModel(attachment_response['id']))	×
444
445	logging.debug(f'Attachment "{path}" was uploaded')	×
446	except Exception as exc:	×
447	logging.error(f'Upload attachment "{path}" status: {exc}')	×
448	else:
449	logging.error(f'File "{path}" was not found!')	×
450	return attachments	×

testit-tms / adapters-python / 16521578080

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous