-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathtest_event.py
More file actions
137 lines (99 loc) · 3.48 KB
/
test_event.py
File metadata and controls
137 lines (99 loc) · 3.48 KB
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
"""
Tests for ``simpy.events.Event``.
"""
# Pytest gets the parameters "env" and "log" from the *conftest.py* file
import re
import pytest
def test_succeed(env):
"""Test for the Environment.event() helper function."""
def child(env, event):
value = yield event
assert value == 'ohai'
assert env.now == 5
def parent(env):
event = env.event()
env.process(child(env, event))
yield env.timeout(5)
event.succeed('ohai')
env.process(parent(env))
env.run()
def test_fail(env):
"""Test for the Environment.event() helper function."""
def child(env, event):
with pytest.raises(ValueError, match='ohai'):
yield event
assert env.now == 5
def parent(env):
event = env.event()
env.process(child(env, event))
yield env.timeout(5)
event.fail(ValueError('ohai'))
env.process(parent(env))
env.run()
def test_names(env):
def pem():
return
yield
assert re.match(r'<Event\(\) object at 0x.*>', str(env.event()))
assert re.match(r'<Timeout\(1\) object at 0x.*>', str(env.timeout(1)))
assert re.match(
r'<Timeout\(1, value=2\) object at 0x.*>', str(env.timeout(1, value=2))
)
assert re.match(
r'<Condition\(all_events, \(<Event\(\) object at 0x.*>, '
r'<Event\(\) object at 0x.*>\)\) object at 0x.*>',
str(env.event() & env.event()),
)
assert re.match(r'<Process\(pem\) object at 0x.*>', str(env.process(pem())))
def test_value(env):
"""After an event has been triggered, its value becomes accessible."""
event = env.timeout(0, 'I am the value')
env.run()
assert event.value == 'I am the value'
def test_unavailable_value(env):
"""If an event has not yet been triggered, its value is not available and
trying to access it will result in a AttributeError."""
event = env.event()
with pytest.raises(AttributeError, match='.* is not yet available$'):
_ = event.value
def test_triggered(env):
def pem(env, event):
value = yield event
return value
event = env.event()
event.succeed('i was already done')
result = env.run(env.process(pem(env, event)))
assert result == 'i was already done'
def test_callback_modification(env):
"""The callbacks of an event will get set to None before actually invoking
the callbacks. This prevents concurrent modifications."""
def callback(event):
assert event.callbacks is None
event = env.event()
event.callbacks.append(callback)
event.succeed()
env.run(until=event)
def test_condition_callback_removal(env):
"""A condition will remove all outstanding callbacks from its events."""
a, b = env.event(), env.event()
a.succeed()
env.run(until=a | b)
# The condition has removed its callback from event b.
assert not a.callbacks
assert not b.callbacks
def test_condition_nested_callback_removal(env):
"""A condition will remove all outstanding callbacks from its events (even
if nested)."""
a, b, c = env.event(), env.event(), env.event()
b_and_c = b & c
a_or_b_and_c = a | b_and_c
a.succeed()
env.run(until=a_or_b_and_c)
# Callbacks from nested conditions are also removed.
assert not a.callbacks
assert not b.callbacks
assert not c.callbacks
for cb in b_and_c.callbacks:
# b_and_c may have a _build_value callback.
assert cb.__name__ != '_check'
assert not a_or_b_and_c.callbacks