24412263511

Committed 14 Apr 2026 05:02PM UTC coverage: 23.905% (+6.5%) from 17.402%

Build # 24412263511

Build Type

Pull #258

github

Committed by

web-flow

Commit Message

Merge 76b75860a into 3330c5056

Pull Request Pull Request #258: add TMS_LEGACY_WORKFLOW flag and debug logs for sync-storage

Coverage Stats

14 of 21 new or added lines in 1 file covered. (66.67%)

393 of 1644 relevant lines covered (23.91%)

0.48 hits per line

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

48.21

/testit-adapter-robotframework/src/testit_adapter_robotframework/models.py

import ast
import re

from attr import Factory, asdict, attrib, s

from robot.api import logger
from testit_python_commons.models.link import Link

from .utils import get_hash


LinkTypes = ['Related', 'BlockedBy', 'Defect', 'Issue', 'Requirement', 'Repository']


def link_type_check(self, attribute, value):
    if value.title() not in LinkTypes:
        raise ValueError(f"Incorrect Link type: {value}")


def url_check(self, attribute, value):
    if not bool(re.match(
            r"(https?|ftp)://"
            r"(\w+(-\w+)*\.)?"
            r"((\w+(-\w+)*)\.(\w+))"
            r"(\.\w+)*"
            r"([\w\-._~/]*)*(?<!\.)",
            value)):
        raise ValueError(f"Incorrect URL: {value}")


class Default:

    def order(self):
        return asdict(self)


def _clean_value(value):
    return str(value).replace("'", "").replace('"', '')


def _parse_testit_tag(tag):
    tag = str(tag).strip()
    if not tag.lower().startswith('testit.'):
        return None, None

    body = tag.split('.', 1)[-1]
    separator = ':' if ':' in body else ('=' if '=' in body else None)
    if not separator:
        logger.error(f"[TestIt] Wrong tag format: {tag}")
        return None, None

    attr, value = body.split(separator, 1)
    return attr.strip().lower(), value.strip()


@s
class StepResult(Default):
    title = attrib(default='')
    description = attrib(default='')
    started_on = attrib(default=None)
    completed_on = attrib(default=None)
    duration = attrib(default=None)
    outcome = attrib(default=None)
    step_results = attrib(default=Factory(list))
    attachments = attrib(default=Factory(list))
    parameters = attrib(default=Factory(dict))


@s
class Step(Default):
    title = attrib()
    description = attrib()
    steps = attrib(default=Factory(list))


@s
class Label:
    name = attrib()


