summaryrefslogtreecommitdiff
path: root/tests/twisted/voip/dtmf.py
blob: 4369bc33211e34da7601b2329a24f411b3656609 (plain)
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
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
"""
Test DTMF dialstring playback and signalling.
"""

from sofiatest import exec_test
from servicetest import (
    call_async, wrap_channel, EventPattern,
    assertEquals, assertContains, assertLength, assertSameSets
    )
from voip_test import VoipTestContext
import constants as cs

def setup_dtmf_channel(context, initial_tones=None):
    q = context.q
    bus = context.bus
    conn = context.conn

    conn.Connect()
    q.expect('dbus-signal', signal='StatusChanged', args=[0, 1])

    request_params = {
            cs.CHANNEL_TYPE: cs.CHANNEL_TYPE_STREAMED_MEDIA,
            cs.TARGET_HANDLE_TYPE: cs.HT_CONTACT,
            cs.TARGET_ID: context.peer,
            cs.INITIAL_AUDIO: True,
        }
    if initial_tones:
        request_params[cs.DTMF_INITIAL_TONES] = initial_tones

    path = conn.Requests.CreateChannel(request_params)[0]

    chan = wrap_channel(bus.get_object(conn.bus_name, path), 'StreamedMedia',
        ['MediaSignalling', 'DTMF'])

    channel_props = chan.Properties.GetAll(cs.CHANNEL)

    assertContains(cs.CHANNEL_IFACE_DTMF, channel_props['Interfaces'])

    dtmf_props = chan.Properties.GetAll(cs.CHANNEL_IFACE_DTMF)

    if initial_tones:
        assertEquals(initial_tones, dtmf_props['InitialTones'])
    else:
        assertEquals('', dtmf_props['InitialTones'])
    assertEquals(False, dtmf_props['CurrentlySendingTones'])

    stream_handler = context.handle_audio_session(chan)

    invite_event = q.expect('sip-invite')

    context.accept(invite_event.sip_message)

    q.expect('dbus-signal', signal='SetRemoteCodecs')

    stream_handler.SupportedCodecs(context.get_audio_codecs_dbus())
    stream_handler.StreamState(cs.MEDIA_STREAM_STATE_CONNECTED)

    return chan

def request_initial_tones(q, bus, conn, sip_proxy, peer='foo@bar.com'):
    context = VoipTestContext(q, conn, bus, sip_proxy, 'sip:testacc@127.0.0.1', peer)

    tones = '123'

    chan = setup_dtmf_channel(context, tones)

    q.expect_many(
            EventPattern('dbus-signal', signal='SendingTones', args=[tones]),
            EventPattern('dbus-signal', signal='StartTelephonyEvent', args=[int(tones[0])]))

    assertEquals(True, chan.Properties.Get(cs.CHANNEL_IFACE_DTMF, 'CurrentlySendingTones'))

    q.expect('dbus-signal', signal='StopTelephonyEvent')

    for i in range(1, len(tones) - 1):
        q.expect('dbus-signal', signal='StartTelephonyEvent', args=[int(tones[i])])
        q.expect('dbus-signal', signal='StopTelephonyEvent')

    q.expect('dbus-signal', signal='StoppedTones', args=[False])

    assertEquals(False, chan.Properties.Get(cs.CHANNEL_IFACE_DTMF, 'CurrentlySendingTones'))

def multiple_tones(q, bus, conn, sip_proxy, peer='foo@bar.com'):

    context = VoipTestContext(q, conn, bus, sip_proxy, 'sip:testacc@127.0.0.1', peer)

    chan = setup_dtmf_channel(context)

    tones_deferred = '78'
    tones = '56w' + tones_deferred

    chan.DTMF.MultipleTones(tones)

    q.expect_many(
            EventPattern('dbus-signal', signal='SendingTones', args=[tones]),
            EventPattern('dbus-signal', signal='StartTelephonyEvent', args=[int(tones[0])]))

    dtmf_props = chan.Properties.GetAll(cs.CHANNEL_IFACE_DTMF)
    assertEquals(True, dtmf_props['CurrentlySendingTones'])
    assertEquals('', dtmf_props['DeferredTones'])

    q.expect('dbus-signal', signal='StopTelephonyEvent')

    q.expect('dbus-signal', signal='StartTelephonyEvent', args=[int(tones[1])])
    q.expect('dbus-signal', signal='StopTelephonyEvent')

    q.expect('dbus-signal', signal='TonesDeferred', args=[tones_deferred])

    dtmf_props = chan.Properties.GetAll(cs.CHANNEL_IFACE_DTMF)

    assertEquals(False, dtmf_props['CurrentlySendingTones'])
    assertEquals(tones_deferred, dtmf_props['DeferredTones'])

    chan.DTMF.MultipleTones(tones_deferred)

    q.expect_many(
            EventPattern('dbus-signal', signal='SendingTones', args=[tones_deferred]),
            EventPattern('dbus-signal', signal='StartTelephonyEvent', args=[int(tones_deferred[0])]))

    dtmf_props = chan.Properties.GetAll(cs.CHANNEL_IFACE_DTMF)

    assertEquals(True, dtmf_props['CurrentlySendingTones'])
    assertEquals('', dtmf_props['DeferredTones'])

    q.expect('dbus-signal', signal='StopTelephonyEvent')

    for i in range(1, len(tones_deferred) - 1):
        q.expect('dbus-signal', signal='StartTelephonyEvent', args=[int(tones_deferred[i])])
        q.expect('dbus-signal', signal='StopTelephonyEvent')

    q.expect('dbus-signal', signal='StoppedTones', args=[False])

def bleep_bloop(q, bus, conn, sip_proxy, peer='foo@bar.com'):

    context = VoipTestContext(q, conn, bus, sip_proxy, 'sip:testacc@127.0.0.1', peer)

    chan = setup_dtmf_channel(context)

    call_async(q, chan.DTMF, 'StartTone', 666, 3)
    q.expect_many(
            EventPattern('dbus-signal', signal='StartTelephonyEvent'),
            EventPattern('dbus-signal', signal='SendingTones', args=['3']),
            EventPattern('dbus-return', method='StartTone'),
            )

    assertEquals(True, chan.Properties.Get(cs.CHANNEL_IFACE_DTMF, 'CurrentlySendingTones'))

    call_async(q, chan.DTMF, 'StopTone', 666)
    q.expect_many(
            EventPattern('dbus-signal', signal='StopTelephonyEvent'),
            EventPattern('dbus-signal', signal='StoppedTones', args=[True]),
            EventPattern('dbus-return', method='StopTone'),
            )

    assertEquals(False, chan.Properties.Get(cs.CHANNEL_IFACE_DTMF, 'CurrentlySendingTones'))

if __name__ == '__main__':
    exec_test(request_initial_tones)
    exec_test(multiple_tones)
    exec_test(bleep_bloop)