Mercurial > repos > shellac > sam_consensus_v3

comparison env/lib/python3.9/site-packages/allure_commons/lifecycle.py @ 0:4f3585e2f14b draft default tip

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
"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 from contextlib import contextmanager
3 from allure_commons._core import plugin_manager
4 from allure_commons.model2 import TestResultContainer
5 from allure_commons.model2 import TestResult
6 from allure_commons.model2 import Attachment, ATTACHMENT_PATTERN
7 from allure_commons.model2 import TestStepResult
8 from allure_commons.model2 import ExecutableItem
9 from allure_commons.model2 import TestBeforeResult
10 from allure_commons.model2 import TestAfterResult
11 from allure_commons.utils import uuid4
12 from allure_commons.utils import now
13 from allure_commons.types import AttachmentType
14
15
16 class AllureLifecycle(object):
17 def __init__(self):
18 self._items = OrderedDict()
19
20 def _get_item(self, uuid=None, item_type=None):
21 uuid = uuid or self._last_item_uuid(item_type=item_type)
22 return self._items.get(uuid)
23
24 def _pop_item(self, uuid=None, item_type=None):
25 uuid = uuid or self._last_item_uuid(item_type=item_type)
26 return self._items.pop(uuid, None)
27
28 def _last_item_uuid(self, item_type=None):
29 for uuid in reversed(self._items):
30 item = self._items.get(uuid)
31 if item_type is None:
32 return uuid
33 elif type(item) == item_type or isinstance(item, item_type):
34 return uuid
35
36 @contextmanager
37 def schedule_test_case(self, uuid=None):
38 test_result = TestResult()
39 test_result.uuid = uuid or uuid4()
40 self._items[test_result.uuid] = test_result
41 yield test_result
42
43 @contextmanager
44 def update_test_case(self, uuid=None):
45 yield self._get_item(uuid=uuid, item_type=TestResult)
46
47 def write_test_case(self, uuid=None):
48 test_result = self._pop_item(uuid=uuid, item_type=TestResult)
49 if test_result:
50 plugin_manager.hook.report_result(result=test_result)
51
52 @contextmanager
53 def start_step(self, parent_uuid=None, uuid=None):
54 parent = self._get_item(uuid=parent_uuid, item_type=ExecutableItem)
55 step = TestStepResult()
56 step.start = now()
57 parent.steps.append(step)
58 self._items[uuid or uuid4()] = step
59 yield step
60
61 @contextmanager
62 def update_step(self, uuid=None):
63 yield self._get_item(uuid=uuid, item_type=TestStepResult)
64
65 def stop_step(self, uuid=None):
66 step = self._pop_item(uuid=uuid, item_type=TestStepResult)
67 if step and not step.stop:
68 step.stop = now()
69
70 @contextmanager
71 def start_container(self, uuid=None):
72 container = TestResultContainer(uuid=uuid or uuid4())
73 self._items[container.uuid] = container
74 yield container
75
76 def containers(self):
77 for item in self._items.values():
78 if type(item) == TestResultContainer:
79 yield item
80
81 @contextmanager
82 def update_container(self, uuid=None):
83 yield self._get_item(uuid=uuid, item_type=TestResultContainer)
84
85 def write_container(self, uuid=None):
86 container = self._pop_item(uuid=uuid, item_type=TestResultContainer)
87 if container and (container.befores or container.afters):
88 plugin_manager.hook.report_container(container=container)
89
90 @contextmanager
91 def start_before_fixture(self, parent_uuid=None, uuid=None):
92 fixture = TestBeforeResult()
93 parent = self._get_item(uuid=parent_uuid, item_type=TestResultContainer)
94 if parent:
95 parent.befores.append(fixture)
96 self._items[uuid or uuid4()] = fixture
97 yield fixture
98
99 @contextmanager
100 def update_before_fixture(self, uuid=None):
101 yield self._get_item(uuid=uuid, item_type=TestBeforeResult)
102
103 def stop_before_fixture(self, uuid=None):
104 fixture = self._pop_item(uuid=uuid, item_type=TestBeforeResult)
105 if fixture and not fixture.stop:
106 fixture.stop = now()
107
108 @contextmanager
109 def start_after_fixture(self, parent_uuid=None, uuid=None):
110 fixture = TestAfterResult()
111 parent = self._get_item(uuid=parent_uuid, item_type=TestResultContainer)
112 if parent:
113 parent.afters.append(fixture)
114 self._items[uuid or uuid4()] = fixture
115 yield fixture
116
117 @contextmanager
118 def update_after_fixture(self, uuid=None):
119 yield self._get_item(uuid=uuid, item_type=TestAfterResult)
120
121 def stop_after_fixture(self, uuid=None):
122 fixture = self._pop_item(uuid=uuid, item_type=TestAfterResult)
123 if fixture and not fixture.stop:
124 fixture.stop = now()
125
126 def _attach(self, uuid, name=None, attachment_type=None, extension=None):
127 mime_type = attachment_type
128 extension = extension if extension else 'attach'
129
130 if type(attachment_type) is AttachmentType:
131 extension = attachment_type.extension
132 mime_type = attachment_type.mime_type
133
134 file_name = ATTACHMENT_PATTERN.format(prefix=uuid, ext=extension)
135 attachment = Attachment(source=file_name, name=name, type=mime_type)
136 uuid = self._last_item_uuid(item_type=ExecutableItem)
137 self._items[uuid].attachments.append(attachment)
138
139 return file_name
140
141 def attach_file(self, uuid, source, name=None, attachment_type=None, extension=None):
142 file_name = self._attach(uuid, name=name, attachment_type=attachment_type, extension=extension)
143 plugin_manager.hook.report_attached_file(source=source, file_name=file_name)
144
145 def attach_data(self, uuid, body, name=None, attachment_type=None, extension=None):
146 file_name = self._attach(uuid, name=name, attachment_type=attachment_type, extension=extension)
147 plugin_manager.hook.report_attached_data(body=body, file_name=file_name)