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
|
"""py.test config"""
import sys
import threading
from collections import deque
from time import sleep
import pytest
from circuits import BaseComponent, Debugger, Manager, handler
from circuits.core.manager import TIMEOUT
class Watcher(BaseComponent):
def init(self):
self._lock = threading.Lock()
self.events = deque()
@handler(channel='*', priority=999.9)
def _on_event(self, event, *args, **kwargs):
with self._lock:
self.events.append(event)
def wait(self, name, channel=None, timeout=6.0):
for _i in range(int(timeout / TIMEOUT)):
if channel is None:
with self._lock:
for event in self.events:
if event.name == name:
return True
else:
with self._lock:
for event in self.events:
if event.name == name and channel in event.channels:
return True
sleep(TIMEOUT)
return None
class Flag:
status = False
class WaitEvent:
def __init__(self, manager, name, channel=None, timeout=6.0):
if channel is None:
channel = getattr(manager, 'channel', None)
self.timeout = timeout
self.manager = manager
flag = Flag()
@handler(name, channel=channel)
def on_event(self, *args, **kwargs):
flag.status = True
self.handler = self.manager.addHandler(on_event)
self.flag = flag
def wait(self):
try:
for _i in range(int(self.timeout / TIMEOUT)):
if self.flag.status:
return True
sleep(TIMEOUT)
finally:
self.manager.removeHandler(self.handler)
@pytest.fixture(scope='session')
def manager(request):
manager = Manager()
def finalizer():
manager.stop()
request.addfinalizer(finalizer)
waiter = WaitEvent(manager, 'started')
manager.start()
assert waiter.wait()
if request.config.option.verbose:
Debugger().register(manager)
return manager
@pytest.fixture()
def watcher(request, manager):
watcher = Watcher().register(manager)
def finalizer():
waiter = WaitEvent(manager, 'unregistered')
watcher.unregister()
waiter.wait()
request.addfinalizer(finalizer)
return watcher
for key, value in {'WaitEvent': WaitEvent, 'PLATFORM': sys.platform, 'PYVER': sys.version_info[:3]}.items():
setattr(pytest, key, value)
|
