Quick-Start - Originating a single call¶
Assuming you’ve gone through the required deployment steps to setup at least one slave, initiating a call becomes
very simple using the switchio
command line:
$ switchio dial vm-host sip-cannon --profile external --proxy myproxy.com --rate 1 --limit 1 --max-offered 1
...
Aug 26 21:59:01 [INFO] switchio cli.py:114 : Slave sip-cannon.qa.sangoma.local SIP address is at 10.10.8.19:5080
Aug 26 21:59:01 [INFO] switchio cli.py:114 : Slave vm-host.qa.sangoma.local SIP address is at 10.10.8.21:5080
Aug 26 21:59:01 [INFO] switchio cli.py:120 : Starting load test for server dut-008.qa.sangoma.local at 1cps using 2 slaves
<Originator: active-calls=0 state=INITIAL total-originated-sessions=0 rate=1 limit=1 max-offered=1 duration=5>
...
<Originator: active-calls=1 state=STOPPED total-originated-sessions=1 rate=1 limit=1 max-offered=1 duration=5>
Waiting on 1 active calls to finish
Waiting on 1 active calls to finish
Waiting on 1 active calls to finish
Waiting on 1 active calls to finish
Dialing session completed!
The switchio
dial sub-command takes several options and a list of minion IP addresses
or hostnames. In this example switchio
connected to the specified hosts, found the
requested SIP profile and initiated a single call with a duration of 5 seconds to the
device under test (set with the proxy option).
For more information on the switchio command line see here.
Originating a single call programatically from Python¶
Making a call with switchio is quite simple using the built-in
sync_caller()
context manager.
Again, if you’ve gone through the required deployment steps, initiating a call becomes as simple as a few lines of python
code
1 2 3 4 5 6 7 8 9 10 11 12 13 14 | from switchio import sync_caller
from switchio.apps.players import TonePlay
# here '192.168.0.10' would be the address of the server running a
# FS process to be used as the call generator
with sync_caller('192.168.0.10', apps={"tone": TonePlay}) as caller:
# initiates a call to the originating profile on port 5080 using
# the `TonePlay` app and block until answered / the originate job completes
sess, waitfor = caller('Fred@{}:{}'.format(caller.client.host, 5080), "tone")
# let the tone play a bit
time.sleep(5)
# tear down the call
sess.hangup()
|
The most important lines are the with statement and line 10. What happens behind the scenes here is the following:
- at the with, necessary internal
switchio
components are instantiated in memory and connected to a FreeSWITCH process listening on the fsip ESL ip address.- at the caller(), an
originate()
command is invoked asynchronously via abgapi()
call.- the background
Job
returned by that command is handled to completion synchronously wherein the call blocks until the originating session has reached the connected state.- the corresponding origininating
Session
is returned along with a reference to aswitchio.handlers.EventListener.waitfor()
blocker method.- the call is kept up for 1 second and then
hungup
.- internal
switchio
components are disconnected from the FreeSWITCH process at the close of the with block.
Note that the sync_caller api is not normally used for stress testing as it used to initiate calls synchronously. It becomes far more useful when using FreeSWITCH for functional testing using your own custom call flow apps.
Example source code¶
Some more extensive examples are found in the unit tests sources :
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 | # This Source Code Form is subject to the terms of the Mozilla Public
# License, v. 2.0. If a copy of the MPL was not distributed with this
# file, You can obtain one at http://mozilla.org/MPL/2.0/.
"""
Tests for synchronous call helper
"""
import time
import pytest
from switchio import sync_caller
from switchio.apps.players import TonePlay, PlayRec
def test_toneplay(fsip):
'''Test the synchronous caller with a simple toneplay
'''
with sync_caller(fsip, apps={"TonePlay": TonePlay}) as caller:
# have the external prof call itself by default
assert 'TonePlay' in caller.app_names
sess, waitfor = caller(
"doggy@{}:{}".format(caller.client.host, 5080),
'TonePlay',
timeout=3,
)
assert sess.is_outbound()
time.sleep(1)
sess.hangup()
time.sleep(0.1)
assert caller.client.listener.count_calls() == 0
@pytest.mark.skip(reason='FS 1.6+ bug in record events')
def test_playrec(fsip):
'''Test the synchronous caller with a simulated conversation using the the
`PlayRec` app. Currently this test does no audio checking but merely
verifies the callback chain is invoked as expected.
'''
with sync_caller(fsip, apps={"PlayRec": PlayRec}) as caller:
# have the external prof call itself by default
caller.apps.PlayRec['PlayRec'].rec_rate = 1
sess, waitfor = caller(
"doggy@{}:{}".format(caller.client.host, 5080),
'PlayRec',
timeout=10,
)
waitfor(sess, 'recorded', timeout=15)
waitfor(sess.call.get_peer(sess), 'recorded', timeout=15)
assert sess.call.vars['record']
time.sleep(1)
assert sess.hungup
def test_alt_call_tracking_header(fsip):
'''Test that an alternate `EventListener.call_tracking_header` (in this
case using the 'Caller-Destination-Number' channel variable) can be used
to associate sessions into calls.
'''
with sync_caller(fsip) as caller:
# use the destination number as the call association var
caller.client.listener.call_tracking_header = 'Caller-Destination-Number'
dest = 'doggy'
# have the external prof call itself by default
sess, waitfor = caller(
"{}@{}:{}".format(dest, caller.client.host, 5080),
'TonePlay', # the default app
timeout=3,
)
assert sess.is_outbound()
# call should be indexed by the req uri username
assert dest in caller.client.listener.calls
call = caller.client.listener.calls[dest]
time.sleep(1)
assert call.first is sess
assert call.last
call.hangup()
time.sleep(0.1)
assert caller.client.listener.count_calls() == 0
def test_untracked_call(fsip):
with sync_caller(fsip) as caller:
# use an invalid chan var for call tracking
caller.client.listener.call_tracking_header = 'doggypants'
# have the external prof call itself by default
sess, waitfor = caller(
"{}@{}:{}".format('jonesy', caller.client.host, 5080),
'TonePlay', # the default app
timeout=3,
)
# calls should be created for both inbound and outbound sessions
# since our tracking variable is nonsense
l = caller.client.listener
# assert len(l.sessions) == len(l.calls) == 2
assert l.count_sessions() == l.count_calls() == 2
sess.hangup()
time.sleep(0.1)
# no calls or sessions should be active
assert l.count_sessions() == l.count_calls() == 0
assert not l.sessions and not l.calls
|
Run manually¶
You can run this code from the unit test directory quite simply:
>>> from tests.test_sync_call import test_toneplay
>>> test_toneplay('fs_slave_hostname')
Run with pytest¶
If you have pytest
installed you can run this test like so:
$ py.test --fshost='fs_slave_hostname' tests/test_sync_caller
Implementation details¶
The implementation of sync_caller()
is shown
below and can be referenced alongside the Internals tutorial to gain a better
understanding of the inner workings of switchio
’s api:
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 | # This Source Code Form is subject to the terms of the Mozilla Public
# License, v. 2.0. If a copy of the MPL was not distributed with this
# file, You can obtain one at http://mozilla.org/MPL/2.0/.
"""
Make calls synchronously
"""
from contextlib import contextmanager
from switchio.apps.players import TonePlay
from switchio.api import get_client
@contextmanager
def sync_caller(host, port='8021', password='ClueCon',
apps={'TonePlay': TonePlay}):
'''Deliver a provisioned synchronous caller function.
A caller let's you make a call synchronously returning control once
it has entered a stable state. The caller returns the active originating
`Session` and a `waitfor` blocker method as output.
'''
with get_client(host, port=port, auth=password, apps=apps) as client:
def caller(dest_url, app_name, timeout=30, waitfor=None,
**orig_kwargs):
# override the channel variable used to look up the intended
# switchio app to be run for this call
if caller.app_lookup_vars:
client.listener.app_id_vars.extend(caller.app_lookup_vars)
job = client.originate(dest_url, app_id=app_name, **orig_kwargs)
job.get(timeout)
if not job.successful():
raise job.result
call = client.listener.sessions[job.sess_uuid].call
orig_sess = call.first # first sess is the originator
if waitfor:
var, time = waitfor
client.listener.event_loop.waitfor(orig_sess, var, time)
return orig_sess, client.listener.event_loop.waitfor
# attach apps handle for easy interactive use
caller.app_lookup_vars = []
caller.apps = client.apps
caller.client = client
caller.app_names = client._apps.keys()
yield caller
|