17801911010

Committed 17 Sep 2025 03:01PM UTC coverage: 37.068% (+0.1%) from 36.943%

Build # 17801911010

Build Type

push

github

Committed by

web-flow

Commit Message

Fix/tms 35114 (#203)

* 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

* fix: fix creating autotests by threads with the "importRealtime = false" parameter.

---------

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

Coverage Stats

11 of 57 new or added lines in 4 files covered. (19.3%)

1287 of 3472 relevant lines covered (37.07%)

0.74 hits per line

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

44.0

/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.__thread_for_create: Dict[str, AutoTestPostModel] = {}
        self.__thread_for_update: Dict[str, AutoTestPutModel] = {}
        self.__threads_results_for_created_autotests: List[Dict[str, AutoTestResultsForTestRunModel]] = []
        self.__threads_results_for_updated_autotests: List[Dict[str, AutoTestResultsForTestRunModel]] = []
        self.__thread_for_autotest_links_to_wi_for_update: Dict[str, List[str]] = {}

    @adapter_logger
    def get_thread_for_create_and_result(self, external_id: str) -> ThreadForCreateAndResult:
        thread_results_for_created_autotests = self.__get_thread_results_for_created_autotests(external_id)

        return ThreadForCreateAndResult(self.__thread_for_create, thread_results_for_created_autotests)

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

    @adapter_logger
    def get_thread_for_update_and_result(self, external_id: str) -> ThreadForUpdateAndResult:
        thread_results_for_updated_autotests = self.__get_thread_results_for_updated_autotests(external_id)

        return ThreadForUpdateAndResult(
            self.__thread_for_update,
            thread_results_for_updated_autotests,
            self.__thread_for_autotest_links_to_wi_for_update
        )

    @adapter_logger
    def get_all_threads_for_update_and_result(self) -> ThreadsForUpdateAndResult:
        return ThreadsForUpdateAndResult(
            self.__thread_for_update,
            self.__threads_results_for_updated_autotests,
            self.__thread_for_autotest_links_to_wi_for_update
        )

    @adapter_logger
    def delete_threads_for_create_and_result(self):
        self.__thread_for_create.clear()
        self.__threads_results_for_created_autotests.clear()

    @adapter_logger
    def delete_threads_for_update_and_result(self):
        self.__thread_for_update.clear()
        self.__threads_results_for_updated_autotests.clear()
        self.__thread_for_autotest_links_to_wi_for_update.clear()

    @adapter_logger
    def __get_thread_results_for_created_autotests(self, external_id: str) -> Dict[str, AutoTestResultsForTestRunModel]:
        for thread in self.__threads_results_for_created_autotests:
            if external_id not in thread.keys():
                return thread

        new_thread: Dict[str, AutoTestResultsForTestRunModel] = {}

        self.__threads_results_for_created_autotests.append(new_thread)

        return new_thread

    @adapter_logger
    def __get_thread_results_for_updated_autotests(self, external_id: str) -> Dict[str, AutoTestResultsForTestRunModel]:
        for thread in self.__threads_results_for_updated_autotests:
            if external_id not in thread.keys():
                return thread

        new_thread: Dict[str, AutoTestResultsForTestRunModel] = {}

        self.__threads_results_for_updated_autotests.append(new_thread)

        return new_thread

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.__thread_for_create: Dict[str, AutoTestPostModel] = {}	×
NEW 20	self.__thread_for_update: Dict[str, AutoTestPutModel] = {}	×
NEW 21	self.__threads_results_for_created_autotests: List[Dict[str, AutoTestResultsForTestRunModel]] = []	×
NEW 22	self.__threads_results_for_updated_autotests: List[Dict[str, AutoTestResultsForTestRunModel]] = []	×
NEW 23	self.__thread_for_autotest_links_to_wi_for_update: 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_results_for_created_autotests = self.__get_thread_results_for_created_autotests(external_id)	×
28
NEW 29	return ThreadForCreateAndResult(self.__thread_for_create, thread_results_for_created_autotests)	×
30
31	@adapter_logger	2✔
32	def get_all_threads_for_create_and_result(self) -> ThreadsForCreateAndResult:	2✔
NEW 33	return ThreadsForCreateAndResult(self.__thread_for_create, self.__threads_results_for_created_autotests)	×
34
35	@adapter_logger	2✔
36	def get_thread_for_update_and_result(self, external_id: str) -> ThreadForUpdateAndResult:	2✔
NEW 37	thread_results_for_updated_autotests = self.__get_thread_results_for_updated_autotests(external_id)	×
38
39	return ThreadForUpdateAndResult(	×
40	self.__thread_for_update,
41	thread_results_for_updated_autotests,
42	self.__thread_for_autotest_links_to_wi_for_update
43	)
44
45	@adapter_logger	2✔
46	def get_all_threads_for_update_and_result(self) -> ThreadsForUpdateAndResult:	2✔
47	return ThreadsForUpdateAndResult(	×
48	self.__thread_for_update,
49	self.__threads_results_for_updated_autotests,
50	self.__thread_for_autotest_links_to_wi_for_update
51	)
52
53	@adapter_logger	2✔
54	def delete_threads_for_create_and_result(self):	2✔
NEW 55	self.__thread_for_create.clear()	×
NEW 56	self.__threads_results_for_created_autotests.clear()	×
57
58	@adapter_logger	2✔
59	def delete_threads_for_update_and_result(self):	2✔
NEW 60	self.__thread_for_update.clear()	×
NEW 61	self.__threads_results_for_updated_autotests.clear()	×
NEW 62	self.__thread_for_autotest_links_to_wi_for_update.clear()	×
63
64	@adapter_logger	2✔
65	def __get_thread_results_for_created_autotests(self, external_id: str) -> Dict[str, AutoTestResultsForTestRunModel]:	2✔
NEW 66	for thread in self.__threads_results_for_created_autotests:	×
67	if external_id not in thread.keys():	×
68	return thread	×
69
NEW 70	new_thread: Dict[str, AutoTestResultsForTestRunModel] = {}	×
71
NEW 72	self.__threads_results_for_created_autotests.append(new_thread)	×
73
74	return new_thread	×
75
76	@adapter_logger	2✔
77	def __get_thread_results_for_updated_autotests(self, external_id: str) -> Dict[str, AutoTestResultsForTestRunModel]:	2✔
NEW 78	for thread in self.__threads_results_for_updated_autotests:	×
79	if external_id not in thread.keys():	×
80	return thread	×
81
NEW 82	new_thread: Dict[str, AutoTestResultsForTestRunModel] = {}	×
83
NEW 84	self.__threads_results_for_updated_autotests.append(new_thread)	×
85
86	return new_thread	×

testit-tms / adapters-python / 17801911010

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