17799076352

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

Build # 17799076352

Build Type

push

github

Committed by

web-flow

Commit Message

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

* fix: TMS-35114: add threads for creating and updating autotests with the "importRealtime = false" parameter.

* fix: fix the names of the methods for api_client and converter.

* chore: code styles and client version

---------

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

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

34.57

/testit-python-commons/src/testit_python_commons/client/helpers/threads_manager.py

from testit_api_client.models import (
    AutoTestPostModel,
    AutoTestPutModel,
    AutoTestResultsForTestRunModel,
)

from testit_python_commons.client.models import (
    ThreadForCreateAndResult,
    ThreadsForCreateAndResult,
    ThreadForUpdateAndResult,
    ThreadsForUpdateAndResult
)
from testit_python_commons.services.logger import adapter_logger
from typing import Dict, List


class ThreadsManager:
    def __init__(self):
        self.__threads_for_create: List[Dict[str, AutoTestPostModel]] = []
        self.__threads_for_update: List[Dict[str, AutoTestPutModel]] = []
        self.__threads_results_for_created_autotests: List[List[AutoTestResultsForTestRunModel]] = []
        self.__threads_results_for_updated_autotests: List[List[AutoTestResultsForTestRunModel]] = []
        self.__threads_for_autotest_links_to_wi_for_update: List[Dict[str, List[str]]] = []

    @adapter_logger
    def get_thread_for_create_and_result(self, external_id: str) -> ThreadForCreateAndResult:
        thread_for_create = self.__get_thread_for_create(external_id)
        thread_index = self.__threads_for_create.index(thread_for_create)
        thread_results_for_created_autotests = self.__get_thread_results_for_created_autotests(thread_index)

        return ThreadForCreateAndResult(thread_for_create, thread_results_for_created_autotests)

    @adapter_logger
    def get_all_threads_for_create_and_result(self) -> ThreadsForCreateAndResult:
        return ThreadsForCreateAndResult(self.__threads_for_create, self.__threads_results_for_created_autotests)

    @adapter_logger
    def get_thread_for_update_and_result(self, external_id: str) -> ThreadForUpdateAndResult:
        thread_for_update = self.__get_thread_for_update(external_id)
        thread_index = self.__threads_for_update.index(thread_for_update)
        thread_results_for_updated_autotests = self.__get_thread_results_for_updated_autotests(thread_index)
        thread_for_autotest_links_to_wi_for_update = self.__get_thread_for_autotest_links_to_wi_for_update(thread_index)

        return ThreadForUpdateAndResult(
            thread_for_update,
            thread_results_for_updated_autotests,
            thread_for_autotest_links_to_wi_for_update
        )

    @adapter_logger
    def get_all_threads_for_update_and_result(self) -> ThreadsForUpdateAndResult:
        return ThreadsForUpdateAndResult(
            self.__threads_for_update,
            self.__threads_results_for_updated_autotests,
            self.__threads_for_autotest_links_to_wi_for_update
        )

    @adapter_logger
    def delete_thread_for_create_and_result(self, thread_for_create_and_result: ThreadForCreateAndResult):
        thread_for_create = thread_for_create_and_result.get_thread_for_create()
        thread_results_for_created_autotests = thread_for_create_and_result.get_thread_results_for_created_autotests()

        self.__threads_for_create.remove(thread_for_create)
        self.__threads_results_for_created_autotests.remove(thread_results_for_created_autotests)

    @adapter_logger
    def delete_thread_for_update_and_result(self, thread_for_update_and_result: ThreadForUpdateAndResult):
        thread_for_update = thread_for_update_and_result.get_thread_for_update()
        thread_results_for_updated_autotests = thread_for_update_and_result.get_thread_results_for_updated_autotests()
        thread_for_autotest_links_to_wi_for_update = thread_for_update_and_result\
            .get_thread_for_autotest_links_to_wi_for_update()

        self.__threads_for_update.remove(thread_for_update)
        self.__threads_results_for_updated_autotests.remove(thread_results_for_updated_autotests)
        self.__threads_for_autotest_links_to_wi_for_update.remove(thread_for_autotest_links_to_wi_for_update)

    @adapter_logger
    def __get_thread_for_create(self, external_id: str) -> Dict[str, AutoTestPostModel]:
        for thread in self.__threads_for_create:
            if external_id not in thread.keys():
                return thread

        new_thread: Dict[str, AutoTestPostModel] = {}

        self.__threads_for_create.append(new_thread)

        return new_thread

    @adapter_logger
    def __get_thread_for_update(self, external_id: str) -> Dict[str, AutoTestPutModel]:
        for thread in self.__threads_for_update:
            if external_id not in thread.keys():
                return thread

        new_thread: Dict[str, AutoTestPutModel] = {}

        self.__threads_for_update.append(new_thread)

        return new_thread

    @adapter_logger
    def __get_thread_results_for_created_autotests(
            self, thread_index: int) -> List[AutoTestResultsForTestRunModel]:
        if 0 <= thread_index < len(self.__threads_results_for_created_autotests):
            return self.__threads_results_for_created_autotests[thread_index]

        new_thread_results_for_created_autotests: List[AutoTestResultsForTestRunModel] = []

        self.__threads_results_for_created_autotests.append(new_thread_results_for_created_autotests)

        return new_thread_results_for_created_autotests

    @adapter_logger
    def __get_thread_results_for_updated_autotests(
            self, thread_index: int) -> List[AutoTestResultsForTestRunModel]:
        if 0 <= thread_index < len(self.__threads_results_for_updated_autotests):
            return self.__threads_results_for_updated_autotests[thread_index]

        new_thread_results_for_updated_autotests: List[AutoTestResultsForTestRunModel] = []

        self.__threads_results_for_updated_autotests.append(new_thread_results_for_updated_autotests)

        return new_thread_results_for_updated_autotests

    @adapter_logger
    def __get_thread_for_autotest_links_to_wi_for_update(
            self, thread_index: int) -> Dict[str, List[str]]:
        if 0 <= thread_index < len(self.__threads_for_autotest_links_to_wi_for_update):
            return self.__threads_for_autotest_links_to_wi_for_update[thread_index]

        new_thread_for_autotest_links_to_wi_for_update: Dict[str, List[str]] = {}

        self.__threads_for_autotest_links_to_wi_for_update.append(new_thread_for_autotest_links_to_wi_for_update)

        return new_thread_for_autotest_links_to_wi_for_update

