17798992833

Committed 17 Sep 2025 01:19PM UTC coverage: 36.943% (-6.5%) from 43.443%

Build # 17798992833

Build Type

Pull #202

github

Committed by

web-flow

Commit Message

Merge bfd5ede57 into b59427724

Pull Request Pull Request #202: fix: TMS-35114: add threads for creating and updating autotests with …

Run Details

102 of 224 new or added lines in 15 files covered. (45.54%)

1 existing line in 1 file now uncovered.

1293 of 3500 relevant lines covered (36.94%)

0.74 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

import hashlib
import logging
import re
import traceback
from typing import List

import pytest

from traceback import format_exception_only

from testit_python_commons.models.link import Link
from testit_python_commons.models.test_result import TestResult
from testit_python_commons.models.test_result_with_all_fixture_step_results_model import TestResultWithAllFixtureStepResults

from testit_adapter_pytest.models.executable_test import ExecutableTest


__ARRAY_TYPES = (frozenset, list, set, tuple,)


def form_test(item) -> ExecutableTest:
    executable_test = ExecutableTest(
        external_id=__get_external_id_from(item),
        name=__get_display_name_from(item),
        duration=0,
        parameters=__get_parameters_from(item),
        properties=__get_properties_from(item),
        namespace=__get_namespace_from(item),
        classname=__get_class_name_from(item),
        title=__get_title_from(item),
        description=__get_description_from(item),
        links=__get_links_from(item),
        labels=__get_labels_from(item),
        work_item_ids=__get_work_item_ids_from(item),
        node_id=item.nodeid
    )

    if item.own_markers:
        for mark in item.own_markers:
            if mark.name == 'skip' or mark.name == 'skipif':
                executable_test.outcome = 'Skipped'
                if mark.args:
                    executable_test.message = mark.args[0]
                if mark.kwargs and 'reason' in mark.kwargs:
                    executable_test.message = mark.kwargs['reason']
            if mark.name == 'xfail':
                if mark.args:
                    executable_test.message = mark.args[0]
                if mark.kwargs and 'reason' in mark.kwargs:
                    executable_test.message = mark.kwargs['reason']

    return executable_test


def __get_display_name_from(item):
    display_name = __search_attribute(item, 'test_displayname')

    if not display_name:
        return item.function.__doc__ if \
            item.function.__doc__ else item.function.__name__

    return collect_parameters_in_string_attribute(display_name, get_all_parameters(item))


def __get_external_id_from(item):
    external_id = __search_attribute(item, 'test_external_id')

    if not external_id:
        return get_hash(item.parent.nodeid + item.function.__name__)

    return collect_parameters_in_string_attribute(external_id, get_all_parameters(item))


def __get_title_from(item):
    title = __search_attribute(item, 'test_title')

    if not title:
        return None

    return collect_parameters_in_string_attribute(title, get_all_parameters(item))


def __get_description_from(item):
    description = __search_attribute(item, 'test_description')

    if not description:
        return None

    return collect_parameters_in_string_attribute(description, get_all_parameters(item))


def __get_namespace_from(item):
    namespace = __search_attribute(item, 'test_namespace')

    if not namespace:
        return item.function.__module__

    return collect_parameters_in_string_attribute(namespace, get_all_parameters(item))


def __get_class_name_from(item):
    class_name = __search_attribute(item, 'test_classname')

    if not class_name:
        i = item.function.__qualname__.find('.')

        if i != -1:
            return item.function.__qualname__[:i]

        return None

    return collect_parameters_in_string_attribute(class_name, get_all_parameters(item))


def __get_links_from(item):
    links = __search_attribute(item, 'test_links')

    if not links:
        return []

    return __set_parameters_to_links(links, get_all_parameters(item))


def __get_parameters_from(item):
    test_parameters = {}

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

    return test_parameters


def __get_properties_from(item):
    if hasattr(item, 'test_properties'):
        return item.test_properties
    return None


