[Telepathy-commits] [telepathy-gabble/master] caps_helper: add more helper code from test-caps-tubes.py

[Guillaume Desmottes]

---
 tests/twisted/caps_helper.py |   83 +++++++++++++++++++++++++++++++++++++++++-
 1 files changed, 82 insertions(+), 1 deletions(-)

diff --git a/tests/twisted/caps_helper.py b/tests/twisted/caps_helper.py
index a299a58..8c66610 100644
--- a/tests/twisted/caps_helper.py
+++ b/tests/twisted/caps_helper.py
@@ -1,8 +1,34 @@
 import hashlib
 import base64
+import dbus
 
-from twisted.words.xish import domish
+from twisted.words.xish import domish, xpath
 from gabbletest import make_result_iq
+from servicetest import EventPattern
+
+from config import PACKAGE_STRING
+import ns
+import constants as cs
+
+text_fixed_properties = dbus.Dictionary({
+    cs.TARGET_HANDLE_TYPE: cs.HT_CONTACT,
+    cs.CHANNEL_TYPE: cs.CHANNEL_TYPE_TEXT
+    })
+text_allowed_properties = dbus.Array([cs.TARGET_HANDLE])
+
+stream_tube_fixed_properties = dbus.Dictionary({
+    cs.TARGET_HANDLE_TYPE: cs.HT_CONTACT,
+    cs.CHANNEL_TYPE: cs.CHANNEL_TYPE_STREAM_TUBE
+    })
+stream_tube_allowed_properties = dbus.Array([cs.TARGET_HANDLE,
+    cs.TARGET_ID, cs.STREAM_TUBE_SERVICE])
+
+dbus_tube_fixed_properties = dbus.Dictionary({
+    cs.TARGET_HANDLE_TYPE: cs.HT_CONTACT,
+    cs.CHANNEL_TYPE: cs.CHANNEL_TYPE_DBUS_TUBE
+    })
+dbus_tube_allowed_properties = dbus.Array([cs.TARGET_HANDLE,
+    cs.TARGET_ID, cs.DBUS_TUBE_SERVICE_NAME])
 
 def compute_caps_hash(identities, features, dataforms):
     S = ''
@@ -30,6 +56,61 @@ def make_caps_disco_reply(stream, req, features):
 
     return iq
 
+def receive_presence_and_ask_caps(q, stream):
+    # receive presence stanza
+    event_stream, event_dbus = q.expect_many(
+            EventPattern('stream-presence'),
+            EventPattern('dbus-signal', signal='ContactCapabilitiesChanged')
+        )
+    assert len(event_dbus.args) == 1
+    signaled_caps = event_dbus.args[0]
+
+    c_nodes = xpath.queryForNodes('/presence/c', event_stream.stanza)
+    assert c_nodes is not None
+    assert len(c_nodes) == 1
+    hash = c_nodes[0].attributes['hash']
+    ver = c_nodes[0].attributes['ver']
+    node = c_nodes[0].attributes['node']
+    assert hash == 'sha-1'
+
+    # ask caps
+    request = """
+<iq from='fake_contact at jabber.org/resource' 
+    id='disco1'
+    to='gabble at jabber.org/resource' 
+    type='get'>
+  <query xmlns='""" + ns.DISCO_INFO + """'
+         node='""" + node + '#' + ver + """'/>
+</iq>
+"""
+    stream.send(request)
+
+    # receive caps
+    event = q.expect('stream-iq', query_ns=ns.DISCO_INFO)
+    caps_str = str(xpath.queryForNodes('/iq/query/feature', event.stanza))
+
+    features = []
+    for feature in xpath.queryForNodes('/iq/query/feature', event.stanza):
+        features.append(feature['var'])
+
+    # Check if the hash matches the announced capabilities
+    assert ver == compute_caps_hash(['client/pc//%s' % PACKAGE_STRING], features, [])
+
+    return (event, caps_str, signaled_caps)
+
+def caps_contain(event, cap):
+    node = xpath.queryForNodes('/iq/query/feature[@var="%s"]'
+            % cap,
+            event.stanza)
+    if node is None:
+        return False
+    if len(node) != 1:
+        return False
+    var = node[0].attributes['var']
+    if var is None:
+        return False
+    return var == cap
+
 if __name__ == '__main__':
     # example from XEP-0115
     assert compute_caps_hash(['client/pc//Exodus 0.9.1'],
-- 
1.5.6.5