[telepathy-gabble/master] Test both incoming and outgoing calls
Sjoerd Simons
sjoerd.simons at collabora.co.uk
Tue Dec 29 05:34:52 PST 2009
---
.../{call-outgoing-basics.py => call-basics.py} | 116 ++++++++++++++------
1 files changed, 84 insertions(+), 32 deletions(-)
rename tests/twisted/jingle/{call-outgoing-basics.py => call-basics.py} (54%)
diff --git a/tests/twisted/jingle/call-outgoing-basics.py b/tests/twisted/jingle/call-basics.py
similarity index 54%
rename from tests/twisted/jingle/call-outgoing-basics.py
rename to tests/twisted/jingle/call-basics.py
index 47b6787..765d151 100644
--- a/tests/twisted/jingle/call-outgoing-basics.py
+++ b/tests/twisted/jingle/call-basics.py
@@ -1,5 +1,5 @@
"""
-Test basic outgoing call handling, using CreateChannel
+Test basic outgoing and incoming call handling
"""
import dbus
@@ -9,30 +9,73 @@ from gabbletest import exec_test
from servicetest import (
make_channel_proxy, wrap_channel,
EventPattern, call_async,
- assertEquals, assertContains, assertLength,
+ assertEquals, assertContains, assertLength, assertNotEquals
)
import constants as cs
from jingletest2 import JingleTest2, test_all_dialects
-def run_test(jp, q, bus, conn, stream):
+def check_and_accept_offer (q, bus, conn, self_handle, remote_handle,
+ content, codecs, offer_path = None):
+
+ [path, codecmap] = content.Get(cs.CALL_CONTENT_IFACE_MEDIA,
+ "CodecOffer", dbus_interface=dbus.PROPERTIES_IFACE)
+
+ if offer_path != None:
+ assertEquals (offer_path, path)
+
+ assertNotEquals ("/", path)
+
+ offer = bus.get_object (conn.bus_name, path)
+ codecmap_property = offer.Get (cs.CALL_CONTENT_CODECOFFER,
+ "RemoteContactCodecMap", dbus_interface=dbus.PROPERTIES_IFACE)
+
+ assertEquals (codecmap, codecmap_property)
+
+ offer.Accept (codecs, dbus_interface=cs.CALL_CONTENT_CODECOFFER)
+
+ current_codecs = content.Get(cs.CALL_CONTENT_IFACE_MEDIA,
+ "ContactCodecMap", dbus_interface=dbus.PROPERTIES_IFACE)
+
+ assertEquals (codecs, current_codecs[self_handle])
+
+ o = q.expect ('dbus-signal', signal='CodecsChanged')
+
+ update, _ = o.args
+ assertEquals ({ self_handle: codecs, remote_handle: codecs} , update)
+
+def run_test(jp, q, bus, conn, stream, incoming):
jt2 = JingleTest2(jp, conn, q, stream, 'test at localhost', 'foo at bar.com/Foo')
jt2.prepare()
self_handle = conn.GetSelfHandle()
remote_handle = conn.RequestHandles(1, ["foo at bar.com/Foo"])[0]
+ # Advertise that we can do new style calls
+ conn.ContactCapabilities.UpdateCapabilities([
+ (cs.CLIENT + ".CallHandler", [
+ { cs.CHANNEL_TYPE: cs.CHANNEL_TYPE_CALL,
+ cs.CALL_INITIAL_AUDIO: True},
+ { cs.CHANNEL_TYPE: cs.CHANNEL_TYPE_CALL,
+ cs.CALL_INITIAL_VIDEO: True},
+ ], [
+ cs.CHANNEL_TYPE_CALL + '/gtalk-p2p',
+ cs.CHANNEL_TYPE_CALL + '/ice-udp',
+ cs.CHANNEL_TYPE_CALL + '/video/h264',
+ ]),
+ ])
+
# Ensure a channel that doesn't exist yet.
- call_async(q, conn.Requests, 'CreateChannel',
+ if incoming:
+ jt2.incoming_call()
+ else:
+ ret = conn.Requests.CreateChannel(
{ cs.CHANNEL_TYPE: cs.CHANNEL_TYPE_CALL,
cs.TARGET_HANDLE_TYPE: cs.HT_CONTACT,
cs.TARGET_HANDLE: remote_handle,
cs.CALL_INITIAL_AUDIO: True,
})
- ret, signal = q.expect_many(
- EventPattern('dbus-return', method='CreateChannel'),
- EventPattern('dbus-signal', signal='NewChannels'),
- )
+ signal = q.expect('dbus-signal', signal='NewChannels')
assertLength(1, signal.args)
assertLength(1, signal.args[0]) # one channel
@@ -46,14 +89,18 @@ def run_test(jp, q, bus, conn, stream):
assertEquals(cs.HT_CONTACT, emitted_props[cs.TARGET_HANDLE_TYPE])
assertEquals('foo at bar.com', emitted_props[cs.TARGET_ID])
- assertEquals(True, emitted_props[cs.REQUESTED])
- assertEquals(self_handle, emitted_props[cs.INITIATOR_HANDLE])
- assertEquals('test at localhost', emitted_props[cs.INITIATOR_ID])
+ assertEquals(not incoming, emitted_props[cs.REQUESTED])
+ if incoming:
+ assertEquals(remote_handle, emitted_props[cs.INITIATOR_HANDLE])
+ assertEquals('foo at bar.com', emitted_props[cs.INITIATOR_ID])
+ else:
+ assertEquals(self_handle, emitted_props[cs.INITIATOR_HANDLE])
+ assertEquals('test at localhost', emitted_props[cs.INITIATOR_ID])
assertEquals(True, emitted_props[cs.CALL_INITIAL_AUDIO])
assertEquals(False, emitted_props[cs.CALL_INITIAL_VIDEO])
- chan = bus.get_object (conn.bus_name, ret.value[0])
+ chan = bus.get_object (conn.bus_name, signal.args[0][0][0])
properties = chan.GetAll(cs.CHANNEL_TYPE_CALL,
dbus_interface=dbus.PROPERTIES_IFACE)
@@ -76,11 +123,15 @@ def run_test(jp, q, bus, conn, stream):
# Setup codecs
codecs = jt2.get_call_audio_codecs_dbus()
- content.SetCodecs(codecs, dbus_interface=cs.CALL_CONTENT_IFACE_MEDIA)
+ if incoming:
+ # We should have a codec offer
+ check_and_accept_offer (q, bus, conn, self_handle, remote_handle,
+ content, codecs)
+ else:
+ content.SetCodecs(codecs, dbus_interface=cs.CALL_CONTENT_IFACE_MEDIA)
current_codecs = content.Get(cs.CALL_CONTENT_IFACE_MEDIA,
"ContactCodecMap", dbus_interface=dbus.PROPERTIES_IFACE)
-
assertEquals (codecs, current_codecs[self_handle])
# Add candidates
@@ -108,29 +159,30 @@ def run_test(jp, q, bus, conn, stream):
assertEquals ([], candidates)
- session_initiate = q.expect('stream-iq',
- predicate=jp.action_predicate('session-initiate'))
+ if incoming:
+ endpoint.SetStreamState (cs.MEDIA_STREAM_STATE_CONNECTED,
+ dbus_interface=cs.CALL_STREAM_ENDPOINT)
- jt2.parse_session_initiate(session_initiate.query)
+ chan.Accept (dbus_interface=cs.CHANNEL_TYPE_CALL)
+ q.expect('stream-iq', predicate=jp.action_predicate('session-accept'))
+ else:
+ session_initiate = q.expect('stream-iq',
+ predicate=jp.action_predicate('session-initiate'))
- jt2.accept()
+ jt2.parse_session_initiate(session_initiate.query)
- o = q.expect ('dbus-signal', signal='NewCodecOffer')
+ jt2.accept()
- [path, codecs ] = o.args
- offer = bus.get_object (conn.bus_name, path)
- ocodecs = offer.Get (cs.CALL_CONTENT_CODECOFFER,
- "RemoteContactCodecMap", dbus_interface=dbus.PROPERTIES_IFACE)
-
- assertEquals (codecs, ocodecs)
-
- codecs = jt2.get_call_audio_codecs_dbus()
- offer.Accept (codecs, dbus_interface=cs.CALL_CONTENT_CODECOFFER)
+ o = q.expect ('dbus-signal', signal='NewCodecOffer')
- o = q.expect ('dbus-signal', signal='CodecsChanged')
+ [path, _ ] = o.args
+ codecs = jt2.get_call_audio_codecs_dbus()
- update, _ = o.args
- assertEquals ({ self_handle: codecs, remote_handle: codecs} , update)
+ check_and_accept_offer (q, bus, conn, self_handle, remote_handle,
+ content, codecs, path)
if __name__ == '__main__':
- test_all_dialects(run_test)
+ test_all_dialects(lambda jp, q, bus, conn, stream:
+ run_test(jp, q, bus, conn, stream, False))
+ test_all_dialects(lambda jp, q, bus, conn, stream:
+ run_test(jp, q, bus, conn, stream, True))
--
1.5.6.5
More information about the telepathy-commits
mailing list