@s(kw_only=True)
class Autotest(Default):
    externalID = attrib(default=None)  # noqa: N815
    autoTestName = attrib()  # noqa: N815
    steps = attrib(default=Factory(list))
    stepResults = attrib(default=Factory(list))  # noqa: N815
    setUp = attrib(default=Factory(list))  # noqa: N815
    setUpResults = attrib(default=Factory(list))  # noqa: N815
    tearDown = attrib(default=Factory(list))  # noqa: N815
    tearDownResults = attrib(default=Factory(list))  # noqa: N815
    resultLinks = attrib(default=Factory(list))  # noqa: N815
    duration = attrib(default=None)
    failureReasonNames = attrib(default=Factory(list))  # noqa: N815
    traces = attrib(default=None)
    outcome = attrib(default=None)
    status_type = attrib(default=None)
    namespace = attrib(default=None)
    attachments = attrib(default=Factory(list))
    parameters = attrib(default=Factory(dict))
    properties = attrib(default=Factory(dict))
    classname = attrib(default=None)
    title = attrib(default=None)
    description = attrib(default=None)
    links = attrib(default=Factory(list))
    labels = attrib(default=Factory(list))
    tags = attrib(default=Factory(list))
    workItemsID = attrib(default=Factory(list))  # noqa: N815
    message = attrib(default="")
    started_on = attrib(default=None)
    completed_on = attrib(default=None)
    externalKey = attrib(default=None)

    step_depth = attrib(default=Factory(list))
    result_depth = attrib(default=Factory(list))

    def add_attributes(self, attrs):
        self.title = attrs['originalname']
        self.autoTestName = attrs['originalname']
        self.externalKey = attrs['originalname']
        self.description = attrs['doc']
        self.template = attrs['template']
        self.classname = attrs['longname'].split('.')[-2]
        for tag in attrs['tags']:
            attr, value = _parse_testit_tag(tag)
            if attr:
                if attr == 'externalid':
                    self.externalID = _clean_value(value)
                elif attr == 'displayname':
                    self.autoTestName = _clean_value(value)
                elif attr == 'title':
                    self.title = _clean_value(value)
                elif attr == 'description':
                    self.description = _clean_value(value)
                elif attr == 'workitemsid' or attr == 'workitemsids':
                    value = ast.literal_eval(value)
                    if isinstance(value, (str, int)):
                        self.workItemsID.append(str(value))
                    elif isinstance(value, list):
                        self.workItemsID.extend([str(i) for i in value])
                    else:
                        logger.error(f"[TestIt] Wrong workitem format: {value}")
                elif attr == 'links':
                    value = ast.literal_eval(value)
                    try:
                        if isinstance(value, dict):
                            self.links.append(Link()\
                                .set_url(value['url'])\
                                .set_title(value.get('title', None))\
                                .set_link_type(value.get('type', None))\
                                .set_description(value.get('description', None)))
                        elif isinstance(value, list):
                            self.links.extend([Link()\
                                .set_url(link['url'])\
                                .set_title(link.get('title', None))\
                                .set_link_type(link.get('type', None))\
                                .set_description(link.get('description', None)) for link in value if isinstance(link, dict)])
                    except ValueError as e:
                        logger.error(f"[TestIt] Link Error: {e}")
                elif attr == 'labels':
                    value = ast.literal_eval(value)
                    if isinstance(value, (str, int)):
                        self.labels.append(Label(value))
                    elif isinstance(value, list):
                        self.labels.extend([Label(item) for item in value if isinstance(item, (str, int))])
                elif attr == 'tags':
                    value = ast.literal_eval(value)
                    if isinstance(value, (str, int)):
                        self.tags.append(str(value))
                    elif isinstance(value, list):
                        self.tags.extend([str(item) for item in value if isinstance(item, (str, int))])
                elif attr == 'namespace':
                    self.namespace = _clean_value(value)
                elif attr == 'classname':
                    self.classname = _clean_value(value)
                else:
                    logger.error(f"[TestIt] Unknown attribute: {attr}")
        if not self.externalID:
            self.externalID = get_hash(attrs['longname'])

    def add_step(self, step_type, title, description, parameters):
        if len(self.step_depth) == 0:
            if step_type.lower() == 'setup':
                self.setUp.append(Step(title, description))
                self.step_depth.append(self.setUp[-1])
                self.setUpResults.append(StepResult(title, description, parameters=parameters))
                self.result_depth.append(self.setUpResults[-1])
            elif step_type.lower() == 'teardown':
                self.tearDown.append(Step(title, description))
                self.step_depth.append(self.tearDown[-1])
                self.tearDownResults.append(StepResult(title, description, parameters=parameters))
                self.result_depth.append(self.tearDownResults[-1])
            else:
                self.steps.append(Step(title, description))
                self.step_depth.append(self.steps[-1])
                self.stepResults.append(StepResult(title, description, parameters=parameters))
                self.result_depth.append(self.stepResults[-1])
        elif 1 <= len(self.step_depth) < 14:
            self.step_depth[-1].steps.append(Step(title, description))
            self.step_depth.append(self.step_depth[-1].steps[-1])
            self.result_depth[-1].step_results.append(StepResult(title, description, parameters=parameters))
            self.result_depth.append(self.result_depth[-1].step_results[-1])

    def add_step_result(self, title, start, complete, duration, outcome, attachments):
        if self.result_depth:
            if self.result_depth[-1].title == title:
                step = self.result_depth.pop()
                step.started_on = start
                step.completed_on = complete
                step.duration = duration
                step.outcome = outcome
                step.attachments = attachments
        if self.step_depth:
            if self.step_depth[-1].title == title:
                self.step_depth.pop()


