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

testit-tms / adapters-python / 23762390641

30 Mar 2026 07:01PM UTC coverage: 17.402% (+0.02%) from 17.381%
23762390641

push

github

web-flow
feat: support custom statuses. (#246)

* feat: support custom statuses.

* feat: support custom statuses.

* fix: set status for behave.

* fix: update sync-storage

* chore: update sync-storage version

* chore: update runner logic to type only

* chore: update version

* chore: update ci

* chore: update sync-storage version

* fix messages from failed steps for robotframework.

* fix: set status for behave.

* fix: small sync-storage fixes

* chore: update ci

---------

Co-authored-by: pavel.butuzov <pavel.butuzov@testit.software>
Co-authored-by: Dmitry Ermakovich <dmitry.ermakovich@testit.software>

7 of 54 new or added lines in 10 files covered. (12.96%)

2 existing lines in 2 files now uncovered.

284 of 1632 relevant lines covered (17.4%)

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
×
NEW
15
from testit_python_commons.models.outcome_type import OutcomeType
×
16

17
from testit_adapter_pytest.models.executable_test import ExecutableTest
×
18

19

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

22

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

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

44
    return executable_test
×
45

46

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

57
            condition_in_args = marker.args and marker.args[0]
×
58
            condition_in_kwargs = marker.kwargs and 'condition' in marker.kwargs and marker.kwargs['condition']
×
59

60
            if (condition_in_kwargs or condition_in_args) and 'reason' in marker.kwargs:
×
61
                executable_test.message = marker.kwargs['reason']
×
62

63
    return executable_test
×
64

65

66
def __get_display_name_from(item):
×
67
    display_name = __search_attribute(item, 'test_displayname')
×
68

69
    if not display_name:
×
70
        return item.function.__doc__ if \
×
71
            item.function.__doc__ else item.function.__name__
72

73
    return collect_parameters_in_string_attribute(display_name, get_all_parameters(item))
×
74

75

76
def __get_external_id_from(item):
×
77
    external_id = __search_attribute(item, 'test_external_id')
×
78

79
    if not external_id:
×
80
        return get_hash(item.parent.nodeid + item.function.__name__)
×
81

82
    return collect_parameters_in_string_attribute(external_id, get_all_parameters(item))
×
83

84

85
def __get_title_from(item):
×
86
    title = __search_attribute(item, 'test_title')
×
87

88
    if not title:
×
89
        return None
×
90

91
    return collect_parameters_in_string_attribute(title, get_all_parameters(item))
×
92

93

94
def __get_description_from(item):
×
95
    description = __search_attribute(item, 'test_description')
×
96

97
    if not description:
×
98
        return None
×
99

100
    return collect_parameters_in_string_attribute(description, get_all_parameters(item))
×
101

102

103
def __get_namespace_from(item):
×
104
    namespace = __search_attribute(item, 'test_namespace')
×
105

106
    if not namespace:
×
107
        return item.function.__module__
×
108

109
    return collect_parameters_in_string_attribute(namespace, get_all_parameters(item))
×
110

111

112
def __get_class_name_from(item):
×
113
    class_name = __search_attribute(item, 'test_classname')
×
114

115
    if not class_name:
×
116
        i = item.function.__qualname__.find('.')
×
117

118
        if i != -1:
×
119
            return item.function.__qualname__[:i]
×
120

121
        return None
×
122

123
    return collect_parameters_in_string_attribute(class_name, get_all_parameters(item))
×
124

125

126
def __get_links_from(item):
×
127
    links = __search_attribute(item, 'test_links')
×
128

129
    if not links:
×
130
        return []
×
131

132
    return __set_parameters_to_links(links, get_all_parameters(item))
×
133

134

135
def __get_parameters_from(item):
×
136
    test_parameters = {}
×
137

138
    if hasattr(item, 'callspec'):
×
139
        for key, parameter in item.callspec.params.items():
×
140
            test_parameters[key] = str(parameter)
×
141

142
    return test_parameters
×
143

144

145
def __get_properties_from(item):
×
146
    if hasattr(item, 'test_properties'):
×
147
        return item.test_properties
×
148
    return None
×
149

150

151
def __set_parameters_to_links(links, all_parameters):
×
152
    if not all_parameters:
×
153
        return links
×
154

155
    links_with_parameters = []
×
156

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

177
    return links_with_parameters
×
178

179

180
def __get_labels_from(item) -> List[dict]:
×
181
    test_labels = __search_attribute(item, 'test_labels')
×
182

183
    if not test_labels:
×
184
        return []
×
185

186
    labels = []
×
187

188
    for label in test_labels:
×
189
        result = collect_parameters_in_mass_attribute(
×
190
            label,
191
            get_all_parameters(item))
192

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

203
    return labels
×
204

205

206
def __get_tags_from(item) -> List[str]:
×
207
    test_tags = __search_attribute(item, 'test_tags')
×
208

209
    if not test_tags:
×
210
        return []
×
211

212
    tags = []
×
213

214
    for tag in test_tags:
×
215
        result = collect_parameters_in_mass_attribute(
×
216
            tag,
217
            get_all_parameters(item))
218

219
        if isinstance(result, __ARRAY_TYPES):
×
220
            for t in result:
×
221
                tags.append(str(t))
×
222
        else:
223
            tags.append(str(result))
×
224

225
    return tags
×
226

227

228
def __get_work_item_ids_from(item):
×
229
    test_workitem_ids = __search_attribute(item, 'test_workitems_id')
×
230

231
    if not test_workitem_ids:
×
232
        return []
×
233

234
    workitem_ids = []
×
235

236
    for workitem_id in test_workitem_ids:
×
237
        result = collect_parameters_in_mass_attribute(
×
238
            workitem_id,
239
            get_all_parameters(item))
240

241
        if isinstance(result, __ARRAY_TYPES):
×
242
            workitem_ids += list(map(str, result))
×
243
        else:
244
            workitem_ids.append(str(result))
×
245

246
    return workitem_ids
×
247

248

249
def collect_parameters_in_string_attribute(attribute, all_parameters):
×
250
    param_keys = re.findall(r"\{(.*?)\}", attribute)
×
251

252
    if len(param_keys) > 0:
×
253
        for param_key in param_keys:
×
254
            parameter = get_parameter(param_key, all_parameters)
×
255

256
            if parameter is not None:
×
257
                attribute = attribute.replace("{" + param_key + "}", str(parameter))
×
258

259
    return attribute
×
260

261

262
def collect_parameters_in_mass_attribute(attribute, all_parameters):
×
263
    param_keys = re.findall(r"\{(.*?)\}", attribute)
×
264

265
    if len(param_keys) == 1:
×
266
        parameter = get_parameter(param_keys[0], all_parameters)
×
267

268
        if parameter is not None:
×
269
            return parameter
×
270

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

274
    return attribute
×
275

276

277
def get_parameter(key_for_parameter, all_parameters):
×
278
    id_keys_in_parameter = re.findall(r'\[(.*?)\]', key_for_parameter)
×
279

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

283
        return
×
284

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

289
            return
×
290

291
        return all_parameters[key_for_parameter]
×
292

293
    parameter_key = key_for_parameter.replace("[" + id_keys_in_parameter[0] + "]", "")
×
294
    id_key_in_parameter = id_keys_in_parameter[0].strip("\'\"")
×
295

296
    if id_key_in_parameter.isdigit() and int(id_key_in_parameter) in range(len(all_parameters[parameter_key])):
×
297
        return all_parameters[parameter_key][int(id_key_in_parameter)]
×
298

299
    if id_key_in_parameter.isalnum() and id_key_in_parameter in all_parameters[parameter_key].keys():
×
300
        return all_parameters[parameter_key][id_key_in_parameter]
×
301

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

304

305
def get_all_parameters(item):
×
306
    params = {}
×
307

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

311
    if hasattr(item, 'callspec'):
×
312
        params.update(item.callspec.params)
×
313

314
    return params
×
315

316

317
def __search_attribute(item, attribute):
×
318
    if hasattr(item.function, attribute):
×
319
        return getattr(item.function, attribute)
×
320

321
    if hasattr(item.cls, attribute):
×
322
        return getattr(item.cls, attribute)
×
323

324
    return
×
325

326

327
def get_hash(value: str):
×
328
    md = hashlib.sha256(bytes(value, encoding='utf-8'))
×
329
    return md.hexdigest()
×
330

331

332
def convert_executable_test_to_test_result_model(executable_test: ExecutableTest) -> TestResult:
×
333
    pytest_autotest_keys = convert_node_id_to_pytest_autotest_keys(executable_test.node_id)
×
334

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

360

361
def convert_node_id_to_pytest_autotest_keys(node_id: str) -> str:
×
362
    test_path_parts = __get_test_path_parts_by_node_id(node_id)
×
363
    directories_in_project = __get_directories_in_project_by_test_path_parts(test_path_parts)
×
364
    test_node_parts_from_module = __get_test_node_parts_from_module_by_test_path_parts(test_path_parts)
×
365

366
    return __join_test_node_parts_to_pytest_autotest_keys(directories_in_project + test_node_parts_from_module)
×
367

368

369
def __get_test_path_parts_by_node_id(node_id: str):
×
370
    return node_id.split('/')
×
371

372

373
def __get_directories_in_project_by_test_path_parts(test_path_parts: List[str]):
×
374
    return test_path_parts[:-1]
×
375

376

377
def __get_test_node_parts_from_module_by_test_path_parts(test_path_parts: List[str]):
×
378
    test_node_from_module = test_path_parts[-1]
×
379

380
    return test_node_from_module.split('::')
×
381

382

383
def __join_test_node_parts_to_pytest_autotest_keys(test_node_parts: List[str]):
×
384
    return ' and '.join(test_node_parts)
×
385

386

387
def fixtures_containers_to_test_results_with_all_fixture_step_results(
×
388
        fixtures_containers: dict,
389
        test_result_ids: dict) -> List[TestResultWithAllFixtureStepResults]:
390
    test_results_with_all_fixture_step_results = []
×
391

392
    for node_id, test_result_id in test_result_ids.items():
×
393
        test_result_with_all_fixture_step_results = TestResultWithAllFixtureStepResults(test_result_id)
×
394

395
        for uuid, fixtures_container in fixtures_containers.items():
×
396
            if node_id in fixtures_container.node_ids:
×
397
                if fixtures_container.befores:
×
398
                    test_result_with_all_fixture_step_results.set_setup_results(fixtures_container.befores[0].steps)
×
399

400
                if fixtures_container.afters:
×
401
                    test_result_with_all_fixture_step_results.set_teardown_results(fixtures_container.afters[0].steps)
×
402

403
        test_results_with_all_fixture_step_results.append(test_result_with_all_fixture_step_results)
×
404

405
    return test_results_with_all_fixture_step_results
×
406

407

408
def get_status(exception):
×
409
    if exception:
×
410
        if isinstance(exception, pytest.skip.Exception):
×
NEW
411
            return OutcomeType.SKIPPED
×
NEW
412
        return OutcomeType.FAILED
×
413
    else:
NEW
414
        return OutcomeType.PASSED
×
415

416

417
def get_outcome_status(outcome):
×
418
    _, exception, _ = outcome.excinfo or (None, None, None)
×
419
    return get_status(exception)
×
420

421

422
def get_traceback(exc_traceback):
×
423
    return ''.join(traceback.format_tb(exc_traceback)) if exc_traceback else None
×
424

425

426
def get_message(etype, value):
×
427
    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