1""" 2Tests for ``simpy.events.Timeout``. 3 4""" 5# Pytest gets the parameters "env" and "log" from the *conftest.py* file 6import pytest 7 8 9def test_discrete_time_steps(env, log): 10 """envple envulation with discrete time steps.""" 11 def pem(env, log): 12 while True: 13 log.append(env.now) 14 yield env.timeout(delay=1) 15 16 env.process(pem(env, log)) 17 env.run(until=3) 18 19 assert log == [0, 1, 2] 20 21 22def test_negative_timeout(env): 23 """Don't allow negative timeout times.""" 24 def pem(env): 25 yield env.timeout(-1) 26 27 env.process(pem(env)) 28 pytest.raises(ValueError, env.run) 29 30 31def test_timeout_value(env): 32 """You can pass an additional *value* to *timeout* which will be 33 directly yielded back into the PEM. This is useful to implement some 34 kinds of resources or other additions. 35 36 See :class:`envpy.resources.Store` for an example. 37 38 """ 39 def pem(env): 40 val = yield env.timeout(1, 'ohai') 41 assert val == 'ohai' 42 43 env.process(pem(env)) 44 env.run() 45 46 47def test_shared_timeout(env, log): 48 def child(env, timeout, id, log): 49 yield timeout 50 log.append((id, env.now)) 51 52 timeout = env.timeout(1) 53 for i in range(3): 54 env.process(child(env, timeout, i, log)) 55 56 env.run() 57 assert log == [(0, 1), (1, 1), (2, 1)] 58 59 60def test_triggered_timeout(env): 61 def process(env): 62 def child(env, event): 63 value = yield event 64 env.exit(value) 65 66 event = env.timeout(1, 'i was already done') 67 # Start the child after the timeout has already happened. 68 yield env.timeout(2) 69 value = yield env.process(child(env, event)) 70 assert value == 'i was already done' 71 72 env.run(env.process(process(env))) 73