class Option:

    def __init__(self, **kwargs):
        if kwargs.get('tmsUrl', None):
            self.set_url = kwargs.get('tmsUrl', None)
        if kwargs.get('tmsPrivateToken', None):
            self.set_private_token = kwargs.get('tmsPrivateToken', None)
        if kwargs.get('tmsProjectId', None):
            self.set_project_id = kwargs.get('tmsProjectId', None)
        if kwargs.get('tmsConfigurationId', None):
            self.set_configuration_id = kwargs.get('tmsConfigurationId', None)
        if kwargs.get('tmsTestRunId', None):
            self.set_test_run_id = kwargs.get('tmsTestRunId', None)
        if kwargs.get('tmsProxy', None):
            self.set_tms_proxy = kwargs.get('tmsProxy', None)
        if kwargs.get('tmsTestRunName', None):
            self.set_test_run_name = kwargs.get('tmsTestRunName', None)
        if kwargs.get('tmsAdapterMode', None):
            self.set_adapter_mode = kwargs.get('tmsAdapterMode', None)
        if kwargs.get('tmsConfigFile', None):
            self.set_config_file = kwargs.get('tmsConfigFile', None)
        if kwargs.get('tmsCertValidation', None):
            self.set_cert_validation = kwargs.get('tmsCertValidation', None)
        if kwargs.get('tmsAutomaticCreationTestCases', None):
            self.set_automatic_creation_test_cases = kwargs.get('tmsAutomaticCreationTestCases', None)
        if kwargs.get('tmsAutomaticUpdationLinksToTestCases', None):
            self.set_automatic_updation_links_to_test_cases = kwargs.get('tmsAutomaticUpdationLinksToTestCases', None)
        if kwargs.get('tmsImportRealtime', None):
            self.set_import_realtime = kwargs.get('tmsImportRealtime', None)

