1"""
2Tests outgoing calls created with InitialAudio and/or InitialVideo, and
3exposing the initial contents of incoming calls as values of InitialAudio and
4InitialVideo
5"""
6
7import operator
8
9from servicetest import (
10    assertContains, assertEquals, assertLength,
11    wrap_channel, EventPattern, call_async, make_channel_proxy)
12
13from jingletest2 import JingleTest2, test_all_dialects
14
15import constants as cs
16
17from config import VOIP_ENABLED
18
19if not VOIP_ENABLED:
20    print "NOTE: built with --disable-voip"
21    raise SystemExit(77)
22
23def outgoing(jp, q, bus, conn, stream):
24    remote_jid = 'flames@cold.mountain/beyond'
25    jt = JingleTest2(jp, conn, q, stream, 'test@localhost', remote_jid)
26    jt.prepare()
27
28    self_handle = conn.GetSelfHandle()
29    remote_handle = conn.RequestHandles(cs.HT_CONTACT, [remote_jid])[0]
30
31    rccs = conn.Properties.Get(cs.CONN_IFACE_REQUESTS, 'RequestableChannelClasses')
32    media_classes = [ rcc for rcc in rccs
33        if rcc[0][cs.CHANNEL_TYPE] == cs.CHANNEL_TYPE_STREAMED_MEDIA ]
34
35    assertLength(1, media_classes)
36    fixed, allowed = media_classes[0]
37    assertContains(cs.INITIAL_AUDIO, allowed)
38    assertContains(cs.INITIAL_VIDEO, allowed)
39
40    check_neither(q, conn, bus, stream, remote_handle)
41    check_iav(jt, q, conn, bus, stream, remote_handle, True, False)
42    check_iav(jt, q, conn, bus, stream, remote_handle, False, True)
43    check_iav(jt, q, conn, bus, stream, remote_handle, True, True)
44
45def check_neither(q, conn, bus, stream, remote_handle):
46    """
47    Make a channel without specifying InitialAudio or InitialVideo; check
48    that it's announced with both False, and that they're both present and
49    false in GetAll().
50    """
51
52    path, props = conn.Requests.CreateChannel({
53        cs.CHANNEL_TYPE: cs.CHANNEL_TYPE_STREAMED_MEDIA,
54        cs.TARGET_HANDLE_TYPE: cs.HT_CONTACT,
55        cs.TARGET_HANDLE: remote_handle})
56
57    assertContains((cs.INITIAL_AUDIO, False), props.items())
58    assertContains((cs.INITIAL_VIDEO, False), props.items())
59
60    chan = wrap_channel(bus.get_object(conn.bus_name, path),
61        cs.CHANNEL_TYPE_STREAMED_MEDIA, ['MediaSignalling'])
62    props = chan.Properties.GetAll(cs.CHANNEL_TYPE_STREAMED_MEDIA)
63    assertContains(('InitialAudio', False), props.items())
64    assertContains(('InitialVideo', False), props.items())
65
66    # We shouldn't have started a session yet, so there shouldn't be any
67    # session handlers. Strictly speaking, there could be a session handler
68    # with no stream handlers, but...
69    session_handlers = chan.MediaSignalling.GetSessionHandlers()
70    assertLength(0, session_handlers)
71
72def check_iav(jt, q, conn, bus, stream, remote_handle, initial_audio,
73              initial_video):
74    """
75    Make a channel and check that its InitialAudio and InitialVideo properties
76    come out correctly.
77    """
78
79    call_async(q, conn.Requests, 'CreateChannel', {
80        cs.CHANNEL_TYPE: cs.CHANNEL_TYPE_STREAMED_MEDIA,
81        cs.TARGET_HANDLE_TYPE: cs.HT_CONTACT,
82        cs.TARGET_HANDLE: remote_handle,
83        cs.INITIAL_AUDIO: initial_audio,
84        cs.INITIAL_VIDEO: initial_video,
85        })
86    if initial_video and (not jt.jp.can_do_video()
87            or (not initial_audio and not jt.jp.can_do_video_only ())):
88        # Some protocols can't do video
89        event = q.expect('dbus-error', method='CreateChannel')
90        assertEquals(cs.NOT_CAPABLE, event.error.get_dbus_name())
91    else:
92        path, props = q.expect('dbus-return', method='CreateChannel').value
93
94        assertContains((cs.INITIAL_AUDIO, initial_audio), props.items())
95        assertContains((cs.INITIAL_VIDEO, initial_video), props.items())
96
97        chan = wrap_channel(bus.get_object(conn.bus_name, path),
98            cs.CHANNEL_TYPE_STREAMED_MEDIA, ['MediaSignalling'])
99        props = chan.Properties.GetAll(cs.CHANNEL_TYPE_STREAMED_MEDIA)
100        assertContains(('InitialAudio', initial_audio), props.items())
101        assertContains(('InitialVideo', initial_video), props.items())
102
103        session_handlers = chan.MediaSignalling.GetSessionHandlers()
104
105        assertLength(1, session_handlers)
106        path, type = session_handlers[0]
107        assertEquals('rtp', type)
108        session_handler = make_channel_proxy(conn, path, 'Media.SessionHandler')
109        session_handler.Ready()
110
111        stream_handler_paths = []
112        stream_handler_types = []
113
114        for x in [initial_audio, initial_video]:
115            if x:
116                e = q.expect('dbus-signal', signal='NewStreamHandler')
117                stream_handler_paths.append(e.args[0])
118                stream_handler_types.append(e.args[2])
119
120        if initial_audio:
121            assertContains(cs.MEDIA_STREAM_TYPE_AUDIO, stream_handler_types)
122
123        if initial_video:
124            assertContains(cs.MEDIA_STREAM_TYPE_VIDEO, stream_handler_types)
125
126        for x in xrange (0, len(stream_handler_paths)):
127            p = stream_handler_paths[x]
128            t = stream_handler_types[x]
129            sh = make_channel_proxy(conn, p, 'Media.StreamHandler')
130            sh.NewNativeCandidate("fake", jt.get_remote_transports_dbus())
131            if t == cs.MEDIA_STREAM_TYPE_AUDIO:
132                sh.Ready(jt.get_audio_codecs_dbus())
133            else:
134                sh.Ready(jt.get_video_codecs_dbus())
135            sh.StreamState(cs.MEDIA_STREAM_STATE_CONNECTED)
136
137        e = q.expect('stream-iq',
138            predicate=jt.jp.action_predicate('session-initiate'))
139        jt.parse_session_initiate (e.query)
140
141        jt.accept()
142
143        events = reduce(operator.concat,
144            [ [ EventPattern('dbus-signal', signal='SetRemoteCodecs', path=p),
145                EventPattern('dbus-signal', signal='SetStreamPlaying', path=p),
146              ] for p in stream_handler_paths
147            ], [])
148        q.expect_many(*events)
149
150        chan.Close()
151
152def incoming(jp, q, bus, conn, stream):
153    remote_jid = 'skinny.fists@heaven/antennas'
154    jt = JingleTest2(jp, conn, q, stream, 'test@localhost', remote_jid)
155    jt.prepare()
156
157    self_handle = conn.GetSelfHandle()
158    remote_handle = conn.RequestHandles(cs.HT_CONTACT, [remote_jid])[0]
159
160    for a, v in [("audio1", None), (None, "video1"), ("audio1", "video1")]:
161        if v!= None and not jp.can_do_video():
162            continue
163        if a == None and v != None and not jp.can_do_video_only():
164            continue
165
166        jt.incoming_call(audio=a, video=v)
167        e = q.expect('dbus-signal', signal='NewChannels',
168            predicate=lambda e:
169                cs.CHANNEL_TYPE_CONTACT_LIST not in e.args[0][0][1].values())
170        chans = e.args[0]
171        assertLength(1, chans)
172
173        path, props = chans[0]
174
175        assertEquals(cs.CHANNEL_TYPE_STREAMED_MEDIA, props[cs.CHANNEL_TYPE])
176        assertEquals(a != None, props[cs.INITIAL_AUDIO])
177        assertEquals(v != None, props[cs.INITIAL_VIDEO])
178
179        # FIXME: This doesn't check non-Google contacts that can only do one
180        # media type, as such contacts as simulated by JingleTest2 can always
181        # do both.
182        assertEquals(not jp.can_do_video() or not jp.can_do_video_only(),
183            props[cs.IMMUTABLE_STREAMS])
184
185        chan = wrap_channel(bus.get_object(conn.bus_name, path),
186            cs.CHANNEL_TYPE_STREAMED_MEDIA)
187        chan.Close()
188
189
190if __name__ == '__main__':
191    test_all_dialects(outgoing)
192    test_all_dialects(incoming)
193