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

testit-tms / adapters-python / 22724059698

05 Mar 2026 03:05PM UTC coverage: 17.381% (-20.8%) from 38.156%
22724059698

push

github

web-flow
Feat/sync storage (#243)

feat: add sync-storage implementation

refactor app_properties
add sync_storage_port variable

1 of 22 new or added lines in 4 files covered. (4.55%)

51 existing lines in 6 files now uncovered.

280 of 1611 relevant lines covered (17.38%)

0.35 hits per line

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

0.0
/testit-adapter-pytest/src/testit_adapter_pytest/utils.py
1
import hashlib
×
2
import logging
×
3
import re
×
4
import traceback
×
5
from typing import List
×
6

7
import pytest
×
8

9
from traceback import format_exception_only
×
10

11
from _pytest.mark import Mark
×
12
from testit_python_commons.models.link import Link
×
13
from testit_python_commons.models.test_result import TestResult
×
14
from testit_python_commons.models.test_result_with_all_fixture_step_results_model import TestResultWithAllFixtureStepResults
×
15

16
from testit_adapter_pytest.models.executable_test import ExecutableTest
×
17

18

19
__ARRAY_TYPES = (frozenset, list, set, tuple,)
×
20

21

22
def form_test(item) -> ExecutableTest:
×
23
    executable_test = ExecutableTest(
×
24
        external_id=__get_external_id_from(item),
25
        name=__get_display_name_from(item),
26
        duration=0,
27
        parameters=__get_parameters_from(item),
28
        properties=__get_properties_from(item),
29
        namespace=__get_namespace_from(item),
30
        classname=__get_class_name_from(item),
31
        title=__get_title_from(item),
32
        description=__get_description_from(item),
33
        links=__get_links_from(item),
34
        labels=__get_labels_from(item),
35
        tags=__get_tags_from(item),
36
        work_item_ids=__get_work_item_ids_from(item),
37
        node_id=item.nodeid
38
    )
39

40
    if item.own_markers:
×
41
        executable_test = __set_outcome_and_message_from_markers(executable_test, item.own_markers)
×
42

43
    return executable_test
×
44

45

46
def __set_outcome_and_message_from_markers(executable_test: ExecutableTest, markers: List[Mark]) -> ExecutableTest:
×
47
    for marker in markers:
×
48
        if marker.name in ('skip', 'skipif'):
×
49
            executable_test.outcome = 'Skipped'
×
50
        if marker.name in ('skip', 'skipif', 'xfail'):
×
51
            if len(marker.args) == 1 and isinstance(marker.args, str):
×
52
                executable_test.message = marker.args[0]
×
53

54
            condition_in_args = marker.args and marker.args[0]
×
55
            condition_in_kwargs = marker.kwargs and 'condition' in marker.kwargs and marker.kwargs['condition']
×
56

57
            if (condition_in_kwargs or condition_in_args) and 'reason' in marker.kwargs:
×
58
                executable_test.message = marker.kwargs['reason']
×
59

60
    return executable_test
×
61

62

63
def __get_display_name_from(item):
×
64
    display_name = __search_attribute(item, 'test_displayname')
×
65

66
    if not display_name:
×
67
        return item.function.__doc__ if \
×
68
            item.function.__doc__ else item.function.__name__
69

70
    return collect_parameters_in_string_attribute(display_name, get_all_parameters(item))
×
71

72

73
def __get_external_id_from(item):
×
74
    external_id = __search_attribute(item, 'test_external_id')
×
75

76
    if not external_id:
×
77
        return get_hash(item.parent.nodeid + item.function.__name__)
×
78

79
    return collect_parameters_in_string_attribute(external_id, get_all_parameters(item))
×
80

81

82
def __get_title_from(item):
×
83
    title = __search_attribute(item, 'test_title')
×
84

85
    if not title:
×
86
        return None
×
87

88
    return collect_parameters_in_string_attribute(title, get_all_parameters(item))
×
89

90

91
def __get_description_from(item):
×
92
    description = __search_attribute(item, 'test_description')
×
93

94
    if not description:
×
95
        return None
×
96

97
    return collect_parameters_in_string_attribute(description, get_all_parameters(item))
×
98

99

100
def __get_namespace_from(item):
×
101
    namespace = __search_attribute(item, 'test_namespace')
×
102

103
    if not namespace:
×
104
        return item.function.__module__
×
105

106
    return collect_parameters_in_string_attribute(namespace, get_all_parameters(item))
×
107

108

109
def __get_class_name_from(item):
×
110
    class_name = __search_attribute(item, 'test_classname')
×
111

112
    if not class_name:
×
113
        i = item.function.__qualname__.find('.')
×
114

115
        if i != -1:
×
116
            return item.function.__qualname__[:i]
×
117

118
        return None
×
119

120
    return collect_parameters_in_string_attribute(class_name, get_all_parameters(item))
×
121

122

123
def __get_links_from(item):
×
124
    links = __search_attribute(item, 'test_links')
×
125

126
    if not links:
×
127
        return []
×
128

129
    return __set_parameters_to_links(links, get_all_parameters(item))
×
130

131

132
def __get_parameters_from(item):
×
133
    test_parameters = {}
×
134

135
    if hasattr(item, 'callspec'):
×
136
        for key, parameter in item.callspec.params.items():
×
137
            test_parameters[key] = str(parameter)
×
138

139
    return test_parameters
×
140

141

142
def __get_properties_from(item):
×
143
    if hasattr(item, 'test_properties'):
×
144
        return item.test_properties
×
145
    return None
×
146

147

148
def __set_parameters_to_links(links, all_parameters):
×
149
    if not all_parameters:
×
150
        return links
×
151

152
    links_with_parameters = []
×
153

154
    for link in links:
×
155
        links_with_parameters.append(
×
156
            Link()
157
            .set_url(
158
                collect_parameters_in_string_attribute(
159
                    link.get_url(),
160
                    all_parameters))
161
            .set_title(
162
                collect_parameters_in_string_attribute(
163
                    link.get_title(),
164
                    all_parameters) if link.get_title() else None)
165
            .set_link_type(
166
                collect_parameters_in_string_attribute(
167
                    link.get_link_type(),
168
                    all_parameters) if link.get_link_type() else None)
169
            .set_description(
170
                collect_parameters_in_string_attribute(
171
                    link.get_description(),
172
                    all_parameters) if link.get_description() else None))
173

174
    return links_with_parameters
×
175

176

177
def __get_labels_from(item) -> List[dict]:
×
178
    test_labels = __search_attribute(item, 'test_labels')
×
179

180
    if not test_labels:
×
181
        return []
×
182

183
    labels = []
×
184

185
    for label in test_labels:
×
186
        result = collect_parameters_in_mass_attribute(
×
187
            label,
188
            get_all_parameters(item))
189

190
        if isinstance(result, __ARRAY_TYPES):
×
191
            for l in result:
×
192
                labels.append({
×
193
                    'name': str(l)
194
                })
195
        else:
196
            labels.append({
×
197
                'name': str(result)
198
            })
199

200
    return labels
×
201

202

203
def __get_tags_from(item) -> List[str]:
×
204
    test_tags = __search_attribute(item, 'test_tags')
×
205

206
    if not test_tags:
×
207
        return []
×
208

209
    tags = []
×
210

211
    for tag in test_tags:
×
212
        result = collect_parameters_in_mass_attribute(
×
213
            tag,
214
            get_all_parameters(item))
215

216
        if isinstance(result, __ARRAY_TYPES):
×
217
            for t in result:
×
218
                tags.append(str(t))
×
219
        else:
220
            tags.append(str(result))
×
221

222
    return tags
×
223

224

225
def __get_work_item_ids_from(item):
×
226
    test_workitem_ids = __search_attribute(item, 'test_workitems_id')
×
227

228
    if not test_workitem_ids:
×
229
        return []
×
230

231
    workitem_ids = []
×
232

233
    for workitem_id in test_workitem_ids:
×
234
        result = collect_parameters_in_mass_attribute(
×
235
            workitem_id,
236
            get_all_parameters(item))
237

238
        if isinstance(result, __ARRAY_TYPES):
×
239
            workitem_ids += list(map(str, result))
×
240
        else:
241
            workitem_ids.append(str(result))
×
242

243
    return workitem_ids
×
244

245

246
def collect_parameters_in_string_attribute(attribute, all_parameters):
×
247
    param_keys = re.findall(r"\{(.*?)\}", attribute)
×
248

249
    if len(param_keys) > 0:
×
250
        for param_key in param_keys:
×
251
            parameter = get_parameter(param_key, all_parameters)
×
252

253
            if parameter is not None:
×
254
                attribute = attribute.replace("{" + param_key + "}", str(parameter))
×
255

256
    return attribute
×
257

258

259
def collect_parameters_in_mass_attribute(attribute, all_parameters):
×
260
    param_keys = re.findall(r"\{(.*?)\}", attribute)
×
261

262
    if len(param_keys) == 1:
×
263
        parameter = get_parameter(param_keys[0], all_parameters)
×
264

265
        if parameter is not None:
×
266
            return parameter
×
267

268
    if len(param_keys) > 1:
×
269
        logging.error(f'(For type tuple, list, set) support only one key!')
×
270

271
    return attribute
×
272

273

274
def get_parameter(key_for_parameter, all_parameters):
×
275
    id_keys_in_parameter = re.findall(r'\[(.*?)\]', key_for_parameter)
×
276

277
    if len(id_keys_in_parameter) > 1:
×
278
        logging.error("(For type tuple, list, set, dict) support only one level!")
×
279

280
        return
×
281

282
    if len(id_keys_in_parameter) == 0:
×
283
        if key_for_parameter not in all_parameters:
×
284
            logging.error(f"Key for parameter {key_for_parameter} not found")
×
285

286
            return
×
287

288
        return all_parameters[key_for_parameter]
×
289

290
    parameter_key = key_for_parameter.replace("[" + id_keys_in_parameter[0] + "]", "")
×
291
    id_key_in_parameter = id_keys_in_parameter[0].strip("\'\"")
×
292

293
    if id_key_in_parameter.isdigit() and int(id_key_in_parameter) in range(len(all_parameters[parameter_key])):
×
294
        return all_parameters[parameter_key][int(id_key_in_parameter)]
×
295

296
    if id_key_in_parameter.isalnum() and id_key_in_parameter in all_parameters[parameter_key].keys():
×
297
        return all_parameters[parameter_key][id_key_in_parameter]
×
298

299
    logging.error(f"Not key: {key_for_parameter} in run parameters or other keys problem")
×
300

301

302
def get_all_parameters(item):
×
303
    params = {}
×
304

305
    if hasattr(item, 'test_properties'):
×
306
        params.update(item.test_properties)
×
307

308
    if hasattr(item, 'callspec'):
×
309
        params.update(item.callspec.params)
×
310

311
    return params
×
312

313

314
def __search_attribute(item, attribute):
×
315
    if hasattr(item.function, attribute):
×
316
        return getattr(item.function, attribute)
×
317

318
    if hasattr(item.cls, attribute):
×
319
        return getattr(item.cls, attribute)
×
320

321
    return
×
322

323

324
def get_hash(value: str):
×
325
    md = hashlib.sha256(bytes(value, encoding='utf-8'))
×
326
    return md.hexdigest()
×
327

328

329
def convert_executable_test_to_test_result_model(executable_test: ExecutableTest) -> TestResult:
×
330
    pytest_autotest_keys = convert_node_id_to_pytest_autotest_keys(executable_test.node_id)
×
331

332
    return TestResult()\
×
333
        .set_external_id(executable_test.external_id)\
334
        .set_autotest_name(executable_test.name)\
335
        .set_step_results(executable_test.step_results)\
336
        .set_setup_results(executable_test.setup_step_results)\
337
        .set_teardown_results(executable_test.teardown_step_results)\
338
        .set_duration(executable_test.duration)\
339
        .set_outcome(executable_test.outcome)\
340
        .set_traces(executable_test.traces)\
341
        .set_attachments(executable_test.attachments)\
342
        .set_parameters(executable_test.parameters)\
343
        .set_properties(executable_test.properties)\
344
        .set_namespace(executable_test.namespace)\
345
        .set_classname(executable_test.classname)\
346
        .set_title(executable_test.title)\
347
        .set_description(executable_test.description)\
348
        .set_links(executable_test.links)\
349
        .set_result_links(executable_test.result_links)\
350
        .set_labels(executable_test.labels)\
351
        .set_tags(executable_test.tags)\
352
        .set_work_item_ids(executable_test.work_item_ids) \
353
        .set_message(executable_test.message) \
354
        .set_external_key(pytest_autotest_keys)
355

356

357
def convert_node_id_to_pytest_autotest_keys(node_id: str) -> str:
×
358
    test_path_parts = __get_test_path_parts_by_node_id(node_id)
×
359
    directories_in_project = __get_directories_in_project_by_test_path_parts(test_path_parts)
×
UNCOV
360
    test_node_parts_from_module = __get_test_node_parts_from_module_by_test_path_parts(test_path_parts)
×
361

UNCOV
362
    return __join_test_node_parts_to_pytest_autotest_keys(directories_in_project + test_node_parts_from_module)
×
363

364

365
def __get_test_path_parts_by_node_id(node_id: str):
×
UNCOV
366
    return node_id.split('/')
×
367

368

369
def __get_directories_in_project_by_test_path_parts(test_path_parts: List[str]):
×
UNCOV
370
    return test_path_parts[:-1]
×
371

372

373
def __get_test_node_parts_from_module_by_test_path_parts(test_path_parts: List[str]):
×
UNCOV
374
    test_node_from_module = test_path_parts[-1]
×
375

UNCOV
376
    return test_node_from_module.split('::')
×
377

378

379
def __join_test_node_parts_to_pytest_autotest_keys(test_node_parts: List[str]):
×
UNCOV
380
    return ' and '.join(test_node_parts)
×
381

382

UNCOV
383
def fixtures_containers_to_test_results_with_all_fixture_step_results(
×
384
        fixtures_containers: dict,
385
        test_result_ids: dict) -> List[TestResultWithAllFixtureStepResults]:
UNCOV
386
    test_results_with_all_fixture_step_results = []
×
387

388
    for node_id, test_result_id in test_result_ids.items():
×
UNCOV
389
        test_result_with_all_fixture_step_results = TestResultWithAllFixtureStepResults(test_result_id)
×
390

391
        for uuid, fixtures_container in fixtures_containers.items():
×
392
            if node_id in fixtures_container.node_ids:
×
393
                if fixtures_container.befores:
×
UNCOV
394
                    test_result_with_all_fixture_step_results.set_setup_results(fixtures_container.befores[0].steps)
×
395

396
                if fixtures_container.afters:
×
UNCOV
397
                    test_result_with_all_fixture_step_results.set_teardown_results(fixtures_container.afters[0].steps)
×
398

UNCOV
399
        test_results_with_all_fixture_step_results.append(test_result_with_all_fixture_step_results)
×
400

UNCOV
401
    return test_results_with_all_fixture_step_results
×
402

403

404
def get_status(exception):
×
405
    if exception:
×
406
        if isinstance(exception, pytest.skip.Exception):
×
407
            return "Skipped"
×
UNCOV
408
        return "Failed"
×
409
    else:
UNCOV
410
        return "Passed"
×
411

412

413
def get_outcome_status(outcome):
×
414
    _, exception, _ = outcome.excinfo or (None, None, None)
×
UNCOV
415
    return get_status(exception)
×
416

417

418
def get_traceback(exc_traceback):
×
UNCOV
419
    return ''.join(traceback.format_tb(exc_traceback)) if exc_traceback else None
×
420

421

422
def get_message(etype, value):
×
UNCOV
423
    return '\n'.join(format_exception_only(etype, value)) if etype or value else None
×
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