def __set_parameters_to_links(links, all_parameters):
    if not all_parameters:
        return links

    links_with_parameters = []

    for link in links:
        links_with_parameters.append(
            Link()
            .set_url(
                collect_parameters_in_string_attribute(
                    link.get_url(),
                    all_parameters))
            .set_title(
                collect_parameters_in_string_attribute(
                    link.get_title(),
                    all_parameters) if link.get_title() else None)
            .set_link_type(
                collect_parameters_in_string_attribute(
                    link.get_link_type(),
                    all_parameters) if link.get_link_type() else None)
            .set_description(
                collect_parameters_in_string_attribute(
                    link.get_description(),
                    all_parameters) if link.get_description() else None))

    return links_with_parameters


def __get_labels_from(item):
    test_labels = __search_attribute(item, 'test_labels')

    if not test_labels:
        return []

    labels = []

    for label in test_labels:
        result = collect_parameters_in_mass_attribute(
            label,
            get_all_parameters(item))

        if isinstance(result, __ARRAY_TYPES):
            for l in result:
                labels.append({
                    'name': str(l)
                })
        else:
            labels.append({
                'name': str(result)
            })

    return labels


def __get_work_item_ids_from(item):
    test_workitem_ids = __search_attribute(item, 'test_workitems_id')

    if not test_workitem_ids:
        return []

    workitem_ids = []

    for workitem_id in test_workitem_ids:
        result = collect_parameters_in_mass_attribute(
            workitem_id,
            get_all_parameters(item))

        if isinstance(result, __ARRAY_TYPES):
            workitem_ids += list(map(str, result))
        else:
            workitem_ids.append(str(result))

    return workitem_ids


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

    if len(param_keys) > 0:
        for param_key in param_keys:
            parameter = get_parameter(param_key, all_parameters)

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

    return attribute


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

    if len(param_keys) == 1:
        parameter = get_parameter(param_keys[0], all_parameters)

        if parameter is not None:
            return parameter

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

    return attribute


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

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

        return

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

            return

        return all_parameters[key_for_parameter]

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

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

    if id_key_in_parameter.isalnum() and id_key_in_parameter in all_parameters[parameter_key].keys():
        return all_parameters[parameter_key][id_key_in_parameter]

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


def get_all_parameters(item):
    params = {}

    if hasattr(item, 'test_properties'):
        params.update(item.test_properties)

    if hasattr(item, 'callspec'):
        params.update(item.callspec.params)

    return params


def __search_attribute(item, attribute):
    if hasattr(item.function, attribute):
        return getattr(item.function, attribute)

    if hasattr(item.cls, attribute):
        return getattr(item.cls, attribute)

    return


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


def convert_executable_test_to_test_result_model(executable_test: ExecutableTest) -> TestResult:
    pytest_autotest_keys = convert_node_id_to_pytest_autotest_keys(executable_test.node_id)

    return TestResult()\
        .set_external_id(executable_test.external_id)\
        .set_autotest_name(executable_test.name)\
        .set_step_results(executable_test.step_results)\
        .set_setup_results(executable_test.setup_step_results)\
        .set_teardown_results(executable_test.teardown_step_results)\
        .set_duration(executable_test.duration)\
        .set_outcome(executable_test.outcome)\
        .set_traces(executable_test.traces)\
        .set_attachments(executable_test.attachments)\
        .set_parameters(executable_test.parameters)\
        .set_properties(executable_test.properties)\
        .set_namespace(executable_test.namespace)\
        .set_classname(executable_test.classname)\
        .set_title(executable_test.title)\
        .set_description(executable_test.description)\
        .set_links(executable_test.links)\
        .set_result_links(executable_test.result_links)\
        .set_labels(executable_test.labels)\
        .set_work_item_ids(executable_test.work_item_ids) \
        .set_message(executable_test.message) \
        .set_external_key(pytest_autotest_keys)