1	import ast	2✔
2	import re	2✔
3
4	from attr import Factory, asdict, attrib, s	2✔
5
6	from robot.api import logger	2✔
7	from testit_python_commons.models.link import Link	2✔
8
9	from .utils import get_hash	2✔
10
11
12	LinkTypes = ['Related', 'BlockedBy', 'Defect', 'Issue', 'Requirement', 'Repository']	2✔
13
14
15	def link_type_check(self, attribute, value):	2✔
16	if value.title() not in LinkTypes:	×
17	raise ValueError(f"Incorrect Link type: {value}")	×
18
19
20	def url_check(self, attribute, value):	2✔
21	if not bool(re.match(	×
22	r"(https?\|ftp)://"
23	r"(\w+(-\w+)*\.)?"
24	r"((\w+(-\w+)*)\.(\w+))"
25	r"(\.\w+)*"
26	r"([\w\-._~/])(?<!\.)",
27	value)):
28	raise ValueError(f"Incorrect URL: {value}")	×
29
30
31	class Default:	2✔
32
33	def order(self):	2✔
34	return asdict(self)	×
35
36
37	def _clean_value(value):	2✔
38	return str(value).replace("'", "").replace('"', '')	2✔
39
40
41	def _parse_testit_tag(tag):	2✔
42	tag = str(tag).strip()	2✔
43	if not tag.lower().startswith('testit.'):	2✔
NEW 44	return None, None	×
45
46	body = tag.split('.', 1)[-1]	2✔
47	separator = ':' if ':' in body else ('=' if '=' in body else None)	2✔
48	if not separator:	2✔
NEW 49	logger.error(f"[TestIt] Wrong tag format: {tag}")	×
NEW 50	return None, None	×
51
52	attr, value = body.split(separator, 1)	2✔
53	return attr.strip().lower(), value.strip()	2✔
54
55
56	@s	2✔
57	class StepResult(Default):	2✔
58	title = attrib(default='')	2✔
59	description = attrib(default='')	2✔
60	started_on = attrib(default=None)	2✔
61	completed_on = attrib(default=None)	2✔
62	duration = attrib(default=None)	2✔
63	outcome = attrib(default=None)	2✔
64	step_results = attrib(default=Factory(list))	2✔
65	attachments = attrib(default=Factory(list))	2✔
66	parameters = attrib(default=Factory(dict))	2✔
67
68
69	@s	2✔
70	class Step(Default):	2✔
71	title = attrib()	2✔
72	description = attrib()	2✔
73	steps = attrib(default=Factory(list))	2✔
74
75
76	@s	2✔
77	class Label:	2✔
78	name = attrib()	2✔
79
80
81	@s(kw_only=True)	2✔
82	class Autotest(Default):	2✔
83	externalID = attrib(default=None) # noqa: N815	2✔
84	autoTestName = attrib() # noqa: N815	2✔
85	steps = attrib(default=Factory(list))	2✔
86	stepResults = attrib(default=Factory(list)) # noqa: N815	2✔
87	setUp = attrib(default=Factory(list)) # noqa: N815	2✔
88	setUpResults = attrib(default=Factory(list)) # noqa: N815	2✔
89	tearDown = attrib(default=Factory(list)) # noqa: N815	2✔
90	tearDownResults = attrib(default=Factory(list)) # noqa: N815	2✔
91	resultLinks = attrib(default=Factory(list)) # noqa: N815	2✔
92	duration = attrib(default=None)	2✔
93	failureReasonNames = attrib(default=Factory(list)) # noqa: N815	2✔
94	traces = attrib(default=None)	2✔
95	outcome = attrib(default=None)	2✔
96	status_type = attrib(default=None)	2✔
97	namespace = attrib(default=None)	2✔
98	attachments = attrib(default=Factory(list))	2✔
99	parameters = attrib(default=Factory(dict))	2✔
100	properties = attrib(default=Factory(dict))	2✔
101	classname = attrib(default=None)	2✔
102	title = attrib(default=None)	2✔
103	description = attrib(default=None)	2✔
104	links = attrib(default=Factory(list))	2✔
105	labels = attrib(default=Factory(list))	2✔
106	tags = attrib(default=Factory(list))	2✔
107	workItemsID = attrib(default=Factory(list)) # noqa: N815	2✔
108	message = attrib(default="")	2✔
109	started_on = attrib(default=None)	2✔
110	completed_on = attrib(default=None)	2✔
111	externalKey = attrib(default=None)	2✔
112
113	step_depth = attrib(default=Factory(list))	2✔
114	result_depth = attrib(default=Factory(list))	2✔
115
116	def add_attributes(self, attrs):	2✔
117	self.title = attrs['originalname']	2✔
118	self.autoTestName = attrs['originalname']	2✔
119	self.externalKey = attrs['originalname']	2✔
120	self.description = attrs['doc']	2✔
121	self.template = attrs['template']	2✔
122	self.classname = attrs['longname'].split('.')[-2]	2✔
123	for tag in attrs['tags']:	2✔
124	attr, value = _parse_testit_tag(tag)	2✔
125	if attr:	2✔
126	if attr == 'externalid':	2✔
NEW 127	self.externalID = _clean_value(value)	×
128	elif attr == 'displayname':	2✔
129	self.autoTestName = _clean_value(value)	2✔
130	elif attr == 'title':	2✔
131	self.title = _clean_value(value)	2✔
132	elif attr == 'description':	×
NEW 133	self.description = _clean_value(value)	×
134	elif attr == 'workitemsid' or attr == 'workitemsids':	×
135	value = ast.literal_eval(value)	×
136	if isinstance(value, (str, int)):	×
137	self.workItemsID.append(str(value))	×
138	elif isinstance(value, list):	×
139	self.workItemsID.extend([str(i) for i in value])	×
140	else:
141	logger.error(f"[TestIt] Wrong workitem format: {value}")	×
142	elif attr == 'links':	×
143	value = ast.literal_eval(value)	×
144	try:	×
145	if isinstance(value, dict):	×
146	self.links.append(Link()\	×
147	.set_url(value['url'])\
148	.set_title(value.get('title', None))\
149	.set_link_type(value.get('type', None))\
150	.set_description(value.get('description', None)))
151	elif isinstance(value, list):	×
152	self.links.extend([Link()\	×
153	.set_url(link['url'])\
154	.set_title(link.get('title', None))\
155	.set_link_type(link.get('type', None))\
156	.set_description(link.get('description', None)) for link in value if isinstance(link, dict)])
157	except ValueError as e:	×
158	logger.error(f"[TestIt] Link Error: {e}")	×
159	elif attr == 'labels':	×
160	value = ast.literal_eval(value)	×
161	if isinstance(value, (str, int)):	×
162	self.labels.append(Label(value))	×
163	elif isinstance(value, list):	×
164	self.labels.extend([Label(item) for item in value if isinstance(item, (str, int))])	×
165	elif attr == 'tags':	×
166	value = ast.literal_eval(value)	×
167	if isinstance(value, (str, int)):	×
168	self.tags.append(str(value))	×
169	elif isinstance(value, list):	×
170	self.tags.extend([str(item) for item in value if isinstance(item, (str, int))])	×
171	elif attr == 'namespace':	×
NEW 172	self.namespace = _clean_value(value)	×
173	elif attr == 'classname':	×
NEW 174	self.classname = _clean_value(value)	×
175	else:
176	logger.error(f"[TestIt] Unknown attribute: {attr}")	×
177	if not self.externalID:	2✔
178	self.externalID = get_hash(attrs['longname'])	2✔
179
180	def add_step(self, step_type, title, description, parameters):	2✔
181	if len(self.step_depth) == 0:	×
182	if step_type.lower() == 'setup':	×
183	self.setUp.append(Step(title, description))	×
184	self.step_depth.append(self.setUp[-1])	×
185	self.setUpResults.append(StepResult(title, description, parameters=parameters))	×
186	self.result_depth.append(self.setUpResults[-1])	×
187	elif step_type.lower() == 'teardown':	×
188	self.tearDown.append(Step(title, description))	×
189	self.step_depth.append(self.tearDown[-1])	×
190	self.tearDownResults.append(StepResult(title, description, parameters=parameters))	×
191	self.result_depth.append(self.tearDownResults[-1])	×
192	else:
193	self.steps.append(Step(title, description))	×
194	self.step_depth.append(self.steps[-1])	×
195	self.stepResults.append(StepResult(title, description, parameters=parameters))	×
196	self.result_depth.append(self.stepResults[-1])	×
197	elif 1 <= len(self.step_depth) < 14:	×
198	self.step_depth[-1].steps.append(Step(title, description))	×
199	self.step_depth.append(self.step_depth[-1].steps[-1])	×
200	self.result_depth[-1].step_results.append(StepResult(title, description, parameters=parameters))	×
201	self.result_depth.append(self.result_depth[-1].step_results[-1])	×
202
203	def add_step_result(self, title, start, complete, duration, outcome, attachments):	2✔
204	if self.result_depth:	×
205	if self.result_depth[-1].title == title:	×
206	step = self.result_depth.pop()	×
207	step.started_on = start	×
208	step.completed_on = complete	×
209	step.duration = duration	×
210	step.outcome = outcome	×
211	step.attachments = attachments	×
212	if self.step_depth:	×
213	if self.step_depth[-1].title == title:	×
214	self.step_depth.pop()	×
215
216
217	class Option:	2✔
218
219	def __init__(self, **kwargs):	2✔
220	if kwargs.get('tmsUrl', None):	×
221	self.set_url = kwargs.get('tmsUrl', None)	×
222	if kwargs.get('tmsPrivateToken', None):	×
223	self.set_private_token = kwargs.get('tmsPrivateToken', None)	×
224	if kwargs.get('tmsProjectId', None):	×
225	self.set_project_id = kwargs.get('tmsProjectId', None)	×
226	if kwargs.get('tmsConfigurationId', None):	×
227	self.set_configuration_id = kwargs.get('tmsConfigurationId', None)	×
228	if kwargs.get('tmsTestRunId', None):	×
229	self.set_test_run_id = kwargs.get('tmsTestRunId', None)	×
230	if kwargs.get('tmsProxy', None):	×
231	self.set_tms_proxy = kwargs.get('tmsProxy', None)	×
232	if kwargs.get('tmsTestRunName', None):	×
233	self.set_test_run_name = kwargs.get('tmsTestRunName', None)	×
234	if kwargs.get('tmsAdapterMode', None):	×
235	self.set_adapter_mode = kwargs.get('tmsAdapterMode', None)	×
236	if kwargs.get('tmsConfigFile', None):	×
237	self.set_config_file = kwargs.get('tmsConfigFile', None)	×
238	if kwargs.get('tmsCertValidation', None):	×
239	self.set_cert_validation = kwargs.get('tmsCertValidation', None)	×
240	if kwargs.get('tmsAutomaticCreationTestCases', None):	×
241	self.set_automatic_creation_test_cases = kwargs.get('tmsAutomaticCreationTestCases', None)	×
242	if kwargs.get('tmsAutomaticUpdationLinksToTestCases', None):	×
243	self.set_automatic_updation_links_to_test_cases = kwargs.get('tmsAutomaticUpdationLinksToTestCases', None)	×
244	if kwargs.get('tmsImportRealtime', None):	×
245	self.set_import_realtime = kwargs.get('tmsImportRealtime', None)	×

testit-tms / adapters-python / 24412263511

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