• Home
  • Features
  • Pricing
  • Docs
  • Announcements
  • Sign In

testit-tms / adapters-python / 17091364703

20 Aug 2025 07:15AM UTC coverage: 36.848% (+0.4%) from 36.441%
17091364703

Pull #199

github

web-flow
Merge a58423d45 into e79fbf989
Pull Request #199: fix: fix updating autotest links to workitems.

53 of 147 new or added lines in 3 files covered. (36.05%)

1 existing line in 1 file now uncovered.

1237 of 3357 relevant lines covered (36.85%)

0.74 hits per line

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

32.6
/testit-python-commons/src/testit_python_commons/client/api_client.py
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
    LinkAutoTestToWorkItemRequest,
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.client.helpers.bulk_autotest_helper import BulkAutotestHelper
2✔
22
from testit_python_commons.models.test_result import TestResult
2✔
23
from testit_python_commons.services.logger import adapter_logger
2✔
24
from testit_python_commons.services.retry import retry
2✔
25
from testit_python_commons.utils.html_escape_utils import HtmlEscapeUtils
2✔
26

27

28
class ApiClientWorker:
2✔
29
    __max_tests_for_write = 100
2✔
30

31
    def __init__(self, config: ClientConfiguration):
2✔
32
        api_client_config = self.__get_api_client_configuration(
×
33
            url=config.get_url(),
34
            verify_ssl=config.get_cert_validation() != 'false',
35
            proxy=config.get_proxy())
36
        api_client = self.__get_api_client(api_client_config, config.get_private_token())
×
37

38
        self.__test_run_api = TestRunsApi(api_client=api_client)
×
39
        self.__autotest_api = AutoTestsApi(api_client=api_client)
×
40
        self.__attachments_api = AttachmentsApi(api_client=api_client)
×
41
        self.__test_results_api = TestResultsApi(api_client=api_client)
×
42
        self.__work_items_api = WorkItemsApi(api_client=api_client)
×
43
        self.__config = config
×
44

45
    @staticmethod
2✔
46
    @adapter_logger
2✔
47
    def __get_api_client_configuration(url: str, verify_ssl: bool = True, proxy: str = None) -> Configuration:
2✔
48
        api_client_configuration = Configuration(host=url)
×
49
        api_client_configuration.verify_ssl = verify_ssl
×
50
        api_client_configuration.proxy = proxy
×
51

52
        return api_client_configuration
×
53

54
    @staticmethod
2✔
55
    @adapter_logger
2✔
56
    def __get_api_client(api_client_config: Configuration, token: str) -> ApiClient:
2✔
57
        return ApiClient(
×
58
            configuration=api_client_config,
59
            header_name='Authorization',
60
            header_value='PrivateToken ' + token)
61

62
    @staticmethod
2✔
63
    def _escape_html_in_model(model):
2✔
64
        """Apply HTML escaping to all models before sending to API"""
65
        return HtmlEscapeUtils.escape_html_in_object(model)
×
66

67
    @adapter_logger
2✔
68
    def create_test_run(self):
2✔
69
        test_run_name = f'TestRun_{datetime.today().strftime("%Y-%m-%dT%H:%M:%S")}' if \
×
70
            not self.__config.get_test_run_name() else self.__config.get_test_run_name()
71
        model = Converter.test_run_to_test_run_short_model(
×
72
            self.__config.get_project_id(),
73
            test_run_name
74
        )
75
        model = self._escape_html_in_model(model)
×
76

77
        response = self.__test_run_api.create_empty(create_empty_request=model)
×
78

79
        return Converter.get_id_from_create_test_run_response(response)
×
80

81
    @adapter_logger
2✔
82
    def set_test_run_id(self, test_run_id: str):
2✔
83
        self.__config.set_test_run_id(test_run_id)
×
84

85
    @adapter_logger
2✔
86
    def get_autotests_by_test_run_id(self):
2✔
87
        response = self.__test_run_api.get_test_run_by_id(self.__config.get_test_run_id())
×
88

89
        return Converter.get_resolved_autotests_from_get_test_run_response(
×
90
            response,
91
            self.__config.get_configuration_id())
92

93
    @adapter_logger
2✔
94
    def __get_autotests_by_external_id(self, external_id: str) -> list:
