[Telepathy-commits] [telepathy-salut/master] add test-request-im.py

[Guillaume Desmottes]

---
 tests/twisted/avahi/test-request-im.py |  183 ++++++++++++++++++++++++++++++++
 1 files changed, 183 insertions(+), 0 deletions(-)
 create mode 100644 tests/twisted/avahi/test-request-im.py

diff --git a/tests/twisted/avahi/test-request-im.py b/tests/twisted/avahi/test-request-im.py
new file mode 100644
index 0000000..94d4588
--- /dev/null
+++ b/tests/twisted/avahi/test-request-im.py
@@ -0,0 +1,183 @@
+
+"""
+Test requesting of text 1-1 channels using the old and new request API.
+"""
+
+import dbus
+
+from saluttest import exec_test
+from servicetest import call_async, EventPattern, \
+        tp_name_prefix, make_channel_proxy
+from avahitest import get_host_name, AvahiAnnouncer
+from xmppstream import setup_stream_listener
+
+CHANNEL_TYPE_TEXT = 'org.freedesktop.Telepathy.Channel.Type.Text'
+
+HT_CONTACT = 1
+HT_CONTACT_LIST = 3
+
+def test(q, bus, conn):
+    self_name = 'testsuite' + '@' + get_host_name()
+
+    conn.Connect()
+    q.expect('dbus-signal', signal='StatusChanged', args=[0L, 0L])
+
+    basic_txt = { "txtvers": "1", "status": "avail" }
+    contact_name = "test-request-im@" + get_host_name()
+    listener, port = setup_stream_listener(q, contact_name)
+
+    # FIXME: this is a hack to be sure to have all the contact list channels
+    # announced so they won't interfere with the muc ones announces.
+    # publish
+    q.expect('dbus-signal', signal='NewChannel')
+    # subscribe
+    q.expect('dbus-signal', signal='NewChannel')
+    # known
+    q.expect('dbus-signal', signal='NewChannel')
+
+    AvahiAnnouncer(contact_name, "_presence._tcp", port, basic_txt)
+
+    publish_handle = conn.RequestHandles(HT_CONTACT_LIST, ["publish"])[0]
+    publish = conn.RequestChannel(
+        "org.freedesktop.Telepathy.Channel.Type.ContactList",
+        HT_CONTACT_LIST, publish_handle, False)
+
+    handle = 0
+    # Wait until the record shows up in publish
+    while handle == 0:
+        e = q.expect('dbus-signal', signal='MembersChanged', path=publish)
+        for h in e.args[1]:
+            name = conn.InspectHandles(HT_CONTACT, [h])[0]
+            if name == contact_name:
+                handle = h
+
+    # check if we can request roomlist channels
+    properties = conn.GetAll(
+            tp_name_prefix + '.Connection.Interface.Requests',
+            dbus_interface='org.freedesktop.DBus.Properties')
+    assert ({tp_name_prefix + '.Channel.ChannelType':
+                CHANNEL_TYPE_TEXT,
+             tp_name_prefix + '.Channel.TargetHandleType': HT_CONTACT,
+             },
+             [tp_name_prefix + '.Channel.TargetHandle',
+              tp_name_prefix + '.Channel.TargetID'],
+             ) in properties.get('RequestableChannelClasses'),\
+                     properties['RequestableChannelClasses']
+
+    # request a muc channel using the old API
+    handle = conn.RequestHandles(HT_CONTACT, [contact_name])[0]
+    call_async(q, conn, 'RequestChannel', CHANNEL_TYPE_TEXT, HT_CONTACT, handle, True)
+
+    ret, old_sig, new_sig = q.expect_many(
+        EventPattern('dbus-return', method='RequestChannel'),
+        EventPattern('dbus-signal', signal='NewChannel'),
+        EventPattern('dbus-signal', signal='NewChannels'),
+        )
+
+    path1 = ret.value[0]
+    chan = make_channel_proxy(conn, path1, "Channel")
+
+    assert new_sig.args[0][0][0] == path1
+
+    props = new_sig.args[0][0][1]
+    assert props[tp_name_prefix + '.Channel.ChannelType'] ==\
+            CHANNEL_TYPE_TEXT
+    assert props[tp_name_prefix + '.Channel.TargetHandleType'] == HT_CONTACT
+    assert props[tp_name_prefix + '.Channel.TargetHandle'] == handle
+    assert props[tp_name_prefix + '.Channel.TargetID'] == contact_name
+    assert props[tp_name_prefix + '.Channel.Requested'] == True
+    assert props[tp_name_prefix + '.Channel.InitiatorHandle'] \
+            == conn.GetSelfHandle()
+    assert props[tp_name_prefix + '.Channel.InitiatorID'] \
+            == self_name
+
+    assert old_sig.args[0] == path1
+    assert old_sig.args[1] == CHANNEL_TYPE_TEXT
+    assert old_sig.args[2] == HT_CONTACT     # handle type
+    assert old_sig.args[3] == handle         # handle
+
+    # Exercise basic Channel Properties from spec 0.17.7
+    channel_props = chan.GetAll(
+            tp_name_prefix + '.Channel',
+            dbus_interface='org.freedesktop.DBus.Properties')
+    assert channel_props.get('TargetHandle') == handle,\
+            channel_props.get('TargetHandle')
+    assert channel_props['TargetID'] == contact_name, channel_props
+    assert channel_props.get('TargetHandleType') == HT_CONTACT,\
+            channel_props.get('TargetHandleType')
+    assert channel_props.get('ChannelType') == \
+            CHANNEL_TYPE_TEXT, channel_props.get('ChannelType')
+    assert channel_props['Requested'] == True
+    assert channel_props['InitiatorID'] == self_name
+    assert channel_props['InitiatorHandle'] == conn.GetSelfHandle()
+
+    requestotron = dbus.Interface(conn,
+            tp_name_prefix + '.Connection.Interface.Requests')
+
+    chan.Close()
+    q.expect_many(
+        EventPattern('dbus-signal', signal='ChannelClosed', args=[path1]),
+        EventPattern('dbus-signal', signal='Closed', path=path1))
+
+    # create muc channel using new API
+    call_async(q, requestotron, 'CreateChannel',
+            { tp_name_prefix + '.Channel.ChannelType':
+                CHANNEL_TYPE_TEXT,
+              tp_name_prefix + '.Channel.TargetHandleType': HT_CONTACT,
+              tp_name_prefix + '.Channel.TargetID': contact_name,
+              })
+
+    ret, old_sig, new_sig = q.expect_many(
+        EventPattern('dbus-return', method='CreateChannel'),
+        EventPattern('dbus-signal', signal='NewChannel'),
+        EventPattern('dbus-signal', signal='NewChannels'),
+        )
+    path2 = ret.value[0]
+    chan = make_channel_proxy(conn, path2, "Channel")
+
+    props = ret.value[1]
+    assert props[tp_name_prefix + '.Channel.ChannelType'] ==\
+            CHANNEL_TYPE_TEXT
+    assert props[tp_name_prefix + '.Channel.TargetHandleType'] == HT_CONTACT
+    assert props[tp_name_prefix + '.Channel.TargetHandle'] == handle
+    assert props[tp_name_prefix + '.Channel.TargetID'] == contact_name
+    assert props[tp_name_prefix + '.Channel.Requested'] == True
+    assert props[tp_name_prefix + '.Channel.InitiatorHandle'] \
+            == conn.GetSelfHandle()
+    assert props[tp_name_prefix + '.Channel.InitiatorID'] \
+            == self_name
+
+    assert new_sig.args[0][0][0] == path2
+    assert new_sig.args[0][0][1] == props
+
+    assert old_sig.args[0] == path2
+    assert old_sig.args[1] == CHANNEL_TYPE_TEXT
+    assert old_sig.args[2] == HT_CONTACT     # handle type
+    assert old_sig.args[3] == handle      # handle
+    assert old_sig.args[4] == True        # suppress handler
+
+    # ensure roomlist channel
+    call_async(q, requestotron, 'EnsureChannel',
+            { tp_name_prefix + '.Channel.ChannelType':
+                CHANNEL_TYPE_TEXT,
+              tp_name_prefix + '.Channel.TargetHandleType': HT_CONTACT,
+              tp_name_prefix + '.Channel.TargetHandle': handle,
+              })
+
+    ret = q.expect('dbus-return', method='EnsureChannel')
+    yours, ensured_path, ensured_props = ret.value
+
+    assert not yours
+    assert ensured_path == path2, (ensured_path, path2)
+
+    conn.Disconnect()
+
+    q.expect_many(
+            EventPattern('dbus-signal', signal='Closed',
+                path=path2),
+            EventPattern('dbus-signal', signal='ChannelClosed', args=[path2]),
+            EventPattern('dbus-signal', signal='StatusChanged', args=[2, 1]),
+            )
+
+if __name__ == '__main__':
+    exec_test(test)
-- 
1.5.6.5