diff --git a/tests/test_coroutines.py b/tests/test_coroutines.py index 966e4b6..4a3234b 100644 --- a/tests/test_coroutines.py +++ b/tests/test_coroutines.py @@ -88,3 +88,39 @@ async def timeout_on_hangup(self, sess): assert task.done() with pytest.raises(asyncio.TimeoutError): task.result() + + +@pytest.mark.parametrize('sleep', [2, 5]) +def test_simplest_flow(fssock, scenario, client, ael, sleep): + """Verify that a coroutine can satisfy SIPp's simplest call flow. + + Additionally, verify that when calls are not answered and left in the park + state they time out and are rejected after 3 seconds (according to the CI + dialplan). + """ + class MyApp: + @coroutine("CHANNEL_PARK") + async def answer_pause_hangup(self, sess): + await asyncio.sleep(sleep) + sess.answer() + await sess.recv("CHANNEL_HANGUP") + + + client.connect() + client.listener = ael + # assigning a listener overrides it's call lookup var so restore it + client.listener.call_tracking_header = 'variable_call_uuid' + assert 'default' == client.load_app(MyApp, on_value="default") + + uac = scenario.prepare()[1] + uac.proxyaddr = None + uac.destaddr = fssock + + # make the call + if sleep > 3: + # XML dialplan's `park_timeout` should reject the call + with pytest.raises(RuntimeError): + uac() + else: + # call should be hung up by this UAC + uac()