2✔
NEW
95
        model = Converter.project_id_and_external_id_to_auto_tests_search_post_request(
×
96
            self.__config.get_project_id(),
97
            external_id)
98

NEW
99
        return self.__autotest_api.api_v2_auto_tests_search_post(api_v2_auto_tests_search_post_request=model)
×
100

101
    @adapter_logger
2✔
102
    def write_test(self, test_result: TestResult):
2✔
103
        model = Converter.project_id_and_external_id_to_auto_tests_search_post_request(
×
104
            self.__config.get_project_id(),
105
            test_result.get_external_id())
106

107
        autotests = self.__autotest_api.api_v2_auto_tests_search_post(api_v2_auto_tests_search_post_request=model)
×
108

109
        if autotests:
×
110
            self.__update_test(test_result, autotests[0])
×
111

NEW
112
            autotest_id = autotests[0].id
×
113

NEW
114
            self.__update_autotest_link_from_work_items(autotest_id, test_result.get_work_item_ids())
×
115
        else:
116
            self.__create_test(test_result)
×
117

118
        return self.__load_test_result(test_result)
×
119

120
    @adapter_logger
2✔
121
    def write_tests(self, test_results: typing.List[TestResult], fixture_containers: dict):
2✔
NEW
122
        bulk_autotest_helper = BulkAutotestHelper(self.__autotest_api, self.__test_run_api, self.__config)
×
123

124
        for test_result in test_results:
×
125
            test_result = self.__add_fixtures_to_test_result(test_result, fixture_containers)
×
126

127
            test_result_model = Converter.test_result_to_testrun_result_post_model(
×
128
                test_result,
129
                self.__config.get_configuration_id())
130

NEW
131
            work_item_ids_for_link_with_auto_test = self.__get_work_item_uuids_for_link_with_auto_test(
×
132
                test_result.get_work_item_ids())
133

NEW
134
            autotests = self.__get_autotests_by_external_id(test_result.get_external_id())
×
135

NEW
136
            if autotests:
×
NEW
137
                autotest_links_to_wi_for_update = {}
×
NEW
138
                autotest_for_update = Converter.prepare_to_mass_update_autotest(
×
139
                    test_result,
140
                    autotests[0],
141
                    self.__config.get_project_id())
142

NEW
143
                autotest_id = autotests[0].id
×
NEW
144
                autotest_links_to_wi_for_update[autotest_id] = test_result.get_work_item_ids()
×
145

NEW
146
                bulk_autotest_helper.add_for_update(
×
147
                    autotest_for_update,
148
                    test_result_model,
149
                    autotest_links_to_wi_for_update)
150
            else:
NEW
151
                autotest_for_create = Converter.prepare_to_mass_create_autotest(
×
152
                    test_result,
153
                    self.__config.get_project_id(),
154
                    work_item_ids_for_link_with_auto_test)
155

NEW
156
                bulk_autotest_helper.add_for_create(autotest_for_create, test_result_model)
×
157

NEW
158
        bulk_autotest_helper.teardown()
×
159

160
    @staticmethod
2✔
161
    @adapter_logger
2✔
162
    def __add_fixtures_to_test_result(
2✔
163
            test_result: TestResult,
164
            fixtures_containers: dict) -> TestResult:
165
        setup_results = []
×
166
        teardown_results = []
×
167

168
        for uuid, fixtures_container in fixtures_containers.items():
×
169
            if test_result.get_external_id() in fixtures_container.external_ids:
×
170
                if fixtures_container.befores:
×
171
                    setup_results += fixtures_container.befores[0].steps
×
172

173
                if fixtures_container.afters:
×
174
                    teardown_results = fixtures_container.afters[0].steps + teardown_results
×
175

176
        test_result.set_setup_results(setup_results)
×
177
        test_result.set_teardown_results(teardown_results)
×
178

179
        return test_result
×
180

181
    @adapter_logger
2✔
182
    def __get_work_item_uuids_for_link_with_auto_test(
2✔
183
            self,
184
            work_item_ids: list,
185
            autotest_global_id: str = None) -> list:
186
        linked_work_items = []
×
187

188
        if autotest_global_id:
×
189
            linked_work_items = self.__get_work_items_linked_to_autotest(autotest_global_id)
×
190

191
        work_item_uuids = self.__prepare_list_of_work_item_uuids(linked_work_items, work_item_ids)
