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):
try:
yield event
pytest.fail('Should not get here.')
except ValueError as err:
assert err.args[0] == 'ohai'
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():
yield env.exit()
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 availabe and
trying to access it will result in a AttributeError."""
event = env.event()
try:
event.value
assert False, 'Expected an exception'
except AttributeError as e:
assert e.args[0].endswith('is not yet available')
def test_triggered(env):
def pem(env, event):
value = yield event
env.exit(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 and 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
|