1	from testit_api_client.models import (	2✔
2	AutoTestPostModel,
3	AutoTestPutModel,
4	AutoTestResultsForTestRunModel,
5	)
6
7	from testit_python_commons.client.models import (	2✔
8	ThreadForCreateAndResult,
9	ThreadsForCreateAndResult,
10	ThreadForUpdateAndResult,
11	ThreadsForUpdateAndResult
12	)
13	from testit_python_commons.services.logger import adapter_logger	2✔
14	from typing import Dict, List	2✔
15
16
17	class ThreadsManager:	2✔
18	def __init__(self):	2✔
NEW 19	self.__threads_for_create: List[Dict[str, AutoTestPostModel]] = []	×
NEW 20	self.__threads_for_update: List[Dict[str, AutoTestPutModel]] = []	×
NEW 21	self.__threads_results_for_created_autotests: List[List[AutoTestResultsForTestRunModel]] = []	×
NEW 22	self.__threads_results_for_updated_autotests: List[List[AutoTestResultsForTestRunModel]] = []	×
NEW 23	self.__threads_for_autotest_links_to_wi_for_update: List[Dict[str, List[str]]] = []	×
24
25	@adapter_logger	2✔
26	def get_thread_for_create_and_result(self, external_id: str) -> ThreadForCreateAndResult:	2✔
NEW 27	thread_for_create = self.__get_thread_for_create(external_id)	×
NEW 28	thread_index = self.__threads_for_create.index(thread_for_create)	×
NEW 29	thread_results_for_created_autotests = self.__get_thread_results_for_created_autotests(thread_index)	×
30
NEW 31	return ThreadForCreateAndResult(thread_for_create, thread_results_for_created_autotests)	×
32
33	@adapter_logger	2✔
34	def get_all_threads_for_create_and_result(self) -> ThreadsForCreateAndResult:	2✔
NEW 35	return ThreadsForCreateAndResult(self.__threads_for_create, self.__threads_results_for_created_autotests)	×
36
37	@adapter_logger	2✔
38	def get_thread_for_update_and_result(self, external_id: str) -> ThreadForUpdateAndResult:	2✔
NEW 39	thread_for_update = self.__get_thread_for_update(external_id)	×
NEW 40	thread_index = self.__threads_for_update.index(thread_for_update)	×
NEW 41	thread_results_for_updated_autotests = self.__get_thread_results_for_updated_autotests(thread_index)	×
NEW 42	thread_for_autotest_links_to_wi_for_update = self.__get_thread_for_autotest_links_to_wi_for_update(thread_index)	×
43
NEW 44	return ThreadForUpdateAndResult(	×
45	thread_for_update,
46	thread_results_for_updated_autotests,
47	thread_for_autotest_links_to_wi_for_update
48	)
49
50	@adapter_logger	2✔
51	def get_all_threads_for_update_and_result(self) -> ThreadsForUpdateAndResult:	2✔
NEW 52	return ThreadsForUpdateAndResult(	×
53	self.__threads_for_update,
54	self.__threads_results_for_updated_autotests,
55	self.__threads_for_autotest_links_to_wi_for_update
56	)
57
58	@adapter_logger	2✔
59	def delete_thread_for_create_and_result(self, thread_for_create_and_result: ThreadForCreateAndResult):	2✔
NEW 60	thread_for_create = thread_for_create_and_result.get_thread_for_create()	×
NEW 61	thread_results_for_created_autotests = thread_for_create_and_result.get_thread_results_for_created_autotests()	×
62
NEW 63	self.__threads_for_create.remove(thread_for_create)	×
NEW 64	self.__threads_results_for_created_autotests.remove(thread_results_for_created_autotests)	×
65
66	@adapter_logger	2✔
67	def delete_thread_for_update_and_result(self, thread_for_update_and_result: ThreadForUpdateAndResult):	2✔
NEW 68	thread_for_update = thread_for_update_and_result.get_thread_for_update()	×
NEW 69	thread_results_for_updated_autotests = thread_for_update_and_result.get_thread_results_for_updated_autotests()	×
NEW 70	thread_for_autotest_links_to_wi_for_update = thread_for_update_and_result\	×
71	.get_thread_for_autotest_links_to_wi_for_update()
72
NEW 73	self.__threads_for_update.remove(thread_for_update)	×
NEW 74	self.__threads_results_for_updated_autotests.remove(thread_results_for_updated_autotests)	×
NEW 75	self.__threads_for_autotest_links_to_wi_for_update.remove(thread_for_autotest_links_to_wi_for_update)	×
76
77	@adapter_logger	2✔
78	def __get_thread_for_create(self, external_id: str) -> Dict[str, AutoTestPostModel]:	2✔
NEW 79	for thread in self.__threads_for_create:	×
NEW 80	if external_id not in thread.keys():	×
NEW 81	return thread	×
82
NEW 83	new_thread: Dict[str, AutoTestPostModel] = {}	×
84
NEW 85	self.__threads_for_create.append(new_thread)	×
86
NEW 87	return new_thread	×
88
89	@adapter_logger	2✔
90	def __get_thread_for_update(self, external_id: str) -> Dict[str, AutoTestPutModel]:	2✔
NEW 91	for thread in self.__threads_for_update:	×
NEW 92	if external_id not in thread.keys():	×
NEW 93	return thread	×
94
NEW 95	new_thread: Dict[str, AutoTestPutModel] = {}	×
96
NEW 97	self.__threads_for_update.append(new_thread)	×
98
NEW 99	return new_thread	×
100
101	@adapter_logger	2✔
102	def __get_thread_results_for_created_autotests(	2✔
103	self, thread_index: int) -> List[AutoTestResultsForTestRunModel]:
NEW 104	if 0 <= thread_index < len(self.__threads_results_for_created_autotests):	×
NEW 105	return self.__threads_results_for_created_autotests[thread_index]	×
106
NEW 107	new_thread_results_for_created_autotests: List[AutoTestResultsForTestRunModel] = []	×
108
NEW 109	self.__threads_results_for_created_autotests.append(new_thread_results_for_created_autotests)	×
110
NEW 111	return new_thread_results_for_created_autotests	×
112
113	@adapter_logger	2✔
114	def __get_thread_results_for_updated_autotests(	2✔
115	self, thread_index: int) -> List[AutoTestResultsForTestRunModel]:
NEW 116	if 0 <= thread_index < len(self.__threads_results_for_updated_autotests):	×
NEW 117	return self.__threads_results_for_updated_autotests[thread_index]	×
118
NEW 119	new_thread_results_for_updated_autotests: List[AutoTestResultsForTestRunModel] = []	×
120
NEW 121	self.__threads_results_for_updated_autotests.append(new_thread_results_for_updated_autotests)	×
122
NEW 123	return new_thread_results_for_updated_autotests	×
124
125	@adapter_logger	2✔
126	def __get_thread_for_autotest_links_to_wi_for_update(	2✔
127	self, thread_index: int) -> Dict[str, List[str]]:
NEW 128	if 0 <= thread_index < len(self.__threads_for_autotest_links_to_wi_for_update):	×
NEW 129	return self.__threads_for_autotest_links_to_wi_for_update[thread_index]	×
130
NEW 131	new_thread_for_autotest_links_to_wi_for_update: Dict[str, List[str]] = {}	×
132
NEW 133	self.__threads_for_autotest_links_to_wi_for_update.append(new_thread_for_autotest_links_to_wi_for_update)	×
134
NEW 135	return new_thread_for_autotest_links_to_wi_for_update	×

testit-tms / adapters-python / 17799076352

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