×
192

193
        return work_item_uuids
×
194

195
    @adapter_logger
2✔
196
    def __prepare_list_of_work_item_uuids(
2✔
197
            self,
198
            linked_work_items: list,
199
            work_item_ids: list) -> list:
200
        work_item_uuids = []
×
201

202
        for linked_work_item in linked_work_items:
×
203
            linked_work_item_id = str(linked_work_item.global_id)
×
204
            linked_work_item_uuid = linked_work_item.id
×
205

206
            if linked_work_item_id in work_item_ids:
×
207
                work_item_ids.remove(linked_work_item_id)
×
208
                work_item_uuids.append(linked_work_item_uuid)
×
209

210
                continue
×
211

212
            if self.__config.get_automatic_updation_links_to_test_cases() != 'true':
×
213
                work_item_uuids.append(linked_work_item_uuid)
×
214

215
        for work_item_id in work_item_ids:
×
216
            work_item_uuid = self.__get_work_item_uuid_by_work_item_id(work_item_id)
×
217

218
            if work_item_uuid:
×
219
                work_item_uuids.append(work_item_uuid)
×
220

221
        return work_item_uuids
×
222

223
    @adapter_logger
2✔
224
    def __get_work_item_uuid_by_work_item_id(self, work_item_id: str) -> str or None:
2✔
225
        logging.debug('Getting workitem by id ' + work_item_id)
×
226

227
        try:
×
228
            work_item = self.__work_items_api.get_work_item_by_id(id=work_item_id)
×
229

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

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

236
    @adapter_logger
2✔
237
    def __get_work_items_linked_to_autotest(self, autotest_global_id: str) -> typing.List[WorkItemIdentifierModel]:
2✔
238
        return self.__autotest_api.get_work_items_linked_to_auto_test(id=autotest_global_id)
×
239

240
    @adapter_logger
2✔
241
    def __update_autotest_link_from_work_items(self, autotest_global_id: str, work_item_ids: list):
2✔
242
        linked_work_items = self.__get_work_items_linked_to_autotest(autotest_global_id)
×
243

244
        for linked_work_item in linked_work_items:
×
245
            linked_work_item_id = str(linked_work_item.global_id)
×
246

247
            if linked_work_item_id in work_item_ids:
×
248
                work_item_ids.remove(linked_work_item_id)
×
249

250
                continue
×
251

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

255
        for work_item_id in work_item_ids:
×
256
            self.__link_test_to_work_item(autotest_global_id, work_item_id)
×
257

258
    @adapter_logger
2✔
259
    def __create_test(self, test_result: TestResult) -> str:
2✔
260
        logging.debug(f'Autotest "{test_result.get_autotest_name()}" was not found')
×
261

NEW
262
        work_item_ids_for_link_with_auto_test = self.__get_work_item_uuids_for_link_with_auto_test(
×
263
            test_result.get_work_item_ids())
264

NEW
265
        model = Converter.prepare_to_create_autotest(
×
266
            test_result,
267
            self.__config.get_project_id(),
268
            work_item_ids_for_link_with_auto_test)
UNCOV
269
        model = self._escape_html_in_model(model)
×
270

271
        autotest_response = self.__autotest_api.create_auto_test(create_auto_test_request=model)
×
272

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

275
        return autotest_response.id
×
276

277
    @adapter_logger
2✔
278
    def __create_tests(self, autotests_for_create: typing.List[AutoTestPostModel]):
2✔
279
        logging.debug(f'Creating autotests: "{autotests_for_create}')
×
280

281
        autotests_for_create = self._escape_html_in_model(autotests_for_create)
×
282
        self.__autotest_api.create_multiple(auto_test_post_model=autotests_for_create)
×
283

284
        logging.debug(f'Autotests were created')
×
285

286
    @adapter_logger
2✔
287
    def __update_test(self, test_result: TestResult, autotest: AutoTestApiResult):
2✔
288
        logging.debug(f'Autotest "{test_result.get_autotest_name()}" was found')
×
289

NEW
290
        model = Converter.prepare_to_update_autotest(test_result, autotest, self.__config.get_project_id())
×
291
        model = self._escape_html_in_model(model)
×
292

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

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

300
    @adapter_logger
