Mercurial > repos > shellac > sam_consensus_v3
comparison env/lib/python3.9/site-packages/allure_commons/reporter.py @ 0:4f3585e2f14b draft default tip
"planemo upload commit 60cee0fc7c0cda8592644e1aad72851dec82c959"
author | shellac |
---|---|
date | Mon, 22 Mar 2021 18:12:50 +0000 |
parents | |
children |
comparison
equal
deleted
inserted
replaced
-1:000000000000 | 0:4f3585e2f14b |
---|---|
1 from collections import OrderedDict | |
2 | |
3 from allure_commons.types import AttachmentType | |
4 from allure_commons.model2 import ExecutableItem | |
5 from allure_commons.model2 import TestResult | |
6 from allure_commons.model2 import Attachment, ATTACHMENT_PATTERN | |
7 from allure_commons.utils import now | |
8 from allure_commons._core import plugin_manager | |
9 | |
10 | |
11 class AllureReporter(object): | |
12 def __init__(self): | |
13 self._items = OrderedDict() | |
14 self._orphan_items = [] | |
15 | |
16 def _update_item(self, uuid, **kwargs): | |
17 item = self._items[uuid] if uuid else self._items[next(reversed(self._items))] | |
18 for name, value in kwargs.items(): | |
19 attr = getattr(item, name) | |
20 if isinstance(attr, list): | |
21 attr.append(value) | |
22 else: | |
23 setattr(item, name, value) | |
24 | |
25 def _last_executable(self): | |
26 for _uuid in reversed(self._items): | |
27 if isinstance(self._items[_uuid], ExecutableItem): | |
28 return _uuid | |
29 | |
30 def get_item(self, uuid): | |
31 return self._items.get(uuid) | |
32 | |
33 def get_last_item(self, item_type=None): | |
34 for _uuid in reversed(self._items): | |
35 if item_type is None: | |
36 return self._items.get(_uuid) | |
37 if type(self._items[_uuid]) == item_type: | |
38 return self._items.get(_uuid) | |
39 | |
40 def start_group(self, uuid, group): | |
41 self._items[uuid] = group | |
42 | |
43 def stop_group(self, uuid, **kwargs): | |
44 self._update_item(uuid, **kwargs) | |
45 group = self._items.pop(uuid) | |
46 plugin_manager.hook.report_container(container=group) | |
47 | |
48 def update_group(self, uuid, **kwargs): | |
49 self._update_item(uuid, **kwargs) | |
50 | |
51 def start_before_fixture(self, parent_uuid, uuid, fixture): | |
52 self._items.get(parent_uuid).befores.append(fixture) | |
53 self._items[uuid] = fixture | |
54 | |
55 def stop_before_fixture(self, uuid, **kwargs): | |
56 self._update_item(uuid, **kwargs) | |
57 self._items.pop(uuid) | |
58 | |
59 def start_after_fixture(self, parent_uuid, uuid, fixture): | |
60 self._items.get(parent_uuid).afters.append(fixture) | |
61 self._items[uuid] = fixture | |
62 | |
63 def stop_after_fixture(self, uuid, **kwargs): | |
64 self._update_item(uuid, **kwargs) | |
65 fixture = self._items.pop(uuid) | |
66 fixture.stop = now() | |
67 | |
68 def schedule_test(self, uuid, test_case): | |
69 self._items[uuid] = test_case | |
70 | |
71 def get_test(self, uuid): | |
72 return self.get_item(uuid) if uuid else self.get_last_item(TestResult) | |
73 | |
74 def close_test(self, uuid): | |
75 test_case = self._items.pop(uuid) | |
76 plugin_manager.hook.report_result(result=test_case) | |
77 | |
78 def drop_test(self, uuid): | |
79 self._items.pop(uuid) | |
80 | |
81 def start_step(self, parent_uuid, uuid, step): | |
82 parent_uuid = parent_uuid if parent_uuid else self._last_executable() | |
83 if parent_uuid is None: | |
84 self._orphan_items.append(uuid) | |
85 else: | |
86 self._items[parent_uuid].steps.append(step) | |
87 self._items[uuid] = step | |
88 | |
89 def stop_step(self, uuid, **kwargs): | |
90 if uuid in self._orphan_items: | |
91 self._orphan_items.remove(uuid) | |
92 else: | |
93 self._update_item(uuid, **kwargs) | |
94 self._items.pop(uuid) | |
95 | |
96 def _attach(self, uuid, name=None, attachment_type=None, extension=None): | |
97 mime_type = attachment_type | |
98 extension = extension if extension else 'attach' | |
99 | |
100 if type(attachment_type) is AttachmentType: | |
101 extension = attachment_type.extension | |
102 mime_type = attachment_type.mime_type | |
103 | |
104 file_name = ATTACHMENT_PATTERN.format(prefix=uuid, ext=extension) | |
105 attachment = Attachment(source=file_name, name=name, type=mime_type) | |
106 last_uuid = self._last_executable() | |
107 self._items[last_uuid].attachments.append(attachment) | |
108 | |
109 return file_name | |
110 | |
111 def attach_file(self, uuid, source, name=None, attachment_type=None, extension=None): | |
112 file_name = self._attach(uuid, name=name, attachment_type=attachment_type, extension=extension) | |
113 plugin_manager.hook.report_attached_file(source=source, file_name=file_name) | |
114 | |
115 def attach_data(self, uuid, body, name=None, attachment_type=None, extension=None): | |
116 file_name = self._attach(uuid, name=name, attachment_type=attachment_type, extension=extension) | |
117 plugin_manager.hook.report_attached_data(body=body, file_name=file_name) |