[Telepathy-commits] [telepathy-gabble/master] rename test-muc-offer-stream-tube-ibb.py to offer-muc-stream-tube-ibb.py
Guillaume Desmottes
guillaume.desmottes at collabora.co.uk
Mon Jan 12 05:06:51 PST 2009
---
tests/twisted/tubes/offer-muc-stream-tube-ibb.py | 307 ++++++++++++++++++++
.../tubes/test-muc-offer-stream-tube-ibb.py | 307 --------------------
2 files changed, 307 insertions(+), 307 deletions(-)
create mode 100644 tests/twisted/tubes/offer-muc-stream-tube-ibb.py
delete mode 100644 tests/twisted/tubes/test-muc-offer-stream-tube-ibb.py
diff --git a/tests/twisted/tubes/offer-muc-stream-tube-ibb.py b/tests/twisted/tubes/offer-muc-stream-tube-ibb.py
new file mode 100644
index 0000000..001a4ad
--- /dev/null
+++ b/tests/twisted/tubes/offer-muc-stream-tube-ibb.py
@@ -0,0 +1,307 @@
+"""Test IBB stream tube support in the context of a MUC."""
+
+import base64
+import errno
+import os
+
+import dbus
+
+from servicetest import call_async, EventPattern, tp_name_prefix, EventProtocolFactory
+from gabbletest import exec_test, make_result_iq, acknowledge_iq
+
+from twisted.words.xish import domish, xpath
+from twisted.internet import reactor
+from twisted.words.protocols.jabber.client import IQ
+
+sample_parameters = dbus.Dictionary({
+ 's': 'hello',
+ 'ay': dbus.ByteArray('hello'),
+ 'u': dbus.UInt32(123),
+ 'i': dbus.Int32(-123),
+ }, signature='sv')
+
+NS_TUBES = 'http://telepathy.freedesktop.org/xmpp/tubes'
+NS_SI = 'http://jabber.org/protocol/si'
+NS_FEATURE_NEG = 'http://jabber.org/protocol/feature-neg'
+NS_IBB = 'http://jabber.org/protocol/ibb'
+NS_MUC_BYTESTREAM = 'http://telepathy.freedesktop.org/xmpp/protocol/muc-bytestream'
+NS_X_DATA = 'jabber:x:data'
+
+
+def set_up_listener_socket(q):
+ factory = EventProtocolFactory(q)
+ try:
+ os.remove(os.getcwd() + '/stream')
+ except OSError, e:
+ if e.errno != errno.ENOENT:
+ raise
+ reactor.listenUNIX(os.getcwd() + '/stream', factory)
+
+def test(q, bus, conn, stream):
+ set_up_listener_socket(q)
+ conn.Connect()
+
+ _, iq_event = q.expect_many(
+ EventPattern('dbus-signal', signal='StatusChanged', args=[0, 1]),
+ EventPattern('stream-iq', to=None, query_ns='vcard-temp',
+ query_name='vCard'))
+
+ acknowledge_iq(stream, iq_event.stanza)
+
+ self_handle = conn.GetSelfHandle()
+ self_name = conn.InspectHandles(1, [self_handle])[0]
+
+ call_async(q, conn, 'RequestHandles', 2,
+ ['chat at conf.localhost'])
+
+ event = q.expect('stream-iq', to='conf.localhost',
+ query_ns='http://jabber.org/protocol/disco#info')
+ result = make_result_iq(stream, event.stanza)
+ feature = result.firstChildElement().addElement('feature')
+ feature['var'] = 'http://jabber.org/protocol/muc'
+ stream.send(result)
+
+ event = q.expect('dbus-return', method='RequestHandles')
+ handles = event.value[0]
+
+ # request tubes channel
+ call_async(q, conn, 'RequestChannel',
+ tp_name_prefix + '.Channel.Type.Tubes', 2, handles[0], True)
+
+ _, stream_event = q.expect_many(
+ EventPattern('dbus-signal', signal='MembersChanged',
+ args=[u'', [], [], [], [2], 0, 0]),
+ EventPattern('stream-presence', to='chat at conf.localhost/test'))
+
+ # Send presence for other member of room.
+ presence = domish.Element((None, 'presence'))
+ presence['from'] = 'chat at conf.localhost/bob'
+ x = presence.addElement(('http://jabber.org/protocol/muc#user', 'x'))
+ item = x.addElement('item')
+ item['affiliation'] = 'owner'
+ item['role'] = 'moderator'
+ stream.send(presence)
+
+ # Send presence for own membership of room.
+ presence = domish.Element((None, 'presence'))
+ presence['from'] = 'chat at conf.localhost/test'
+ x = presence.addElement(('http://jabber.org/protocol/muc#user', 'x'))
+ item = x.addElement('item')
+ item['affiliation'] = 'none'
+ item['role'] = 'participant'
+ stream.send(presence)
+
+ q.expect('dbus-signal', signal='MembersChanged',
+ args=[u'', [2, 3], [], [], [], 0, 0])
+
+ assert conn.InspectHandles(1, [2]) == ['chat at conf.localhost/test']
+ assert conn.InspectHandles(1, [3]) == ['chat at conf.localhost/bob']
+ bob_handle = 3
+
+ old_sig, new_sig, returned = q.expect_many(
+ # first text channel is created
+ EventPattern('dbus-signal', signal='NewChannel'),
+ EventPattern('dbus-signal', signal='NewChannels'),
+ EventPattern('dbus-return', method='RequestChannel'))
+
+ path, type, handle_type, handle, supress = old_sig.args
+ assert type == 'org.freedesktop.Telepathy.Channel.Type.Text'
+ assert handle_type == 2 #room
+ assert handle == handles[0]
+ assert supress == False
+
+ # check text channel properties
+ text_props = new_sig.args[0][0][1]
+ assert text_props[tp_name_prefix + '.Channel.ChannelType'] ==\
+ 'org.freedesktop.Telepathy.Channel.Type.Text'
+ assert text_props[tp_name_prefix + '.Channel.TargetHandleType'] == 2
+ assert text_props[tp_name_prefix + '.Channel.TargetHandle'] == handle
+ assert text_props[tp_name_prefix + '.Channel.TargetID'] == 'chat at conf.localhost'
+ assert text_props[tp_name_prefix + '.Channel.Requested'] == False
+ assert text_props[tp_name_prefix + '.Channel.InitiatorHandle'] \
+ == self_handle
+ assert text_props[tp_name_prefix + '.Channel.InitiatorID'] \
+ == self_name
+
+ # tube channel is created
+ old_sig, new_sig = q.expect_many(
+ EventPattern('dbus-signal', signal='NewChannel'),
+ EventPattern('dbus-signal', signal='NewChannels'))
+ path, type, handle_type, handle, supress = old_sig.args
+ assert type == 'org.freedesktop.Telepathy.Channel.Type.Tubes'
+ assert handle_type == 2 #room
+ assert handle == handles[0]
+ assert supress == True
+
+ # check tubes channel properties
+ tubes_props = new_sig.args[0][0][1]
+ assert tubes_props[tp_name_prefix + '.Channel.ChannelType'] ==\
+ 'org.freedesktop.Telepathy.Channel.Type.Tubes'
+ assert tubes_props[tp_name_prefix + '.Channel.TargetHandleType'] == 2
+ assert tubes_props[tp_name_prefix + '.Channel.TargetHandle'] == handle
+ assert tubes_props[tp_name_prefix + '.Channel.TargetID'] == 'chat at conf.localhost'
+ assert tubes_props[tp_name_prefix + '.Channel.Requested'] == True
+ assert tubes_props[tp_name_prefix + '.Channel.InitiatorHandle'] \
+ == self_handle
+ assert tubes_props[tp_name_prefix + '.Channel.InitiatorID'] \
+ == self_name
+
+ tubes_chan = bus.get_object(conn.bus_name, returned.value[0])
+ tubes_iface = dbus.Interface(tubes_chan,
+ tp_name_prefix + '.Channel.Type.Tubes')
+
+ tubes_self_handle = tubes_chan.GetSelfHandle(
+ dbus_interface=tp_name_prefix + '.Channel.Interface.Group')
+
+ # Unix socket
+ path = os.getcwd() + '/stream'
+ call_async(q, tubes_iface, 'OfferStreamTube',
+ 'echo', sample_parameters, 0, dbus.ByteArray(path), 0, "")
+
+ new_tube_event, stream_event, _ = q.expect_many(
+ EventPattern('dbus-signal', signal='NewTube'),
+ EventPattern('stream-presence', to='chat at conf.localhost/test'),
+ EventPattern('dbus-return', method='OfferStreamTube'))
+
+ # handle new_tube_event
+ stream_tube_id = new_tube_event.args[0]
+ assert new_tube_event.args[1] == tubes_self_handle
+ assert new_tube_event.args[2] == 1 # Stream
+ assert new_tube_event.args[3] == 'echo'
+ assert new_tube_event.args[4] == sample_parameters
+ assert new_tube_event.args[5] == 2 # OPEN
+
+ # handle stream_event
+ # We announce our newly created tube in our muc presence
+ presence = stream_event.stanza
+ x_nodes = xpath.queryForNodes('/presence/x[@xmlns="http://jabber.org/'
+ 'protocol/muc"]', presence)
+ assert x_nodes is not None
+ assert len(x_nodes) == 1
+
+ tubes_nodes = xpath.queryForNodes('/presence/tubes[@xmlns="%s"]'
+ % NS_TUBES, presence)
+ assert tubes_nodes is not None
+ assert len(tubes_nodes) == 1
+
+ tube_nodes = xpath.queryForNodes('/tubes/tube', tubes_nodes[0])
+ assert tube_nodes is not None
+ assert len(tube_nodes) == 1
+ for tube in tube_nodes:
+ assert tube['type'] == 'stream'
+ assert not tube.hasAttribute('initiator')
+ assert tube['service'] == 'echo'
+ assert not tube.hasAttribute('stream-id')
+ assert not tube.hasAttribute('dbus-name')
+ assert tube['id'] == str(stream_tube_id)
+
+ params = {}
+ parameter_nodes = xpath.queryForNodes('/tube/parameters/parameter', tube)
+ for node in parameter_nodes:
+ assert node['name'] not in params
+ params[node['name']] = (node['type'], str(node))
+ assert params == {'ay': ('bytes', 'aGVsbG8='),
+ 's': ('str', 'hello'),
+ 'i': ('int', '-123'),
+ 'u': ('uint', '123'),
+ }
+
+ tubes = tubes_iface.ListTubes(byte_arrays=True)
+ assert tubes == [(
+ stream_tube_id,
+ tubes_self_handle,
+ 1, # Stream
+ 'echo',
+ sample_parameters,
+ 2, # OPEN
+ )]
+
+ # FIXME: if we use an unknown JID here, everything fails
+ # (the code uses lookup where it should use ensure)
+
+ # The CM is the server, so fake a client wanting to talk to it
+ iq = IQ(stream, 'set')
+ iq['to'] = 'test at localhost/Resource'
+ iq['from'] = 'chat at conf.localhost/bob'
+ si = iq.addElement((NS_SI, 'si'))
+ si['id'] = 'alpha'
+ si['profile'] = NS_TUBES
+ feature = si.addElement((NS_FEATURE_NEG, 'feature'))
+ x = feature.addElement((NS_X_DATA, 'x'))
+ x['type'] = 'form'
+ field = x.addElement((None, 'field'))
+ field['var'] = 'stream-method'
+ field['type'] = 'list-single'
+ option = field.addElement((None, 'option'))
+ value = option.addElement((None, 'value'))
+ value.addContent(NS_IBB)
+
+ stream_node = si.addElement((NS_TUBES, 'muc-stream'))
+ stream_node['tube'] = str(stream_tube_id)
+
+ stream.send(iq)
+
+ event = q.expect('socket-connected')
+ protocol = event.protocol
+
+ iq_event, _ = q.expect_many(
+ EventPattern('stream-iq', iq_type='result'),
+ EventPattern('dbus-signal', signal='StreamTubeNewConnection',
+ args=[stream_tube_id, bob_handle]))
+
+ # handle iq_event
+ iq = iq_event.stanza
+ si = xpath.queryForNodes('/iq/si[@xmlns="%s"]' % NS_SI,
+ iq)[0]
+ value = xpath.queryForNodes('/si/feature/x/field/value', si)
+ assert len(value) == 1
+ proto = value[0]
+ assert str(proto) == NS_IBB
+ tube = xpath.queryForNodes('/si/tube[@xmlns="%s"]' % NS_TUBES, si)
+ assert len(tube) == 1
+
+ # have the fake client open the stream
+ iq = IQ(stream, 'set')
+ iq['to'] = 'test at localhost/Resource'
+ iq['from'] = 'chat at conf.localhost/bob'
+ open = iq.addElement((NS_IBB, 'open'))
+ open['sid'] = 'alpha'
+ open['block-size'] = '4096'
+ stream.send(iq)
+
+ q.expect('stream-iq', iq_type='result')
+
+ # have the fake client send us some data
+ message = domish.Element(('jabber:client', 'message'))
+ message['to'] = 'test at localhost/Resource'
+ message['from'] = 'chat at conf.localhost/bob'
+ data_node = message.addElement((NS_IBB, 'data'))
+ data_node['sid'] = 'alpha'
+ data_node['seq'] = '0'
+ data_node.addContent(base64.b64encode('hello initiator'))
+ stream.send(message)
+
+ # the server reply
+ event = q.expect('socket-data', data='hello initiator', protocol=protocol)
+ protocol.sendData('hello joiner')
+
+ # we receive server's data
+ event = q.expect('stream-message', to='chat at conf.localhost/bob')
+ message = event.stanza
+
+ data_nodes = xpath.queryForNodes('/message/data[@xmlns="%s"]' % NS_IBB,
+ message)
+ assert data_nodes is not None
+ assert len(data_nodes) == 1
+ ibb_data = data_nodes[0]
+ assert ibb_data['sid'] == 'alpha'
+ binary = base64.b64decode(str(ibb_data))
+ assert binary == 'hello joiner'
+
+ # OK, we're done
+ conn.Disconnect()
+
+ q.expect('dbus-signal', signal='StatusChanged', args=[2, 1])
+
+if __name__ == '__main__':
+ exec_test(test)
diff --git a/tests/twisted/tubes/test-muc-offer-stream-tube-ibb.py b/tests/twisted/tubes/test-muc-offer-stream-tube-ibb.py
deleted file mode 100644
index 001a4ad..0000000
--- a/tests/twisted/tubes/test-muc-offer-stream-tube-ibb.py
+++ /dev/null
@@ -1,307 +0,0 @@
-"""Test IBB stream tube support in the context of a MUC."""
-
-import base64
-import errno
-import os
-
-import dbus
-
-from servicetest import call_async, EventPattern, tp_name_prefix, EventProtocolFactory
-from gabbletest import exec_test, make_result_iq, acknowledge_iq
-
-from twisted.words.xish import domish, xpath
-from twisted.internet import reactor
-from twisted.words.protocols.jabber.client import IQ
-
-sample_parameters = dbus.Dictionary({
- 's': 'hello',
- 'ay': dbus.ByteArray('hello'),
- 'u': dbus.UInt32(123),
- 'i': dbus.Int32(-123),
- }, signature='sv')
-
-NS_TUBES = 'http://telepathy.freedesktop.org/xmpp/tubes'
-NS_SI = 'http://jabber.org/protocol/si'
-NS_FEATURE_NEG = 'http://jabber.org/protocol/feature-neg'
-NS_IBB = 'http://jabber.org/protocol/ibb'
-NS_MUC_BYTESTREAM = 'http://telepathy.freedesktop.org/xmpp/protocol/muc-bytestream'
-NS_X_DATA = 'jabber:x:data'
-
-
-def set_up_listener_socket(q):
- factory = EventProtocolFactory(q)
- try:
- os.remove(os.getcwd() + '/stream')
- except OSError, e:
- if e.errno != errno.ENOENT:
- raise
- reactor.listenUNIX(os.getcwd() + '/stream', factory)
-
-def test(q, bus, conn, stream):
- set_up_listener_socket(q)
- conn.Connect()
-
- _, iq_event = q.expect_many(
- EventPattern('dbus-signal', signal='StatusChanged', args=[0, 1]),
- EventPattern('stream-iq', to=None, query_ns='vcard-temp',
- query_name='vCard'))
-
- acknowledge_iq(stream, iq_event.stanza)
-
- self_handle = conn.GetSelfHandle()
- self_name = conn.InspectHandles(1, [self_handle])[0]
-
- call_async(q, conn, 'RequestHandles', 2,
- ['chat at conf.localhost'])
-
- event = q.expect('stream-iq', to='conf.localhost',
- query_ns='http://jabber.org/protocol/disco#info')
- result = make_result_iq(stream, event.stanza)
- feature = result.firstChildElement().addElement('feature')
- feature['var'] = 'http://jabber.org/protocol/muc'
- stream.send(result)
-
- event = q.expect('dbus-return', method='RequestHandles')
- handles = event.value[0]
-
- # request tubes channel
- call_async(q, conn, 'RequestChannel',
- tp_name_prefix + '.Channel.Type.Tubes', 2, handles[0], True)
-
- _, stream_event = q.expect_many(
- EventPattern('dbus-signal', signal='MembersChanged',
- args=[u'', [], [], [], [2], 0, 0]),
- EventPattern('stream-presence', to='chat at conf.localhost/test'))
-
- # Send presence for other member of room.
- presence = domish.Element((None, 'presence'))
- presence['from'] = 'chat at conf.localhost/bob'
- x = presence.addElement(('http://jabber.org/protocol/muc#user', 'x'))
- item = x.addElement('item')
- item['affiliation'] = 'owner'
- item['role'] = 'moderator'
- stream.send(presence)
-
- # Send presence for own membership of room.
- presence = domish.Element((None, 'presence'))
- presence['from'] = 'chat at conf.localhost/test'
- x = presence.addElement(('http://jabber.org/protocol/muc#user', 'x'))
- item = x.addElement('item')
- item['affiliation'] = 'none'
- item['role'] = 'participant'
- stream.send(presence)
-
- q.expect('dbus-signal', signal='MembersChanged',
- args=[u'', [2, 3], [], [], [], 0, 0])
-
- assert conn.InspectHandles(1, [2]) == ['chat at conf.localhost/test']
- assert conn.InspectHandles(1, [3]) == ['chat at conf.localhost/bob']
- bob_handle = 3
-
- old_sig, new_sig, returned = q.expect_many(
- # first text channel is created
- EventPattern('dbus-signal', signal='NewChannel'),
- EventPattern('dbus-signal', signal='NewChannels'),
- EventPattern('dbus-return', method='RequestChannel'))
-
- path, type, handle_type, handle, supress = old_sig.args
- assert type == 'org.freedesktop.Telepathy.Channel.Type.Text'
- assert handle_type == 2 #room
- assert handle == handles[0]
- assert supress == False
-
- # check text channel properties
- text_props = new_sig.args[0][0][1]
- assert text_props[tp_name_prefix + '.Channel.ChannelType'] ==\
- 'org.freedesktop.Telepathy.Channel.Type.Text'
- assert text_props[tp_name_prefix + '.Channel.TargetHandleType'] == 2
- assert text_props[tp_name_prefix + '.Channel.TargetHandle'] == handle
- assert text_props[tp_name_prefix + '.Channel.TargetID'] == 'chat at conf.localhost'
- assert text_props[tp_name_prefix + '.Channel.Requested'] == False
- assert text_props[tp_name_prefix + '.Channel.InitiatorHandle'] \
- == self_handle
- assert text_props[tp_name_prefix + '.Channel.InitiatorID'] \
- == self_name
-
- # tube channel is created
- old_sig, new_sig = q.expect_many(
- EventPattern('dbus-signal', signal='NewChannel'),
- EventPattern('dbus-signal', signal='NewChannels'))
- path, type, handle_type, handle, supress = old_sig.args
- assert type == 'org.freedesktop.Telepathy.Channel.Type.Tubes'
- assert handle_type == 2 #room
- assert handle == handles[0]
- assert supress == True
-
- # check tubes channel properties
- tubes_props = new_sig.args[0][0][1]
- assert tubes_props[tp_name_prefix + '.Channel.ChannelType'] ==\
- 'org.freedesktop.Telepathy.Channel.Type.Tubes'
- assert tubes_props[tp_name_prefix + '.Channel.TargetHandleType'] == 2
- assert tubes_props[tp_name_prefix + '.Channel.TargetHandle'] == handle
- assert tubes_props[tp_name_prefix + '.Channel.TargetID'] == 'chat at conf.localhost'
- assert tubes_props[tp_name_prefix + '.Channel.Requested'] == True
- assert tubes_props[tp_name_prefix + '.Channel.InitiatorHandle'] \
- == self_handle
- assert tubes_props[tp_name_prefix + '.Channel.InitiatorID'] \
- == self_name
-
- tubes_chan = bus.get_object(conn.bus_name, returned.value[0])
- tubes_iface = dbus.Interface(tubes_chan,
- tp_name_prefix + '.Channel.Type.Tubes')
-
- tubes_self_handle = tubes_chan.GetSelfHandle(
- dbus_interface=tp_name_prefix + '.Channel.Interface.Group')
-
- # Unix socket
- path = os.getcwd() + '/stream'
- call_async(q, tubes_iface, 'OfferStreamTube',
- 'echo', sample_parameters, 0, dbus.ByteArray(path), 0, "")
-
- new_tube_event, stream_event, _ = q.expect_many(
- EventPattern('dbus-signal', signal='NewTube'),
- EventPattern('stream-presence', to='chat at conf.localhost/test'),
- EventPattern('dbus-return', method='OfferStreamTube'))
-
- # handle new_tube_event
- stream_tube_id = new_tube_event.args[0]
- assert new_tube_event.args[1] == tubes_self_handle
- assert new_tube_event.args[2] == 1 # Stream
- assert new_tube_event.args[3] == 'echo'
- assert new_tube_event.args[4] == sample_parameters
- assert new_tube_event.args[5] == 2 # OPEN
-
- # handle stream_event
- # We announce our newly created tube in our muc presence
- presence = stream_event.stanza
- x_nodes = xpath.queryForNodes('/presence/x[@xmlns="http://jabber.org/'
- 'protocol/muc"]', presence)
- assert x_nodes is not None
- assert len(x_nodes) == 1
-
- tubes_nodes = xpath.queryForNodes('/presence/tubes[@xmlns="%s"]'
- % NS_TUBES, presence)
- assert tubes_nodes is not None
- assert len(tubes_nodes) == 1
-
- tube_nodes = xpath.queryForNodes('/tubes/tube', tubes_nodes[0])
- assert tube_nodes is not None
- assert len(tube_nodes) == 1
- for tube in tube_nodes:
- assert tube['type'] == 'stream'
- assert not tube.hasAttribute('initiator')
- assert tube['service'] == 'echo'
- assert not tube.hasAttribute('stream-id')
- assert not tube.hasAttribute('dbus-name')
- assert tube['id'] == str(stream_tube_id)
-
- params = {}
- parameter_nodes = xpath.queryForNodes('/tube/parameters/parameter', tube)
- for node in parameter_nodes:
- assert node['name'] not in params
- params[node['name']] = (node['type'], str(node))
- assert params == {'ay': ('bytes', 'aGVsbG8='),
- 's': ('str', 'hello'),
- 'i': ('int', '-123'),
- 'u': ('uint', '123'),
- }
-
- tubes = tubes_iface.ListTubes(byte_arrays=True)
- assert tubes == [(
- stream_tube_id,
- tubes_self_handle,
- 1, # Stream
- 'echo',
- sample_parameters,
- 2, # OPEN
- )]
-
- # FIXME: if we use an unknown JID here, everything fails
- # (the code uses lookup where it should use ensure)
-
- # The CM is the server, so fake a client wanting to talk to it
- iq = IQ(stream, 'set')
- iq['to'] = 'test at localhost/Resource'
- iq['from'] = 'chat at conf.localhost/bob'
- si = iq.addElement((NS_SI, 'si'))
- si['id'] = 'alpha'
- si['profile'] = NS_TUBES
- feature = si.addElement((NS_FEATURE_NEG, 'feature'))
- x = feature.addElement((NS_X_DATA, 'x'))
- x['type'] = 'form'
- field = x.addElement((None, 'field'))
- field['var'] = 'stream-method'
- field['type'] = 'list-single'
- option = field.addElement((None, 'option'))
- value = option.addElement((None, 'value'))
- value.addContent(NS_IBB)
-
- stream_node = si.addElement((NS_TUBES, 'muc-stream'))
- stream_node['tube'] = str(stream_tube_id)
-
- stream.send(iq)
-
- event = q.expect('socket-connected')
- protocol = event.protocol
-
- iq_event, _ = q.expect_many(
- EventPattern('stream-iq', iq_type='result'),
- EventPattern('dbus-signal', signal='StreamTubeNewConnection',
- args=[stream_tube_id, bob_handle]))
-
- # handle iq_event
- iq = iq_event.stanza
- si = xpath.queryForNodes('/iq/si[@xmlns="%s"]' % NS_SI,
- iq)[0]
- value = xpath.queryForNodes('/si/feature/x/field/value', si)
- assert len(value) == 1
- proto = value[0]
- assert str(proto) == NS_IBB
- tube = xpath.queryForNodes('/si/tube[@xmlns="%s"]' % NS_TUBES, si)
- assert len(tube) == 1
-
- # have the fake client open the stream
- iq = IQ(stream, 'set')
- iq['to'] = 'test at localhost/Resource'
- iq['from'] = 'chat at conf.localhost/bob'
- open = iq.addElement((NS_IBB, 'open'))
- open['sid'] = 'alpha'
- open['block-size'] = '4096'
- stream.send(iq)
-
- q.expect('stream-iq', iq_type='result')
-
- # have the fake client send us some data
- message = domish.Element(('jabber:client', 'message'))
- message['to'] = 'test at localhost/Resource'
- message['from'] = 'chat at conf.localhost/bob'
- data_node = message.addElement((NS_IBB, 'data'))
- data_node['sid'] = 'alpha'
- data_node['seq'] = '0'
- data_node.addContent(base64.b64encode('hello initiator'))
- stream.send(message)
-
- # the server reply
- event = q.expect('socket-data', data='hello initiator', protocol=protocol)
- protocol.sendData('hello joiner')
-
- # we receive server's data
- event = q.expect('stream-message', to='chat at conf.localhost/bob')
- message = event.stanza
-
- data_nodes = xpath.queryForNodes('/message/data[@xmlns="%s"]' % NS_IBB,
- message)
- assert data_nodes is not None
- assert len(data_nodes) == 1
- ibb_data = data_nodes[0]
- assert ibb_data['sid'] == 'alpha'
- binary = base64.b64decode(str(ibb_data))
- assert binary == 'hello joiner'
-
- # OK, we're done
- conn.Disconnect()
-
- q.expect('dbus-signal', signal='StatusChanged', args=[2, 1])
-
-if __name__ == '__main__':
- exec_test(test)
--
1.5.6.5
More information about the Telepathy-commits
mailing list