def convert_node_id_to_pytest_autotest_keys(node_id: str) -> str:
    test_path_parts = __get_test_path_parts_by_node_id(node_id)
    directories_in_project = __get_directories_in_project_by_test_path_parts(test_path_parts)
    test_node_parts_from_module = __get_test_node_parts_from_module_by_test_path_parts(test_path_parts)

    return __join_test_node_parts_to_pytest_autotest_keys(directories_in_project + test_node_parts_from_module)


def __get_test_path_parts_by_node_id(node_id: str):
    return node_id.split('/')


def __get_directories_in_project_by_test_path_parts(test_path_parts: List[str]):
    return test_path_parts[:-1]


def __get_test_node_parts_from_module_by_test_path_parts(test_path_parts: List[str]):
    test_node_from_module = test_path_parts[-1]

    return test_node_from_module.split('::')


def __join_test_node_parts_to_pytest_autotest_keys(test_node_parts: List[str]):
    return ' and '.join(test_node_parts)


def fixtures_containers_to_test_results_with_all_fixture_step_results(
        fixtures_containers: dict,
        test_result_ids: dict) -> List[TestResultWithAllFixtureStepResults]:
    test_results_with_all_fixture_step_results = []

    for node_id, test_result_id in test_result_ids.items():
        test_result_with_all_fixture_step_results = TestResultWithAllFixtureStepResults(test_result_id)

        for uuid, fixtures_container in fixtures_containers.items():
            if node_id in fixtures_container.node_ids:
                if fixtures_container.befores:
                    test_result_with_all_fixture_step_results.set_setup_results(fixtures_container.befores[0].steps)

                if fixtures_container.afters:
                    test_result_with_all_fixture_step_results.set_teardown_results(fixtures_container.afters[0].steps)

        test_results_with_all_fixture_step_results.append(test_result_with_all_fixture_step_results)

    return test_results_with_all_fixture_step_results


def get_status(exception):
    if exception:
        if isinstance(exception, pytest.skip.Exception):
            return "Skipped"
        return "Failed"
    else:
        return "Passed"


def get_outcome_status(outcome):
    _, exception, _ = outcome.excinfo or (None, None, None)
    return get_status(exception)


def get_traceback(exc_traceback):
    return ''.join(traceback.format_tb(exc_traceback)) if exc_traceback else None


def get_message(etype, value):
    return '\n'.join(format_exception_only(etype, value)) if etype or value else None