2✔
301
    def __update_tests(self, autotests_for_update: typing.List[AutoTestPutModel]):
2✔
302
        logging.debug(f'Updating autotests: {autotests_for_update}')
×
303

304
        autotests_for_update = self._escape_html_in_model(autotests_for_update)
×
305
        self.__autotest_api.update_multiple(auto_test_put_model=autotests_for_update)
×
306

307
        logging.debug(f'Autotests were updated')
×
308

309
    @adapter_logger
2✔
310
    @retry
2✔
311
    def __unlink_test_to_work_item(self, autotest_global_id: str, work_item_id: str):
2✔
312
        self.__autotest_api.delete_auto_test_link_from_work_item(
×
313
            id=autotest_global_id,
314
            work_item_id=work_item_id)
315

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

318
    @adapter_logger
2✔
319
    @retry
2✔
320
    def __link_test_to_work_item(self, autotest_global_id: str, work_item_id: str):
2✔
321
        self.__autotest_api.link_auto_test_to_work_item(
×
322
            autotest_global_id,
323
            link_auto_test_to_work_item_request=LinkAutoTestToWorkItemRequest(id=work_item_id))
324

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

327
    @adapter_logger
2✔
328
    def __load_test_result(self, test_result: TestResult) -> str:
2✔
329
        model = Converter.test_result_to_testrun_result_post_model(
×
330
            test_result,
331
            self.__config.get_configuration_id())
332
        model = self._escape_html_in_model(model)
×
333

334
        response = self.__test_run_api.set_auto_test_results_for_test_run(
×
335
            id=self.__config.get_test_run_id(),
336
            auto_test_results_for_test_run_model=[model])
337

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

341
        return Converter.get_test_result_id_from_testrun_result_post_response(response)
×
342

343
    @adapter_logger
2✔
344
    def __load_test_results(self, test_results: typing.List[AutoTestResultsForTestRunModel]):
2✔
345
        logging.debug(f'Loading test results: {test_results}')
×
346

347
        test_results = self._escape_html_in_model(test_results)
×
348
        self.__test_run_api.set_auto_test_results_for_test_run(
×
349
            id=self.__config.get_test_run_id(),
350
            auto_test_results_for_test_run_model=test_results)
351

352
    @adapter_logger
2✔
353
    def get_test_result_by_id(self, test_result_id: str) -> TestResultResponse:
2✔
354
        return self.__test_results_api.api_v2_test_results_id_get(id=test_result_id)
×
355

356
    @adapter_logger
2✔
357
    def update_test_results(self, fixtures_containers: dict, test_result_ids: dict):
2✔
358
        test_results = Converter.fixtures_containers_to_test_results_with_all_fixture_step_results(
×
359
            fixtures_containers, test_result_ids)
360

361
        for test_result in test_results:
×
362
            model = Converter.convert_test_result_model_to_test_results_id_put_request(
×
363
                self.get_test_result_by_id(test_result.get_test_result_id()))
364

365
            model.setup_results = Converter.step_results_to_auto_test_step_result_update_request(
×
366
                    test_result.get_setup_results())
367
            model.teardown_results = Converter.step_results_to_auto_test_step_result_update_request(
×
368
                    test_result.get_teardown_results())
369
            
370
            model = self._escape_html_in_model(model)
×
371

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

379
    @adapter_logger
2✔
380
    def load_attachments(self, attach_paths: list or tuple):
2✔
381
        attachments = []
×
382

383
        for path in attach_paths:
×
384
            if os.path.isfile(path):
×
385
                try:
×
386
                    attachment_response = self.__attachments_api.api_v2_attachments_post(file=open(path, "rb"))
×
387

388
                    attachments.append(AttachmentPutModel(attachment_response['id']))
×
389

390
                    logging.debug(f'Attachment "{path}" was uploaded')
×
391
                except Exception as exc:
×
392
                    logging.error(f'Upload attachment "{path}" status: {exc}')
×
393
            else:
394
                logging.error(f'File "{path}" was not found!')
×
395
        return attachments
×
STATUS · Troubleshooting · Open an Issue · Sales · Support · CAREERS · ENTERPRISE · START FREE · SCHEDULE DEMO
ANNOUNCEMENTS · TWITTER · TOS & SLA · Supported CI Services · What's a CI service? · Automated Testing

© 2026 Coveralls, Inc