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 Switchy command line:

$ switchy dial vm-host sip-cannon --profile external --proxy myproxy.com --rate 1 --limit 1 --max-offered 1

...

Aug 26 21:59:01 [INFO] switchy cli.py:114 : Slave sip-cannon.qa.sangoma.local SIP address is at 10.10.8.19:5080
Aug 26 21:59:01 [INFO] switchy cli.py:114 : Slave vm-host.qa.sangoma.local SIP address is at 10.10.8.21:5080
Aug 26 21:59:01 [INFO] switchy 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 Switchy dial sub-command takes several options and a list of minion IP addresses or hostnames. In this example switchy 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 switchy command line see here.

Originating a single call programatically from Python

Making a call with switchy 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 switchy import sync_caller
from switchy.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 Switchy 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 a bgapi() 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 a switchy.observe.EventListener.waitfor() blocker method.
  • the call is kept up for 1 second and then hungup.
  • internal Switchy 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 :

test_sync_call.py
 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
# 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
from switchy import sync_caller
from switchy.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


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 Switchy’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 switchy.apps.players import TonePlay
from switchy.observe import active_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 active_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
            # switchy 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.waitfor(orig_sess, var, time)

            return orig_sess, client.listener.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