1	import hashlib	×
2	import logging	×
3	import re	×
4	import traceback	×
NEW 5	from typing import List	×
6
UNCOV 7	import pytest	×
8
9	from traceback import format_exception_only	×
10
11	from testit_python_commons.models.link import Link	×
12	from testit_python_commons.models.test_result import TestResult	×
13	from testit_python_commons.models.test_result_with_all_fixture_step_results_model import TestResultWithAllFixtureStepResults	×
14
15	from testit_adapter_pytest.models.executable_test import ExecutableTest	×
16
17
18	__ARRAY_TYPES = (frozenset, list, set, tuple,)	×
19
20
21	def form_test(item) -> ExecutableTest:	×
22	executable_test = ExecutableTest(	×
23	external_id=__get_external_id_from(item),
24	name=__get_display_name_from(item),
25	duration=0,
26	parameters=__get_parameters_from(item),
27	properties=__get_properties_from(item),
28	namespace=__get_namespace_from(item),
29	classname=__get_class_name_from(item),
30	title=__get_title_from(item),
31	description=__get_description_from(item),
32	links=__get_links_from(item),
33	labels=__get_labels_from(item),
34	work_item_ids=__get_work_item_ids_from(item),
35	node_id=item.nodeid
36	)
37
38	if item.own_markers:	×
39	for mark in item.own_markers:	×
40	if mark.name == 'skip' or mark.name == 'skipif':	×
41	executable_test.outcome = 'Skipped'	×
42	if mark.args:	×
43	executable_test.message = mark.args[0]	×
44	if mark.kwargs and 'reason' in mark.kwargs:	×
45	executable_test.message = mark.kwargs['reason']	×
46	if mark.name == 'xfail':	×
47	if mark.args:	×
48	executable_test.message = mark.args[0]	×
49	if mark.kwargs and 'reason' in mark.kwargs:	×
50	executable_test.message = mark.kwargs['reason']	×
51
52	return executable_test	×
53
54
55	def __get_display_name_from(item):	×
56	display_name = __search_attribute(item, 'test_displayname')	×
57
58	if not display_name:	×
59	return item.function.__doc__ if \	×
60	item.function.__doc__ else item.function.__name__
61
62	return collect_parameters_in_string_attribute(display_name, get_all_parameters(item))	×
63
64
65	def __get_external_id_from(item):	×
66	external_id = __search_attribute(item, 'test_external_id')	×
67
68	if not external_id:	×
69	return get_hash(item.parent.nodeid + item.function.__name__)	×
70
71	return collect_parameters_in_string_attribute(external_id, get_all_parameters(item))	×
72
73
74	def __get_title_from(item):	×
75	title = __search_attribute(item, 'test_title')	×
76
77	if not title:	×
78	return None	×
79
80	return collect_parameters_in_string_attribute(title, get_all_parameters(item))	×
81
82
83	def __get_description_from(item):	×
84	description = __search_attribute(item, 'test_description')	×
85
86	if not description:	×
87	return None	×
88
89	return collect_parameters_in_string_attribute(description, get_all_parameters(item))	×
90
91
92	def __get_namespace_from(item):	×
93	namespace = __search_attribute(item, 'test_namespace')	×
94
95	if not namespace:	×
96	return item.function.__module__	×
97
98	return collect_parameters_in_string_attribute(namespace, get_all_parameters(item))	×
99
100
101	def __get_class_name_from(item):	×
102	class_name = __search_attribute(item, 'test_classname')	×
103
104	if not class_name:	×
105	i = item.function.__qualname__.find('.')	×
106
107	if i != -1:	×
108	return item.function.__qualname__[:i]	×
109
110	return None	×
111
112	return collect_parameters_in_string_attribute(class_name, get_all_parameters(item))	×
113
114
115	def __get_links_from(item):	×
116	links = __search_attribute(item, 'test_links')	×
117
118	if not links:	×
119	return []	×
120
121	return __set_parameters_to_links(links, get_all_parameters(item))	×
122
123
124	def __get_parameters_from(item):	×
125	test_parameters = {}	×
126
127	if hasattr(item, 'callspec'):	×
128	for key, parameter in item.callspec.params.items():	×
129	test_parameters[key] = str(parameter)	×
130
131	return test_parameters	×
132
133
134	def __get_properties_from(item):	×
135	if hasattr(item, 'test_properties'):	×
136	return item.test_properties	×
137	return None	×
138
139
140	def __set_parameters_to_links(links, all_parameters):	×
141	if not all_parameters:	×
142	return links	×
143
144	links_with_parameters = []	×
145
146	for link in links:	×
147	links_with_parameters.append(	×
148	Link()
149	.set_url(
150	collect_parameters_in_string_attribute(
151	link.get_url(),
152	all_parameters))
153	.set_title(
154	collect_parameters_in_string_attribute(
155	link.get_title(),
156	all_parameters) if link.get_title() else None)
157	.set_link_type(
158	collect_parameters_in_string_attribute(
159	link.get_link_type(),
160	all_parameters) if link.get_link_type() else None)
161	.set_description(
162	collect_parameters_in_string_attribute(
163	link.get_description(),
164	all_parameters) if link.get_description() else None))
165
166	return links_with_parameters	×
167
168
169	def __get_labels_from(item):	×
170	test_labels = __search_attribute(item, 'test_labels')	×
171
172	if not test_labels:	×
173	return []	×
174
175	labels = []	×
176
177	for label in test_labels:	×
178	result = collect_parameters_in_mass_attribute(	×
179	label,
180	get_all_parameters(item))
181
182	if isinstance(result, __ARRAY_TYPES):	×
183	for l in result:	×
184	labels.append({	×
185	'name': str(l)
186	})
187	else:
188	labels.append({	×
189	'name': str(result)
190	})
191
192	return labels	×
193
194
195	def __get_work_item_ids_from(item):	×
196	test_workitem_ids = __search_attribute(item, 'test_workitems_id')	×
197
198	if not test_workitem_ids:	×
199	return []	×
200
201	workitem_ids = []	×
202
203	for workitem_id in test_workitem_ids:	×
204	result = collect_parameters_in_mass_attribute(	×
205	workitem_id,
206	get_all_parameters(item))
207
208	if isinstance(result, __ARRAY_TYPES):	×
209	workitem_ids += list(map(str, result))	×
210	else:
211	workitem_ids.append(str(result))	×
212
213	return workitem_ids	×
214
215
216	def collect_parameters_in_string_attribute(attribute, all_parameters):	×
217	param_keys = re.findall(r"\{(.*?)\}", attribute)	×
218
219	if len(param_keys) > 0:	×
220	for param_key in param_keys:	×
221	parameter = get_parameter(param_key, all_parameters)	×
222
223	if parameter is not None:	×
224	attribute = attribute.replace("{" + param_key + "}", str(parameter))	×
225
226	return attribute	×
227
228
229	def collect_parameters_in_mass_attribute(attribute, all_parameters):	×
230	param_keys = re.findall(r"\{(.*?)\}", attribute)	×
231
232	if len(param_keys) == 1:	×
233	parameter = get_parameter(param_keys[0], all_parameters)	×
234
235	if parameter is not None:	×
236	return parameter	×
237
238	if len(param_keys) > 1:	×
239	logging.error(f'(For type tuple, list, set) support only one key!')	×
240
241	return attribute	×
242
243
244	def get_parameter(key_for_parameter, all_parameters):	×
245	id_keys_in_parameter = re.findall(r'\[(.*?)\]', key_for_parameter)	×
246
247	if len(id_keys_in_parameter) > 1:	×
248	logging.error("(For type tuple, list, set, dict) support only one level!")	×
249
250	return	×
251
252	if len(id_keys_in_parameter) == 0:	×
253	if key_for_parameter not in all_parameters:	×
254	logging.error(f"Key for parameter {key_for_parameter} not found")	×
255
256	return	×
257
258	return all_parameters[key_for_parameter]	×
259
260	parameter_key = key_for_parameter.replace("[" + id_keys_in_parameter[0] + "]", "")	×
261	id_key_in_parameter = id_keys_in_parameter[0].strip("\'\"")	×
262
263	if id_key_in_parameter.isdigit() and int(id_key_in_parameter) in range(len(all_parameters[parameter_key])):	×
264	return all_parameters[parameter_key][int(id_key_in_parameter)]	×
265
266	if id_key_in_parameter.isalnum() and id_key_in_parameter in all_parameters[parameter_key].keys():	×
267	return all_parameters[parameter_key][id_key_in_parameter]	×
268
269	logging.error(f"Not key: {key_for_parameter} in run parameters or other keys problem")	×
270
271
272	def get_all_parameters(item):	×
273	params = {}	×
274
275	if hasattr(item, 'test_properties'):	×
276	params.update(item.test_properties)	×
277
278	if hasattr(item, 'callspec'):	×
279	params.update(item.callspec.params)	×
280
281	return params	×
282
283
284	def __search_attribute(item, attribute):	×
285	if hasattr(item.function, attribute):	×
286	return getattr(item.function, attribute)	×
287
288	if hasattr(item.cls, attribute):	×
289	return getattr(item.cls, attribute)	×
290
291	return	×
292
293
294	def get_hash(value: str):	×
295	md = hashlib.sha256(bytes(value, encoding='utf-8'))	×
296	return md.hexdigest()	×
297
298
299	def convert_executable_test_to_test_result_model(executable_test: ExecutableTest) -> TestResult:	×
300	pytest_autotest_keys = convert_node_id_to_pytest_autotest_keys(executable_test.node_id)	×
301
302	return TestResult()\	×
303	.set_external_id(executable_test.external_id)\
304	.set_autotest_name(executable_test.name)\
305	.set_step_results(executable_test.step_results)\
306	.set_setup_results(executable_test.setup_step_results)\
307	.set_teardown_results(executable_test.teardown_step_results)\
308	.set_duration(executable_test.duration)\
309	.set_outcome(executable_test.outcome)\
310	.set_traces(executable_test.traces)\
311	.set_attachments(executable_test.attachments)\
312	.set_parameters(executable_test.parameters)\
313	.set_properties(executable_test.properties)\
314	.set_namespace(executable_test.namespace)\
315	.set_classname(executable_test.classname)\
316	.set_title(executable_test.title)\
317	.set_description(executable_test.description)\
318	.set_links(executable_test.links)\
319	.set_result_links(executable_test.result_links)\
320	.set_labels(executable_test.labels)\
321	.set_work_item_ids(executable_test.work_item_ids) \
322	.set_message(executable_test.message) \
323	.set_external_key(pytest_autotest_keys)
324
325
326	def convert_node_id_to_pytest_autotest_keys(node_id: str) -> str:	×
327	test_path_parts = __get_test_path_parts_by_node_id(node_id)	×
328	directories_in_project = __get_directories_in_project_by_test_path_parts(test_path_parts)	×
329	test_node_parts_from_module = __get_test_node_parts_from_module_by_test_path_parts(test_path_parts)	×
330
331	return __join_test_node_parts_to_pytest_autotest_keys(directories_in_project + test_node_parts_from_module)	×
332
333
334	def __get_test_path_parts_by_node_id(node_id: str):	×
335	return node_id.split('/')	×
336
337
NEW 338	def __get_directories_in_project_by_test_path_parts(test_path_parts: List[str]):	×
339	return test_path_parts[:-1]	×
340
341
NEW 342	def __get_test_node_parts_from_module_by_test_path_parts(test_path_parts: List[str]):	×
343	test_node_from_module = test_path_parts[-1]	×
344
345	return test_node_from_module.split('::')	×
346
347
NEW 348	def __join_test_node_parts_to_pytest_autotest_keys(test_node_parts: List[str]):	×
349	return ' and '.join(test_node_parts)	×
350
351
352	def fixtures_containers_to_test_results_with_all_fixture_step_results(	×
353	fixtures_containers: dict,
354	test_result_ids: dict) -> List[TestResultWithAllFixtureStepResults]:
355	test_results_with_all_fixture_step_results = []	×
356
357	for node_id, test_result_id in test_result_ids.items():	×
358	test_result_with_all_fixture_step_results = TestResultWithAllFixtureStepResults(test_result_id)	×
359
360	for uuid, fixtures_container in fixtures_containers.items():	×
361	if node_id in fixtures_container.node_ids:	×
362	if fixtures_container.befores:	×
363	test_result_with_all_fixture_step_results.set_setup_results(fixtures_container.befores[0].steps)	×
364
365	if fixtures_container.afters:	×
366	test_result_with_all_fixture_step_results.set_teardown_results(fixtures_container.afters[0].steps)	×
367
368	test_results_with_all_fixture_step_results.append(test_result_with_all_fixture_step_results)	×
369
370	return test_results_with_all_fixture_step_results	×
371
372
373	def get_status(exception):	×
374	if exception:	×
375	if isinstance(exception, pytest.skip.Exception):	×
376	return "Skipped"	×
377	return "Failed"	×
378	else:
379	return "Passed"	×
380
381
382	def get_outcome_status(outcome):	×
383	_, exception, _ = outcome.excinfo or (None, None, None)	×
384	return get_status(exception)	×
385
386
387	def get_traceback(exc_traceback):	×
388	return ''.join(traceback.format_tb(exc_traceback)) if exc_traceback else None	×
389
390
391	def get_message(etype, value):	×
392	return '\n'.join(format_exception_only(etype, value)) if etype or value else None	×

testit-tms / adapters-python / 17798992833

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