Skip to content

Commit

Permalink
Add a dialplan timeout test
Browse files Browse the repository at this point in the history
Verify that when a session is parked but not handled within the timeout
period (in this case 3 seconds as set in our CI dialplan) it is cancelled
by FS. In the contrary case verify that a simple `session.answer()` within
the timeout results in a successful SIPp `uac` client scenario.

Resolves #47
  • Loading branch information
Tyler Goodlet committed Nov 30, 2017
1 parent ec2f03a commit 00e1f02
Showing 1 changed file with 36 additions and 0 deletions.
36 changes: 36 additions & 0 deletions tests/test_coroutines.py
Original file line number Diff line number Diff line change
Expand Up @@ -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()

0 comments on commit 00e1f02

Please sign in to comment.