17402521449

Committed 02 Sep 2025 11:47AM UTC coverage: 36.535% (-0.3%) from 36.848%

Build # 17402521449

Build Type

Pull #200

github

Committed by

web-flow

Commit Message

Merge 9ddb47f3d into 6559ad79f

Pull Request Pull Request #200: feat: TMS-33268: support the "status code" field.

Run Details

31 of 78 new or added lines in 5 files covered. (39.74%)

4 existing lines in 2 files now uncovered.

1240 of 3394 relevant lines covered (36.54%)

0.73 hits per line

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

29.25

/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 (
    ApiV2TestResultsSearchPostRequest,
    AutoTestApiResult,
    AutoTestPostModel,
    AutoTestPutModel,
    AttachmentPutModel,
    ApiV2AutoTestsSearchPostRequest,
    TestResultResponse,
    TestResultShortResponse,
    LinkAutoTestToWorkItemRequest,
    WorkItemIdentifierModel
)

from testit_python_commons.client.client_configuration import ClientConfiguration
from testit_python_commons.client.converter import Converter
from testit_python_commons.client.helpers.bulk_autotest_helper import BulkAutotestHelper
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:
    __tests_limit = 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: typing.Any) -> typing.Any:
        """Apply HTML escaping to all models before sending to API"""
        return HtmlEscapeUtils.escape_html_in_object(model)

    @adapter_logger
    def create_test_run(self) -> str:
        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) -> None:
        self.__config.set_test_run_id(test_run_id)

    @adapter_logger
    def get_external_ids_for_test_run_id(self) -> typing.List[str]:
        test_results: typing.List[TestResultShortResponse] = self.__get_test_results()
        autotest_ids: typing.List[int] = Converter.get_global_ids_from_autotest_response_list (
            test_results,
            self.__config.get_configuration_id())

        if len(autotest_ids) == 0:
            raise Exception('The autotests with the status "InProgress" ' +
                            f'and the configuration id "{self.__config.get_configuration_id()}" were not found!')

        autotests_search_post_request: ApiV2AutoTestsSearchPostRequest = (
            Converter.autotest_ids_to_autotests_search_post_request(autotest_ids))
        autotests: typing.List[AutoTestApiResult] = self.__get_autotests(autotests_search_post_request)
        external_ids: typing.List[str] = Converter.autotest_results_to_external_ids(autotests)

        return external_ids

    def __get_test_results(self) -> typing.List[TestResultShortResponse]:
        all_test_results = []
        skip = 0
        model: ApiV2TestResultsSearchPostRequest = (
            Converter.build_test_results_search_post_request_with_in_progress_outcome(
                self.__config.get_test_run_id(),
                self.__config.get_configuration_id()))

        while skip >= 0:
            logging.debug(f"Getting test results with limit {self.__tests_limit}: {model}")

            test_results: typing.List[TestResultShortResponse] = self.__test_results_api.api_v2_test_results_search_post(
                skip=skip,
                take=self.__tests_limit,
                api_v2_test_results_search_post_request=model)

            logging.debug(f"Got {len(test_results)} test results: {test_results}")

            all_test_results.extend(test_results)
            skip += self.__tests_limit

            if len(test_results) == 0:
                skip = -1

        return all_test_results

    def __get_autotests(self, model: ApiV2AutoTestsSearchPostRequest) \
            -> typing.List[AutoTestApiResult]:
        """Function returns list of AutoTestApiResult."""
        all_autotests = []
        skip = 0

        while skip >= 0:
            logging.debug(f"Getting autotests with limit {self.__tests_limit}: {model}")

            autotests: typing.List[AutoTestApiResult] = self.__autotest_api.api_v2_auto_tests_search_post(
                skip=skip,
                take=self.__tests_limit,
                api_v2_auto_tests_search_post_request=model)

            logging.debug(f"Got autotests: {autotests}")

            all_autotests.extend(autotests)
            skip += self.__tests_limit

            if len(autotests) == 0:
                skip = -1

        return all_autotests

    @adapter_logger
    def __get_autotests_by_external_id(self, external_id: str) -> typing.List[AutoTestApiResult]:
        model = Converter.project_id_and_external_id_to_auto_tests_search_post_request(
            self.__config.get_project_id(),
            external_id)

        return self.__autotest_api.api_v2_auto_tests_search_post(api_v2_auto_tests_search_post_request=model)

    @adapter_logger
    def write_test(self, test_result: TestResult) -> str:
        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])

            autotest_id = autotests[0].id

            self.__update_autotest_link_from_work_items(autotest_id, test_result.get_work_item_ids())
        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) -> None:
        bulk_autotest_helper = BulkAutotestHelper(self.__autotest_api, self.__test_run_api, self.__config)

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

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

            work_item_ids_for_link_with_auto_test = self.__get_work_item_uuids_for_link_with_auto_test(
                test_result.get_work_item_ids())

            autotests = self.__get_autotests_by_external_id(test_result.get_external_id())

            if autotests:
                autotest_links_to_wi_for_update = {}
                autotest_for_update = Converter.prepare_to_mass_update_autotest(
                    test_result,
                    autotests[0],
                    self.__config.get_project_id())

                autotest_id = autotests[0].id
                autotest_links_to_wi_for_update[autotest_id] = test_result.get_work_item_ids()

                bulk_autotest_helper.add_for_update(
                    autotest_for_update,
                    test_result_model,
                    autotest_links_to_wi_for_update)
            else:
                autotest_for_create = Converter.prepare_to_mass_create_autotest(
                    test_result,
                    self.__config.get_project_id(),
                    work_item_ids_for_link_with_auto_test)

                bulk_autotest_helper.add_for_create(autotest_for_create, test_result_model)

        bulk_autotest_helper.teardown()

    @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: typing.List[str],
            autotest_global_id: str = None) -> typing.List[str]:
        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: typing.List[WorkItemIdentifierModel],
            work_item_ids: typing.List[str]) -> typing.List[str]:
        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')

        work_item_ids_for_link_with_auto_test = self.__get_work_item_uuids_for_link_with_auto_test(
            test_result.get_work_item_ids())

        model = Converter.prepare_to_create_autotest(
            test_result,
            self.__config.get_project_id(),
            work_item_ids_for_link_with_auto_test)
        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]) -> None:
        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) -> None:
        logging.debug(f'Autotest "{test_result.get_autotest_name()}" was found')

        model = Converter.prepare_to_update_autotest(test_result, autotest, self.__config.get_project_id())
        model = self._escape_html_in_model(model)

        try:
            self.__autotest_api.update_auto_test(update_auto_test_request=model)
        except Exception as exc:
            logging.error(f'Cannot update autotest "{test_result.get_autotest_name()}" status: {exc}')

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

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

testit-tms / adapters-python / 17402521449

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