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