1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
|
import threading
from collections import OrderedDict, defaultdict
from allure_commons.types import AttachmentType
from allure_commons.model2 import ExecutableItem
from allure_commons.model2 import TestResult
from allure_commons.model2 import Attachment, ATTACHMENT_PATTERN
from allure_commons.utils import now
from allure_commons._core import plugin_manager
class ThreadContextItems:
_thread_context = defaultdict(OrderedDict)
_init_thread: threading.Thread
@property
def thread_context(self):
context = self._thread_context[threading.current_thread()]
if not context and threading.current_thread() is not self._init_thread:
uuid, last_item = next(reversed(self._thread_context[self._init_thread].items()))
context[uuid] = last_item
return context
def __init__(self, *args, **kwargs):
self._init_thread = threading.current_thread()
super().__init__(*args, **kwargs)
def __setitem__(self, key, value):
self.thread_context.__setitem__(key, value)
def __getitem__(self, item):
return self.thread_context.__getitem__(item)
def __iter__(self):
return self.thread_context.__iter__()
def __reversed__(self):
return self.thread_context.__reversed__()
def get(self, key):
return self.thread_context.get(key)
def pop(self, key):
return self.thread_context.pop(key)
def cleanup(self):
stopped_threads = []
for thread in self._thread_context.keys():
if not thread.is_alive():
stopped_threads.append(thread)
for thread in stopped_threads:
del self._thread_context[thread]
class AllureReporter:
def __init__(self):
self._items = ThreadContextItems()
self._orphan_items = []
def _update_item(self, uuid, **kwargs):
item = self._items[uuid] if uuid else self._items[next(reversed(self._items))]
for name, value in kwargs.items():
attr = getattr(item, name)
if isinstance(attr, list):
attr.append(value)
else:
setattr(item, name, value)
def _last_executable(self):
for _uuid in reversed(self._items):
if isinstance(self._items[_uuid], ExecutableItem):
return _uuid
def get_item(self, uuid):
return self._items.get(uuid)
def get_last_item(self, item_type=None):
for _uuid in reversed(self._items):
if item_type is None:
return self._items.get(_uuid)
if isinstance(self._items[_uuid], item_type):
return self._items.get(_uuid)
def start_group(self, uuid, group):
self._items[uuid] = group
def stop_group(self, uuid, **kwargs):
self._update_item(uuid, **kwargs)
group = self._items.pop(uuid)
plugin_manager.hook.report_container(container=group)
def update_group(self, uuid, **kwargs):
self._update_item(uuid, **kwargs)
def start_before_fixture(self, parent_uuid, uuid, fixture):
self._items.get(parent_uuid).befores.append(fixture)
self._items[uuid] = fixture
def stop_before_fixture(self, uuid, **kwargs):
self._update_item(uuid, **kwargs)
self._items.pop(uuid)
def start_after_fixture(self, parent_uuid, uuid, fixture):
self._items.get(parent_uuid).afters.append(fixture)
self._items[uuid] = fixture
def stop_after_fixture(self, uuid, **kwargs):
self._update_item(uuid, **kwargs)
fixture = self._items.pop(uuid)
fixture.stop = now()
def schedule_test(self, uuid, test_case):
self._items[uuid] = test_case
def get_test(self, uuid):
return self.get_item(uuid) if uuid else self.get_last_item(TestResult)
def close_test(self, uuid):
test_case = self._items.pop(uuid)
self._items.cleanup()
plugin_manager.hook.report_result(result=test_case)
def drop_test(self, uuid):
self._items.pop(uuid)
def start_step(self, parent_uuid, uuid, step):
parent_uuid = parent_uuid if parent_uuid else self._last_executable()
if parent_uuid is None:
self._orphan_items.append(uuid)
else:
self._items[parent_uuid].steps.append(step)
self._items[uuid] = step
def stop_step(self, uuid, **kwargs):
if uuid in self._orphan_items:
self._orphan_items.remove(uuid)
else:
self._update_item(uuid, **kwargs)
self._items.pop(uuid)
def _attach(self, uuid, name=None, attachment_type=None, extension=None, parent_uuid=None):
mime_type = attachment_type
extension = extension if extension else 'attach'
if type(attachment_type) is AttachmentType:
extension = attachment_type.extension
mime_type = attachment_type.mime_type
file_name = ATTACHMENT_PATTERN.format(prefix=uuid, ext=extension)
attachment = Attachment(source=file_name, name=name, type=mime_type)
last_uuid = parent_uuid if parent_uuid else self._last_executable()
self._items[last_uuid].attachments.append(attachment)
return file_name
def attach_file(self, uuid, source, name=None, attachment_type=None, extension=None, parent_uuid=None):
file_name = self._attach(uuid, name=name, attachment_type=attachment_type,
extension=extension, parent_uuid=parent_uuid)
plugin_manager.hook.report_attached_file(source=source, file_name=file_name)
def attach_data(self, uuid, body, name=None, attachment_type=None, extension=None, parent_uuid=None):
file_name = self._attach(uuid, name=name, attachment_type=attachment_type,
extension=extension, parent_uuid=parent_uuid)
plugin_manager.hook.report_attached_data(body=body, file_name=file_name)
|
