[Telepathy-commits] [telepathy-gabble/master] Use ns.py everywhere in tubes tests
Will Thompson
will.thompson at collabora.co.uk
Tue Feb 3 12:02:40 PST 2009
---
tests/twisted/ns.py | 2 +
tests/twisted/tubes/accept-muc-stream-tube-ibb.py | 34 +++----
tests/twisted/tubes/close-muc-with-closed-tube.py | 10 +--
tests/twisted/tubes/crash-on-list-channels.py | 16 ++--
tests/twisted/tubes/ensure-si-tube.py | 9 +--
.../offer-accept-private-dbus-stream-tube-ibb.py | 114 +++++++++----------
...offer-accept-private-dbus-stream-tube-socks5.py | 94 ++++++++---------
...offer-accept-private-stream-tube-si-fallback.py | 67 +++++-------
tests/twisted/tubes/offer-muc-dbus-tube.py | 12 +--
tests/twisted/tubes/offer-muc-stream-tube-ibb.py | 36 +++----
tests/twisted/tubes/test-get-available-tubes.py | 9 +--
11 files changed, 175 insertions(+), 228 deletions(-)
diff --git a/tests/twisted/ns.py b/tests/twisted/ns.py
index 974d0b4..fbb0ad2 100644
--- a/tests/twisted/ns.py
+++ b/tests/twisted/ns.py
@@ -1,4 +1,5 @@
AMP = "http://jabber.org/protocol/amp"
+BYTESTREAMS = 'http://jabber.org/protocol/bytestreams'
CHAT_STATES = 'http://jabber.org/protocol/chatstates'
CAPS = "http://jabber.org/protocol/caps"
DISCO_INFO = "http://jabber.org/protocol/disco#info"
@@ -24,6 +25,7 @@ OLPC_CURRENT_ACTIVITY = "http://laptop.org/xmpp/current-activity"
OLPC_CURRENT_ACTIVITY_NOTIFY = "%s+notify" % OLPC_CURRENT_ACTIVITY
PUBSUB = "http://jabber.org/protocol/pubsub"
SI = 'http://jabber.org/protocol/si'
+SI_MULTIPLE = 'http://telepathy.freedesktop.org/xmpp/si-multiple'
STANZA = "urn:ietf:params:xml:ns:xmpp-stanzas"
TUBES = 'http://telepathy.freedesktop.org/xmpp/tubes'
X_DATA = 'jabber:x:data'
diff --git a/tests/twisted/tubes/accept-muc-stream-tube-ibb.py b/tests/twisted/tubes/accept-muc-stream-tube-ibb.py
index b79ccac..5066f3f 100644
--- a/tests/twisted/tubes/accept-muc-stream-tube-ibb.py
+++ b/tests/twisted/tubes/accept-muc-stream-tube-ibb.py
@@ -6,6 +6,7 @@ import dbus
from servicetest import call_async, EventPattern, EventProtocolClientFactory
from gabbletest import exec_test, make_result_iq, acknowledge_iq
from constants import *
+import ns
import tubetestutil as t
from twisted.words.xish import domish, xpath
@@ -20,13 +21,6 @@ sample_parameters = dbus.Dictionary({
'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 test(q, bus, conn, stream):
conn.Connect()
@@ -95,7 +89,7 @@ def test(q, bus, conn, stream):
item = x.addElement('item')
item['affiliation'] = 'owner'
item['role'] = 'moderator'
- tubes = presence.addElement((NS_TUBES, 'tubes'))
+ tubes = presence.addElement((ns.TUBES, 'tubes'))
tube = tubes.addElement((None, 'tube'))
tube['type'] = 'stream'
tube['service'] = 'echo'
@@ -215,17 +209,17 @@ def test(q, bus, conn, stream):
protocol.sendData("hello initiator")
# expect SI request
- event = q.expect('stream-iq', to='chat at conf.localhost/bob', query_ns=NS_SI,
+ event = q.expect('stream-iq', to='chat at conf.localhost/bob', query_ns=ns.SI,
query_name='si')
iq = event.stanza
- si = xpath.queryForNodes('/iq/si[@xmlns="%s"]' % NS_SI,
+ si = xpath.queryForNodes('/iq/si[@xmlns="%s"]' % ns.SI,
iq)[0]
values = xpath.queryForNodes('/si/feature[@xmlns="%s"]/x[@xmlns="%s"]/field/option/value'
% ('http://jabber.org/protocol/feature-neg', 'jabber:x:data'), si)
- assert NS_IBB in [str(v) for v in values]
+ assert ns.IBB in [str(v) for v in values]
muc_stream_node = xpath.queryForNodes('/si/muc-stream[@xmlns="%s"]' %
- NS_TUBES, si)[0]
+ ns.TUBES, si)[0]
assert muc_stream_node is not None
assert muc_stream_node['tube'] == str(stream_tube_id)
stream_id = si['id']
@@ -235,20 +229,20 @@ def test(q, bus, conn, stream):
result["id"] = iq["id"]
result['from'] = 'chat at conf.localhost/bob'
result['to'] = 'chat at conf.localhost/test'
- si = result.addElement((NS_SI, 'si'))
- feature = si.addElement((NS_FEATURE_NEG, 'feature'))
- x = feature.addElement((NS_X_DATA, 'x'))
+ si = result.addElement((ns.SI, 'si'))
+ feature = si.addElement((ns.FEATURE_NEG, 'feature'))
+ x = feature.addElement((ns.X_DATA, 'x'))
x['type'] = 'submit'
field = x.addElement((None, 'field'))
field['var'] = 'stream-method'
value = field.addElement((None, 'value'))
- value.addContent(NS_IBB)
- si.addElement((NS_TUBES, 'tube'))
+ value.addContent(ns.IBB)
+ si.addElement((ns.TUBES, 'tube'))
stream.send(result)
# wait IBB init IQ
event = q.expect('stream-iq', to='chat at conf.localhost/bob',
- query_name='open', query_ns=NS_IBB)
+ query_name='open', query_ns=ns.IBB)
iq = event.stanza
open = xpath.queryForNodes('/iq/open', iq)[0]
assert open['sid'] == stream_id
@@ -259,7 +253,7 @@ def test(q, bus, conn, stream):
event = q.expect('stream-message', to='chat at conf.localhost/bob')
message = event.stanza
- data_nodes = xpath.queryForNodes('/message/data[@xmlns="%s"]' % NS_IBB,
+ data_nodes = xpath.queryForNodes('/message/data[@xmlns="%s"]' % ns.IBB,
message)
assert data_nodes is not None
assert len(data_nodes) == 1
@@ -272,7 +266,7 @@ def test(q, bus, conn, stream):
message = domish.Element(('jabber:client', 'message'))
message['from'] = 'chat at conf.localhost/bob'
message['to'] = 'chat at conf.localhost/test'
- data_node = message.addElement((NS_IBB, 'data'))
+ data_node = message.addElement((ns.IBB, 'data'))
data_node['sid'] = stream_id
data_node['seq'] = '0'
data_node.addContent(base64.b64encode('hi joiner!'))
diff --git a/tests/twisted/tubes/close-muc-with-closed-tube.py b/tests/twisted/tubes/close-muc-with-closed-tube.py
index 83f063c..86cd0f2 100644
--- a/tests/twisted/tubes/close-muc-with-closed-tube.py
+++ b/tests/twisted/tubes/close-muc-with-closed-tube.py
@@ -6,6 +6,7 @@ import dbus
from servicetest import call_async, EventPattern, tp_name_prefix, EventProtocolClientFactory
from gabbletest import exec_test, make_result_iq, acknowledge_iq
from constants import *
+import ns
import tubetestutil as t
from twisted.words.xish import domish, xpath
@@ -19,13 +20,6 @@ sample_parameters = dbus.Dictionary({
'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 test(q, bus, conn, stream):
conn.Connect()
@@ -96,7 +90,7 @@ def test(q, bus, conn, stream):
item = x.addElement('item')
item['affiliation'] = 'owner'
item['role'] = 'moderator'
- tubes = presence.addElement((NS_TUBES, 'tubes'))
+ tubes = presence.addElement((ns.TUBES, 'tubes'))
tube = tubes.addElement((None, 'tube'))
tube['type'] = 'dbus'
tube['service'] = 'org.telepathy.freedesktop.test'
diff --git a/tests/twisted/tubes/crash-on-list-channels.py b/tests/twisted/tubes/crash-on-list-channels.py
index 7989136..1aa3426 100644
--- a/tests/twisted/tubes/crash-on-list-channels.py
+++ b/tests/twisted/tubes/crash-on-list-channels.py
@@ -8,10 +8,10 @@ import dbus
from servicetest import call_async, EventPattern, watch_tube_signals
from gabbletest import exec_test, acknowledge_iq, sync_stream
-from twisted.words.xish import domish, xpath
+import ns
+import constants
-TUBES = 'org.freedesktop.Telepathy.Channel.Type.Tubes'
-NS_TUBES = 'http://telepathy.freedesktop.org/xmpp/tubes'
+from twisted.words.xish import domish, xpath
jid = 'explosions at in.the.sky'
@@ -32,29 +32,29 @@ def test(q, bus, conn, stream):
presence['from'] = '%s/Bob' % jid
presence['to'] = 'test at localhost/Resource'
c = presence.addElement('c')
- c['xmlns'] = 'http://jabber.org/protocol/caps'
+ c['xmlns'] = ns.CAPS
c['node'] = 'http://example.com/ICantBelieveItsNotTelepathy'
c['ver'] = '1.2.3'
stream.send(presence)
event = q.expect('stream-iq', iq_type='get',
- query_ns='http://jabber.org/protocol/disco#info',
+ query_ns=ns.DISCO_INFO,
to=('%s/Bob' % jid))
result = event.stanza
result['type'] = 'result'
assert event.query['node'] == \
'http://example.com/ICantBelieveItsNotTelepathy#1.2.3'
feature = event.query.addElement('feature')
- feature['var'] = NS_TUBES
+ feature['var'] = ns.TUBES
stream.send(result)
sync_stream(q, stream)
h = conn.RequestHandles(1, [jid])[0]
- tubes_path = conn.RequestChannel(TUBES, 1, h, True)
+ tubes_path = conn.RequestChannel(constants.CHANNEL_TYPE_TUBES, 1, h, True)
tubes_chan = bus.get_object(conn.bus_name, tubes_path)
- tubes_iface = dbus.Interface(tubes_chan, TUBES)
+ tubes_iface = dbus.Interface(tubes_chan, constants.CHANNEL_TYPE_TUBES)
tubes_iface.OfferDBusTube('bong.hits', dbus.Dictionary({}, signature='sv'))
diff --git a/tests/twisted/tubes/ensure-si-tube.py b/tests/twisted/tubes/ensure-si-tube.py
index 4b1bf8c..67b6428 100644
--- a/tests/twisted/tubes/ensure-si-tube.py
+++ b/tests/twisted/tubes/ensure-si-tube.py
@@ -12,6 +12,7 @@ from dbus.lowlevel import SignalMessage
from servicetest import call_async, EventPattern, tp_name_prefix, watch_tube_signals
from gabbletest import exec_test, acknowledge_iq
+import ns
from twisted.words.xish import domish, xpath
from twisted.internet.protocol import Factory, Protocol
@@ -20,12 +21,6 @@ from twisted.words.protocols.jabber.client import IQ
import tubetestutil as t
-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_X_DATA = 'jabber:x:data'
-
sample_parameters = dbus.Dictionary({
's': 'hello',
'ay': dbus.ByteArray('hello'),
@@ -85,7 +80,7 @@ def test(q, bus, conn, stream):
assert event.query['node'] == \
'http://example.com/ICantBelieveItsNotTelepathy#1.2.3'
feature = event.query.addElement('feature')
- feature['var'] = NS_TUBES
+ feature['var'] = ns.TUBES
stream.send(result)
bob_handle = conn.RequestHandles(1, ['bob at localhost'])[0]
diff --git a/tests/twisted/tubes/offer-accept-private-dbus-stream-tube-ibb.py b/tests/twisted/tubes/offer-accept-private-dbus-stream-tube-ibb.py
index 0a5b4c5..4259f0a 100644
--- a/tests/twisted/tubes/offer-accept-private-dbus-stream-tube-ibb.py
+++ b/tests/twisted/tubes/offer-accept-private-dbus-stream-tube-ibb.py
@@ -10,6 +10,7 @@ from dbus.lowlevel import SignalMessage
from servicetest import call_async, EventPattern, watch_tube_signals
from gabbletest import exec_test, acknowledge_iq, sync_stream
from constants import *
+import ns
import tubetestutil as t
from dbus import PROPERTIES_IFACE
@@ -18,13 +19,6 @@ from twisted.words.xish import domish, xpath
from twisted.internet import reactor
from twisted.words.protocols.jabber.client import IQ
-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_X_DATA = 'jabber:x:data'
-NS_BYTESTREAMS = 'http://jabber.org/protocol/bytestreams'
-
sample_parameters = dbus.Dictionary({
's': 'hello',
'ay': dbus.ByteArray('hello'),
@@ -109,20 +103,20 @@ def contact_offer_dbus_tube(stream, si_id, tube_id):
iq = IQ(stream, 'set')
iq['to'] = 'test at localhost/Resource'
iq['from'] = 'bob at localhost/Bob'
- si = iq.addElement((NS_SI, 'si'))
+ si = iq.addElement((ns.SI, 'si'))
si['id'] = si_id
- si['profile'] = NS_TUBES
- feature = si.addElement((NS_FEATURE_NEG, 'feature'))
- x = feature.addElement((NS_X_DATA, 'x'))
+ 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)
+ value.addContent(ns.IBB)
- tube = si.addElement((NS_TUBES, 'tube'))
+ tube = si.addElement((ns.TUBES, 'tube'))
tube['type'] = 'dbus'
tube['service'] = 'com.example.TestCase2'
tube['id'] = tube_id
@@ -198,7 +192,7 @@ def test(q, bus, conn, stream):
assert event.query['node'] == \
'http://example.com/ICantBelieveItsNotTelepathy#1.2.3'
feature = event.query.addElement('feature')
- feature['var'] = NS_TUBES
+ feature['var'] = ns.TUBES
stream.send(result)
# A tube request can be done only if the contact has tube capabilities
@@ -322,7 +316,7 @@ def test(q, bus, conn, stream):
message = event.stanza
assert message['to'] == 'bob at localhost/Bob' # check the resource
- tube_nodes = xpath.queryForNodes('/message/tube[@xmlns="%s"]' % NS_TUBES,
+ tube_nodes = xpath.queryForNodes('/message/tube[@xmlns="%s"]' % ns.TUBES,
message)
assert tube_nodes is not None
assert len(tube_nodes) == 1
@@ -413,7 +407,7 @@ def test(q, bus, conn, stream):
message = msg_event.stanza
assert message['to'] == 'bob at localhost/Bob' # check the resource
- tube_nodes = xpath.queryForNodes('/message/tube[@xmlns="%s"]' % NS_TUBES,
+ tube_nodes = xpath.queryForNodes('/message/tube[@xmlns="%s"]' % ns.TUBES,
message)
assert tube_nodes is not None
assert len(tube_nodes) == 1
@@ -463,20 +457,20 @@ def test(q, bus, conn, stream):
iq = IQ(stream, 'set')
iq['to'] = 'test at localhost/Resource'
iq['from'] = 'bob at localhost/Bob'
- si = iq.addElement((NS_SI, 'si'))
+ 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'))
+ 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)
+ value.addContent(ns.IBB)
- stream_node = si.addElement((NS_TUBES, 'stream'))
+ stream_node = si.addElement((ns.TUBES, 'stream'))
stream_node['tube'] = str(stream_tube_id)
stream.send(iq)
@@ -485,13 +479,13 @@ def test(q, bus, conn, stream):
EventPattern('dbus-signal', signal='TubeStateChanged',
args=[stream_tube_id, 2])) # 2 == OPEN
iq = si_reply_event.stanza
- si = xpath.queryForNodes('/iq/si[@xmlns="%s"]' % NS_SI,
+ 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 str(proto) == ns.IBB
+ tube = xpath.queryForNodes('/si/tube[@xmlns="%s"]' % ns.TUBES, si)
assert len(tube) == 1
q.expect('dbus-signal', signal='StreamTubeNewConnection',
@@ -506,20 +500,20 @@ def test(q, bus, conn, stream):
iq = IQ(stream, 'set')
iq['to'] = 'test at localhost/Resource'
iq['from'] = 'bob at localhost/Bob'
- si = iq.addElement((NS_SI, 'si'))
+ si = iq.addElement((ns.SI, 'si'))
si['id'] = 'beta'
- si['profile'] = NS_TUBES
- feature = si.addElement((NS_FEATURE_NEG, 'feature'))
- x = feature.addElement((NS_X_DATA, 'x'))
+ 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)
+ value.addContent(ns.IBB)
- stream_node = si.addElement((NS_TUBES, 'stream'))
+ stream_node = si.addElement((ns.TUBES, 'stream'))
stream_node['tube'] = str(new_stream_tube_id)
stream.send(iq)
@@ -528,13 +522,13 @@ def test(q, bus, conn, stream):
EventPattern('dbus-signal', signal='TubeChannelStateChanged',
args=[2])) # 2 == OPEN
iq = si_reply_event.stanza
- si = xpath.queryForNodes('/iq/si[@xmlns="%s"]' % NS_SI,
+ 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 str(proto) == ns.IBB
+ tube = xpath.queryForNodes('/si/tube[@xmlns="%s"]' % ns.TUBES, si)
assert len(tube) == 1
q.expect('dbus-signal', signal='StreamTubeNewConnection',
@@ -549,7 +543,7 @@ def test(q, bus, conn, stream):
iq = IQ(stream, 'set')
iq['to'] = 'test at localhost/Resource'
iq['from'] = 'bob at localhost/Bob'
- open = iq.addElement((NS_IBB, 'open'))
+ open = iq.addElement((ns.IBB, 'open'))
open['sid'] = 'alpha'
open['block-size'] = '4096'
stream.send(iq)
@@ -559,7 +553,7 @@ def test(q, bus, conn, stream):
message = domish.Element(('jabber:client', 'message'))
message['to'] = 'test at localhost/Resource'
message['from'] = 'bob at localhost/Bob'
- data_node = message.addElement((NS_IBB, 'data'))
+ data_node = message.addElement((ns.IBB, 'data'))
data_node['sid'] = 'alpha'
data_node['seq'] = '0'
data_node.addContent(base64.b64encode('hello, world'))
@@ -568,7 +562,7 @@ def test(q, bus, conn, stream):
event = q.expect('stream-message', to='bob at localhost/Bob')
message = event.stanza
- data_nodes = xpath.queryForNodes('/message/data[@xmlns="%s"]' % NS_IBB,
+ data_nodes = xpath.queryForNodes('/message/data[@xmlns="%s"]' % ns.IBB,
message)
assert data_nodes is not None
assert len(data_nodes) == 1
@@ -582,7 +576,7 @@ def test(q, bus, conn, stream):
iq = IQ(stream, 'set')
iq['to'] = 'test at localhost/Resource'
iq['from'] = 'bob at localhost/Bob'
- open = iq.addElement((NS_IBB, 'open'))
+ open = iq.addElement((ns.IBB, 'open'))
open['sid'] = 'beta'
open['block-size'] = '4096'
stream.send(iq)
@@ -592,7 +586,7 @@ def test(q, bus, conn, stream):
message = domish.Element(('jabber:client', 'message'))
message['to'] = 'test at localhost/Resource'
message['from'] = 'bob at localhost/Bob'
- data_node = message.addElement((NS_IBB, 'data'))
+ data_node = message.addElement((ns.IBB, 'data'))
data_node['sid'] = 'beta'
data_node['seq'] = '0'
data_node.addContent(base64.b64encode('hello, new world'))
@@ -601,7 +595,7 @@ def test(q, bus, conn, stream):
event = q.expect('stream-message', to='bob at localhost/Bob')
message = event.stanza
- data_nodes = xpath.queryForNodes('/message/data[@xmlns="%s"]' % NS_IBB,
+ data_nodes = xpath.queryForNodes('/message/data[@xmlns="%s"]' % ns.IBB,
message)
assert data_nodes is not None
assert len(data_nodes) == 1
@@ -619,7 +613,7 @@ def test(q, bus, conn, stream):
assert si_nodes is not None
assert len(si_nodes) == 1
si = si_nodes[0]
- assert si['profile'] == NS_TUBES
+ assert si['profile'] == ns.TUBES
dbus_stream_id = si['id']
feature = xpath.queryForNodes('/si/feature', si)[0]
@@ -629,9 +623,9 @@ def test(q, bus, conn, stream):
assert field['var'] == 'stream-method'
assert field['type'] == 'list-single'
value = xpath.queryForNodes('/field/option/value', field)[0]
- assert str(value) == NS_BYTESTREAMS
+ assert str(value) == ns.BYTESTREAMS
value = xpath.queryForNodes('/field/option/value', field)[1]
- assert str(value) == NS_IBB
+ assert str(value) == ns.IBB
tube = xpath.queryForNodes('/si/tube', si)[0]
assert tube['initiator'] == 'test at localhost'
@@ -656,21 +650,21 @@ def test(q, bus, conn, stream):
result['id'] = iq['id']
result['from'] = iq['to']
result['to'] = 'test at localhost/Resource'
- res_si = result.addElement((NS_SI, 'si'))
- res_feature = res_si.addElement((NS_FEATURE_NEG, 'feature'))
- res_x = res_feature.addElement((NS_X_DATA, 'x'))
+ res_si = result.addElement((ns.SI, 'si'))
+ res_feature = res_si.addElement((ns.FEATURE_NEG, 'feature'))
+ res_x = res_feature.addElement((ns.X_DATA, 'x'))
res_x['type'] = 'submit'
res_field = res_x.addElement((None, 'field'))
res_field['var'] = 'stream-method'
res_value = res_field.addElement((None, 'value'))
- res_value.addContent(NS_IBB)
+ res_value.addContent(ns.IBB)
stream.send(result)
event = q.expect('stream-iq', iq_type='set', to='bob at localhost/Bob')
iq = event.stanza
open = xpath.queryForNodes('/iq/open', iq)[0]
- assert open.uri == NS_IBB
+ assert open.uri == ns.IBB
assert open['sid'] == dbus_stream_id
result = IQ(stream, 'result')
@@ -705,7 +699,7 @@ def test(q, bus, conn, stream):
assert message['to'] == 'bob at localhost/Bob'
- data_nodes = xpath.queryForNodes('/message/data[@xmlns="%s"]' % NS_IBB,
+ data_nodes = xpath.queryForNodes('/message/data[@xmlns="%s"]' % ns.IBB,
message)
assert data_nodes is not None
assert len(data_nodes) == 1
@@ -732,7 +726,7 @@ def test(q, bus, conn, stream):
msg = domish.Element(('jabber:client', 'message'))
msg['to'] = 'test at localhost/Resource'
msg['from'] = 'bob at localhost/Bob'
- data_node = msg.addElement('data', NS_IBB)
+ data_node = msg.addElement('data', ns.IBB)
data_node['sid'] = dbus_stream_id
data_node['seq'] = str(seq)
data_node.addContent(base64.b64encode(dbus_message))
@@ -745,7 +739,7 @@ def test(q, bus, conn, stream):
msg = domish.Element(('jabber:client', 'message'))
msg['to'] = 'test at localhost/Resource'
msg['from'] = 'bob at localhost/Bob'
- data_node = msg.addElement('data', NS_IBB)
+ data_node = msg.addElement('data', ns.IBB)
data_node['sid'] = dbus_stream_id
data_node['seq'] = str(seq)
data_node.addContent(base64.b64encode(byte))
@@ -757,7 +751,7 @@ def test(q, bus, conn, stream):
msg = domish.Element(('jabber:client', 'message'))
msg['to'] = 'test at localhost/Resource'
msg['from'] = 'bob at localhost/Bob'
- data_node = msg.addElement('data', NS_IBB)
+ data_node = msg.addElement('data', ns.IBB)
data_node['sid'] = dbus_stream_id
data_node['seq'] = str(seq)
data_node.addContent(base64.b64encode(dbus_message + dbus_message))
@@ -793,20 +787,20 @@ def test(q, bus, conn, stream):
event = q.expect('stream-iq', iq_type='result')
iq = event.stanza
- si = xpath.queryForNodes('/iq/si[@xmlns="%s"]' % NS_SI,
+ 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 str(proto) == ns.IBB
+ tube = xpath.queryForNodes('/si/tube[@xmlns="%s"]' % ns.TUBES, si)
assert len(tube) == 1
# Init the IBB bytestream
iq = IQ(stream, 'set')
iq['to'] = 'test at localhost/Resource'
iq['from'] = 'bob at localhost/Bob'
- open = iq.addElement((NS_IBB, 'open'))
+ open = iq.addElement((ns.IBB, 'open'))
open['sid'] = 'beta'
open['block-size'] = '4096'
stream.send(iq)
@@ -851,20 +845,20 @@ def test(q, bus, conn, stream):
event = q.expect('stream-iq', iq_type='result')
iq = event.stanza
- si = xpath.queryForNodes('/iq/si[@xmlns="%s"]' % NS_SI,
+ 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 str(proto) == ns.IBB
+ tube = xpath.queryForNodes('/si/tube[@xmlns="%s"]' % ns.TUBES, si)
assert len(tube) == 1
# Init the IBB bytestream
iq = IQ(stream, 'set')
iq['to'] = 'test at localhost/Resource'
iq['from'] = 'bob at localhost/Bob'
- open = iq.addElement((NS_IBB, 'open'))
+ open = iq.addElement((ns.IBB, 'open'))
open['sid'] = 'gamma'
open['block-size'] = '4096'
stream.send(iq)
diff --git a/tests/twisted/tubes/offer-accept-private-dbus-stream-tube-socks5.py b/tests/twisted/tubes/offer-accept-private-dbus-stream-tube-socks5.py
index d4ee59b..f4558ce 100644
--- a/tests/twisted/tubes/offer-accept-private-dbus-stream-tube-socks5.py
+++ b/tests/twisted/tubes/offer-accept-private-dbus-stream-tube-socks5.py
@@ -13,6 +13,7 @@ from servicetest import call_async, EventPattern, tp_name_prefix, \
watch_tube_signals, sync_dbus
from gabbletest import exec_test, acknowledge_iq, sync_stream
from constants import *
+import ns
import tubetestutil as t
from twisted.words.xish import domish, xpath
@@ -20,13 +21,6 @@ from twisted.internet.protocol import Factory, Protocol
from twisted.internet import reactor
from twisted.words.protocols.jabber.client import IQ
-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_X_DATA = 'jabber:x:data'
-NS_BYTESTREAMS = 'http://jabber.org/protocol/bytestreams'
-
sample_parameters = dbus.Dictionary({
's': 'hello',
'ay': dbus.ByteArray('hello'),
@@ -241,7 +235,7 @@ def test(q, bus, conn, stream):
assert event.query['node'] == \
'http://example.com/ICantBelieveItsNotTelepathy#1.2.3'
feature = event.query.addElement('feature')
- feature['var'] = NS_TUBES
+ feature['var'] = ns.TUBES
stream.send(result)
# A tube request can be done only if the contact has tube capabilities
@@ -349,7 +343,7 @@ def test(q, bus, conn, stream):
event = q.expect('stream-message')
message = event.stanza
assert message['to'] == 'bob at localhost/Bob' # check the resource
- tube_nodes = xpath.queryForNodes('/message/tube[@xmlns="%s"]' % NS_TUBES,
+ tube_nodes = xpath.queryForNodes('/message/tube[@xmlns="%s"]' % ns.TUBES,
message)
assert tube_nodes is not None
assert len(tube_nodes) == 1
@@ -431,7 +425,7 @@ def test(q, bus, conn, stream):
event = q.expect('stream-message')
message = event.stanza
assert message['to'] == 'bob at localhost/Bob' # check the resource
- tube_nodes = xpath.queryForNodes('/message/tube[@xmlns="%s"]' % NS_TUBES,
+ tube_nodes = xpath.queryForNodes('/message/tube[@xmlns="%s"]' % ns.TUBES,
message)
assert tube_nodes is not None
assert len(tube_nodes) == 1
@@ -475,20 +469,20 @@ def test(q, bus, conn, stream):
iq = IQ(stream, 'set')
iq['to'] = 'test at localhost/Resource'
iq['from'] = 'bob at localhost/Bob'
- si = iq.addElement((NS_SI, 'si'))
+ 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'))
+ 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_BYTESTREAMS)
+ value.addContent(ns.BYTESTREAMS)
- stream_node = si.addElement((NS_TUBES, 'stream'))
+ stream_node = si.addElement((ns.TUBES, 'stream'))
stream_node['tube'] = str(stream_tube_id)
stream.send(iq)
@@ -497,13 +491,13 @@ def test(q, bus, conn, stream):
EventPattern('dbus-signal', signal='TubeStateChanged',
args=[stream_tube_id, 2])) # 2 == OPEN
iq = si_reply_event.stanza
- si = xpath.queryForNodes('/iq/si[@xmlns="%s"]' % NS_SI,
+ 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_BYTESTREAMS
- tube = xpath.queryForNodes('/si/tube[@xmlns="%s"]' % NS_TUBES, si)
+ assert str(proto) == ns.BYTESTREAMS
+ tube = xpath.queryForNodes('/si/tube[@xmlns="%s"]' % ns.TUBES, si)
assert len(tube) == 1
q.expect('dbus-signal', signal='StreamTubeNewConnection',
@@ -519,20 +513,20 @@ def test(q, bus, conn, stream):
iq = IQ(stream, 'set')
iq['to'] = 'test at localhost/Resource'
iq['from'] = 'bob at localhost/Bob'
- si = iq.addElement((NS_SI, 'si'))
+ si = iq.addElement((ns.SI, 'si'))
si['id'] = 'beta'
- si['profile'] = NS_TUBES
- feature = si.addElement((NS_FEATURE_NEG, 'feature'))
- x = feature.addElement((NS_X_DATA, 'x'))
+ 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_BYTESTREAMS)
+ value.addContent(ns.BYTESTREAMS)
- stream_node = si.addElement((NS_TUBES, 'stream'))
+ stream_node = si.addElement((ns.TUBES, 'stream'))
stream_node['tube'] = str(new_stream_tube_id)
stream.send(iq)
@@ -541,13 +535,13 @@ def test(q, bus, conn, stream):
EventPattern('dbus-signal', signal='TubeChannelStateChanged',
args=[2])) # 2 == OPEN
iq = si_reply_event.stanza
- si = xpath.queryForNodes('/iq/si[@xmlns="%s"]' % NS_SI,
+ 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_BYTESTREAMS
- tube = xpath.queryForNodes('/si/tube[@xmlns="%s"]' % NS_TUBES, si)
+ assert str(proto) == ns.BYTESTREAMS
+ tube = xpath.queryForNodes('/si/tube[@xmlns="%s"]' % ns.TUBES, si)
assert len(tube) == 1
q.expect('dbus-signal', signal='StreamTubeNewConnection',
@@ -570,7 +564,7 @@ def test(q, bus, conn, stream):
iq = IQ(stream, 'set')
iq['to'] = 'test at localhost/Resource'
iq['from'] = 'bob at localhost/Bob'
- query = iq.addElement((NS_BYTESTREAMS, 'query'))
+ query = iq.addElement((ns.BYTESTREAMS, 'query'))
query['sid'] = 'alpha'
query['mode'] = 'tcp'
# Not working streamhost
@@ -610,7 +604,7 @@ def test(q, bus, conn, stream):
event = q.expect('stream-iq', iq_type='result')
iq = event.stanza
query = xpath.queryForNodes('/iq/query', iq)[0]
- assert query.uri == NS_BYTESTREAMS
+ assert query.uri == ns.BYTESTREAMS
streamhost_used = xpath.queryForNodes('/query/streamhost-used', query)[0]
assert streamhost_used['jid'] == 'bob at localhost/Bob'
@@ -628,7 +622,7 @@ def test(q, bus, conn, stream):
iq = IQ(stream, 'set')
iq['to'] = 'test at localhost/Resource'
iq['from'] = 'bob at localhost/Bob'
- query = iq.addElement((NS_BYTESTREAMS, 'query'))
+ query = iq.addElement((ns.BYTESTREAMS, 'query'))
query['sid'] = 'beta'
query['mode'] = 'tcp'
# Not working streamhost
@@ -668,7 +662,7 @@ def test(q, bus, conn, stream):
event = q.expect('stream-iq', iq_type='result')
iq = event.stanza
query = xpath.queryForNodes('/iq/query', iq)[0]
- assert query.uri == NS_BYTESTREAMS
+ assert query.uri == ns.BYTESTREAMS
streamhost_used = xpath.queryForNodes('/query/streamhost-used', query)[0]
assert streamhost_used['jid'] == 'bob at localhost/Bob'
@@ -686,7 +680,7 @@ def test(q, bus, conn, stream):
assert si_nodes is not None
assert len(si_nodes) == 1
si = si_nodes[0]
- assert si['profile'] == NS_TUBES
+ assert si['profile'] == ns.TUBES
dbus_stream_id = si['id']
feature = xpath.queryForNodes('/si/feature', si)[0]
@@ -696,9 +690,9 @@ def test(q, bus, conn, stream):
assert field['var'] == 'stream-method'
assert field['type'] == 'list-single'
value = xpath.queryForNodes('/field/option/value', field)[0]
- assert str(value) == NS_BYTESTREAMS
+ assert str(value) == ns.BYTESTREAMS
value = xpath.queryForNodes('/field/option/value', field)[1]
- assert str(value) == NS_IBB
+ assert str(value) == ns.IBB
tube = xpath.queryForNodes('/si/tube', si)[0]
assert tube['initiator'] == 'test at localhost'
@@ -723,21 +717,21 @@ def test(q, bus, conn, stream):
result['id'] = iq['id']
result['from'] = iq['to']
result['to'] = 'test at localhost/Resource'
- res_si = result.addElement((NS_SI, 'si'))
- res_feature = res_si.addElement((NS_FEATURE_NEG, 'feature'))
- res_x = res_feature.addElement((NS_X_DATA, 'x'))
+ res_si = result.addElement((ns.SI, 'si'))
+ res_feature = res_si.addElement((ns.FEATURE_NEG, 'feature'))
+ res_x = res_feature.addElement((ns.X_DATA, 'x'))
res_x['type'] = 'submit'
res_field = res_x.addElement((None, 'field'))
res_field['var'] = 'stream-method'
res_value = res_field.addElement((None, 'value'))
- res_value.addContent(NS_BYTESTREAMS)
+ res_value.addContent(ns.BYTESTREAMS)
stream.send(result)
event = q.expect('stream-iq', iq_type='set', to='bob at localhost/Bob')
iq = event.stanza
query = xpath.queryForNodes('/iq/query', iq)[0]
- assert query.uri == NS_BYTESTREAMS
+ assert query.uri == ns.BYTESTREAMS
assert query['mode'] == 'tcp'
assert query['sid'] == dbus_stream_id
streamhost = xpath.queryForNodes('/query/streamhost', query)[0]
@@ -825,20 +819,20 @@ def test(q, bus, conn, stream):
iq = IQ(stream, 'set')
iq['to'] = 'test at localhost/Resource'
iq['from'] = 'bob at localhost/Bob'
- si = iq.addElement((NS_SI, 'si'))
+ si = iq.addElement((ns.SI, 'si'))
si['id'] = 'beta'
- si['profile'] = NS_TUBES
- feature = si.addElement((NS_FEATURE_NEG, 'feature'))
- x = feature.addElement((NS_X_DATA, 'x'))
+ 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_BYTESTREAMS)
+ value.addContent(ns.BYTESTREAMS)
- tube = si.addElement((NS_TUBES, 'tube'))
+ tube = si.addElement((ns.TUBES, 'tube'))
tube['type'] = 'dbus'
tube['service'] = 'com.example.TestCase2'
tube['id'] = '69'
@@ -871,13 +865,13 @@ def test(q, bus, conn, stream):
event = q.expect('stream-iq', iq_type='result')
iq = event.stanza
- si = xpath.queryForNodes('/iq/si[@xmlns="%s"]' % NS_SI,
+ 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_BYTESTREAMS
- tube = xpath.queryForNodes('/si/tube[@xmlns="%s"]' % NS_TUBES, si)
+ assert str(proto) == ns.BYTESTREAMS
+ tube = xpath.queryForNodes('/si/tube[@xmlns="%s"]' % ns.TUBES, si)
assert len(tube) == 1
reactor.listenTCP(5084, S5BFactory(q.append))
@@ -886,7 +880,7 @@ def test(q, bus, conn, stream):
iq = IQ(stream, 'set')
iq['to'] = 'test at localhost/Resource'
iq['from'] = 'bob at localhost/Bob'
- query = iq.addElement((NS_BYTESTREAMS, 'query'))
+ query = iq.addElement((ns.BYTESTREAMS, 'query'))
query['sid'] = 'beta'
query['mode'] = 'tcp'
streamhost = query.addElement('streamhost')
diff --git a/tests/twisted/tubes/offer-accept-private-stream-tube-si-fallback.py b/tests/twisted/tubes/offer-accept-private-stream-tube-si-fallback.py
index dd78d6b..25e7615 100644
--- a/tests/twisted/tubes/offer-accept-private-stream-tube-si-fallback.py
+++ b/tests/twisted/tubes/offer-accept-private-stream-tube-si-fallback.py
@@ -9,6 +9,7 @@ from dbus.lowlevel import SignalMessage
from servicetest import call_async, EventPattern, tp_name_prefix, watch_tube_signals, EventProtocolClientFactory
from gabbletest import exec_test, acknowledge_iq
+import ns
from twisted.words.xish import domish, xpath
from twisted.internet import reactor
@@ -16,14 +17,6 @@ from twisted.words.protocols.jabber.client import IQ
import tubetestutil as t
-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_X_DATA = 'jabber:x:data'
-NS_BYTESTREAMS = 'http://jabber.org/protocol/bytestreams'
-NS_SI_MULTIPLE = 'http://telepathy.freedesktop.org/xmpp/si-multiple'
-
def test(q, bus, conn, stream):
t.set_up_echo('')
@@ -61,7 +54,7 @@ def test(q, bus, conn, stream):
assert event.query['node'] == \
'http://example.com/ICantBelieveItsNotTelepathy#1.2.3'
feature = event.query.addElement('feature')
- feature['var'] = NS_TUBES
+ feature['var'] = ns.TUBES
stream.send(result)
requestotron = dbus.Interface(conn,
@@ -102,7 +95,7 @@ def test(q, bus, conn, stream):
event = q.expect('stream-message')
message = event.stanza
- tube_nodes = xpath.queryForNodes('/message/tube[@xmlns="%s"]' % NS_TUBES,
+ tube_nodes = xpath.queryForNodes('/message/tube[@xmlns="%s"]' % ns.TUBES,
message)
assert tube_nodes is not None
assert len(tube_nodes) == 1
@@ -117,30 +110,30 @@ def test(q, bus, conn, stream):
iq = IQ(stream, 'set')
iq['to'] = 'test at localhost/Resource'
iq['from'] = 'bob at localhost/Bob'
- si = iq.addElement((NS_SI, 'si'))
+ 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'))
+ 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_BYTESTREAMS)
+ value.addContent(ns.BYTESTREAMS)
option = field.addElement((None, 'option'))
value = option.addElement((None, 'value'))
value.addContent("invalid-stream-method")
option = field.addElement((None, 'option'))
value = option.addElement((None, 'value'))
- value.addContent(NS_IBB)
+ value.addContent(ns.IBB)
- stream_node = si.addElement((NS_TUBES, 'stream'))
+ stream_node = si.addElement((ns.TUBES, 'stream'))
stream_node['tube'] = str(stream_tube_id)
# Bob supports multi bytestreams
- si_multiple = si.addElement((NS_SI_MULTIPLE, 'si-multiple'))
+ si_multiple = si.addElement((ns.SI_MULTIPLE, 'si-multiple'))
stream.send(iq)
@@ -152,14 +145,14 @@ def test(q, bus, conn, stream):
iq = si_reply_event.stanza
# check if SI reply contains the 2 bytestreams
methods = xpath.queryForNodes('/iq/si[@xmlns="%s"]/si-multiple[@xmlns="%s"]/value' %
- (NS_SI, NS_SI_MULTIPLE), iq)
+ (ns.SI, ns.SI_MULTIPLE), iq)
assert len(methods) == 2
assert methods[0].name == 'value'
- assert str(methods[0]) == NS_BYTESTREAMS
+ assert str(methods[0]) == ns.BYTESTREAMS
assert methods[1].name == 'value'
- assert str(methods[1]) == NS_IBB
+ assert str(methods[1]) == ns.IBB
tube = xpath.queryForNodes('/iq/si[@xmlns="%s"]/tube[@xmlns="%s"]' %
- (NS_SI, NS_TUBES), iq)
+ (ns.SI, ns.TUBES), iq)
assert len(tube) == 1
q.expect('dbus-signal', signal='StreamTubeNewConnection',
@@ -170,7 +163,7 @@ def test(q, bus, conn, stream):
iq = IQ(stream, 'set')
iq['to'] = 'test at localhost/Resource'
iq['from'] = 'bob at localhost/Bob'
- query = iq.addElement((NS_BYTESTREAMS, 'query'))
+ query = iq.addElement((ns.BYTESTREAMS, 'query'))
query['sid'] = 'alpha'
query['mode'] = 'tcp'
streamhost = query.addElement('streamhost')
@@ -186,7 +179,7 @@ def test(q, bus, conn, stream):
iq = IQ(stream, 'set')
iq['to'] = 'test at localhost/Resource'
iq['from'] = 'bob at localhost/Bob'
- open = iq.addElement((NS_IBB, 'open'))
+ open = iq.addElement((ns.IBB, 'open'))
open['sid'] = 'alpha'
open['block-size'] = '4096'
stream.send(iq)
@@ -198,7 +191,7 @@ def test(q, bus, conn, stream):
message = domish.Element(('jabber:client', 'message'))
message['to'] = 'test at localhost/Resource'
message['from'] = 'bob at localhost/Bob'
- data_node = message.addElement((NS_IBB, 'data'))
+ data_node = message.addElement((ns.IBB, 'data'))
data_node['sid'] = 'alpha'
data_node['seq'] = '0'
data_node.addContent(base64.b64encode('HELLO, WORLD'))
@@ -207,7 +200,7 @@ def test(q, bus, conn, stream):
event = q.expect('stream-message', to='bob at localhost/Bob')
message = event.stanza
- data_nodes = xpath.queryForNodes('/message/data[@xmlns="%s"]' % NS_IBB,
+ data_nodes = xpath.queryForNodes('/message/data[@xmlns="%s"]' % ns.IBB,
message)
assert data_nodes is not None
assert len(data_nodes) == 1
@@ -222,7 +215,7 @@ def test(q, bus, conn, stream):
message['to'] = 'test at localhost/Resource'
message['from'] = 'bob at localhost/Bob'
message['id'] = 'msg-id'
- tube = message.addElement((NS_TUBES, 'tube'))
+ tube = message.addElement((ns.TUBES, 'tube'))
tube['type'] = 'stream'
tube['id'] = '42'
tube['service'] = 'foo-service'
@@ -269,7 +262,7 @@ def test(q, bus, conn, stream):
assert si_nodes is not None
assert len(si_nodes) == 1
si = si_nodes[0]
- assert si['profile'] == NS_TUBES
+ assert si['profile'] == ns.TUBES
dbus_stream_id = si['id']
feature = xpath.queryForNodes('/si/feature', si)[0]
@@ -279,26 +272,26 @@ def test(q, bus, conn, stream):
assert field['var'] == 'stream-method'
assert field['type'] == 'list-single'
value = xpath.queryForNodes('/field/option/value', field)[0]
- assert str(value) == NS_BYTESTREAMS
+ assert str(value) == ns.BYTESTREAMS
value = xpath.queryForNodes('/field/option/value', field)[1]
- assert str(value) == NS_IBB
+ assert str(value) == ns.IBB
# Gabble supports multi-bytestreams extension
si_multiple = xpath.queryForNodes('/si/si-multiple', si)[0]
- assert si_multiple.uri == NS_SI_MULTIPLE
+ assert si_multiple.uri == ns.SI_MULTIPLE
result = IQ(stream, 'result')
result['id'] = iq['id']
result['from'] = iq['to']
result['to'] = 'test at localhost/Resource'
- res_si = result.addElement((NS_SI, 'si'))
+ res_si = result.addElement((ns.SI, 'si'))
# reply using multi-bytestreams extension
- res_multi = res_si.addElement((NS_SI_MULTIPLE, 'si-multiple'))
+ res_multi = res_si.addElement((ns.SI_MULTIPLE, 'si-multiple'))
res_value = res_multi.addElement(('', 'value'))
res_value.addContent('invalid-stream-method')
res_value = res_multi.addElement(('', 'value'))
- res_value.addContent(NS_BYTESTREAMS)
+ res_value.addContent(ns.BYTESTREAMS)
res_value = res_multi.addElement(('', 'value'))
- res_value.addContent(NS_IBB)
+ res_value.addContent(ns.IBB)
stream.send(result)
@@ -306,7 +299,7 @@ def test(q, bus, conn, stream):
event = q.expect('stream-iq', iq_type='set', to='bob at localhost/Bob')
iq = event.stanza
query = xpath.queryForNodes('/iq/query', iq)[0]
- assert query.uri == NS_BYTESTREAMS
+ assert query.uri == ns.BYTESTREAMS
sid = query['sid']
streamhost = xpath.queryForNodes('/iq/query/streamhost', iq)[0]
assert streamhost
@@ -326,7 +319,7 @@ def test(q, bus, conn, stream):
event = q.expect('stream-iq', iq_type='set', to='bob at localhost/Bob')
iq = event.stanza
open = xpath.queryForNodes('/iq/open', iq)[0]
- assert open.uri == NS_IBB
+ assert open.uri == ns.IBB
sid = open['sid']
# IBB is working
diff --git a/tests/twisted/tubes/offer-muc-dbus-tube.py b/tests/twisted/tubes/offer-muc-dbus-tube.py
index e708ca5..60d5dd3 100644
--- a/tests/twisted/tubes/offer-muc-dbus-tube.py
+++ b/tests/twisted/tubes/offer-muc-dbus-tube.py
@@ -9,6 +9,7 @@ from dbus.lowlevel import SignalMessage
from servicetest import call_async, EventPattern, tp_name_prefix
from gabbletest import exec_test, make_result_iq, acknowledge_iq
from constants import *
+import ns
import tubetestutil as t
from twisted.words.xish import domish, xpath
@@ -22,13 +23,6 @@ sample_parameters = dbus.Dictionary({
'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 test(q, bus, conn, stream):
conn.Connect()
@@ -108,7 +102,7 @@ def test(q, bus, conn, stream):
assert len(x_nodes) == 1
tubes_nodes = xpath.queryForNodes('/presence/tubes[@xmlns="%s"]'
- % NS_TUBES, presence)
+ % ns.TUBES, presence)
assert tubes_nodes is not None
assert len(tubes_nodes) == 1
@@ -158,7 +152,7 @@ def test(q, bus, conn, stream):
message_type='groupchat')
message = event.stanza
- data_nodes = xpath.queryForNodes('/message/data[@xmlns="%s"]' % NS_MUC_BYTESTREAM,
+ data_nodes = xpath.queryForNodes('/message/data[@xmlns="%s"]' % ns.MUC_BYTESTREAM,
message)
assert data_nodes is not None
assert len(data_nodes) == 1
diff --git a/tests/twisted/tubes/offer-muc-stream-tube-ibb.py b/tests/twisted/tubes/offer-muc-stream-tube-ibb.py
index 7db278c..1ee6421 100644
--- a/tests/twisted/tubes/offer-muc-stream-tube-ibb.py
+++ b/tests/twisted/tubes/offer-muc-stream-tube-ibb.py
@@ -10,6 +10,7 @@ from dbus import PROPERTIES_IFACE
from servicetest import call_async, EventPattern, tp_name_prefix, EventProtocolFactory
from gabbletest import exec_test, make_result_iq, acknowledge_iq, make_muc_presence
from constants import *
+import ns
import tubetestutil as t
from twisted.words.xish import domish, xpath
@@ -23,13 +24,6 @@ sample_parameters = dbus.Dictionary({
'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, path):
factory = EventProtocolFactory(q)
full_path = os.getcwd() + path
@@ -157,7 +151,7 @@ def test(q, bus, conn, stream):
assert len(x_nodes) == 1
tubes_nodes = xpath.queryForNodes('/presence/tubes[@xmlns="%s"]'
- % NS_TUBES, presence)
+ % ns.TUBES, presence)
assert tubes_nodes is not None
assert len(tubes_nodes) == 1
@@ -224,20 +218,20 @@ def test(q, bus, conn, stream):
iq = IQ(stream, 'set')
iq['to'] = 'test at localhost/Resource'
iq['from'] = 'chat at conf.localhost/bob'
- si = iq.addElement((NS_SI, 'si'))
+ 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'))
+ 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)
+ value.addContent(ns.IBB)
- stream_node = si.addElement((NS_TUBES, 'muc-stream'))
+ stream_node = si.addElement((ns.TUBES, 'muc-stream'))
stream_node['tube'] = str(stream_tube_id)
stream.send(iq)
@@ -252,20 +246,20 @@ def test(q, bus, conn, stream):
# handle iq_event
iq = iq_event.stanza
- si = xpath.queryForNodes('/iq/si[@xmlns="%s"]' % NS_SI,
+ 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 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 = iq.addElement((ns.IBB, 'open'))
open['sid'] = 'alpha'
open['block-size'] = '4096'
stream.send(iq)
@@ -276,7 +270,7 @@ def test(q, bus, conn, stream):
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 = message.addElement((ns.IBB, 'data'))
data_node['sid'] = 'alpha'
data_node['seq'] = '0'
data_node.addContent(base64.b64encode('hello initiator'))
@@ -290,7 +284,7 @@ def test(q, bus, conn, stream):
event = q.expect('stream-message', to='chat at conf.localhost/bob')
message = event.stanza
- data_nodes = xpath.queryForNodes('/message/data[@xmlns="%s"]' % NS_IBB,
+ data_nodes = xpath.queryForNodes('/message/data[@xmlns="%s"]' % ns.IBB,
message)
assert data_nodes is not None
assert len(data_nodes) == 1
@@ -400,7 +394,7 @@ def test(q, bus, conn, stream):
assert len(x_nodes) == 1
tubes_nodes = xpath.queryForNodes('/presence/tubes[@xmlns="%s"]'
- % NS_TUBES, presence)
+ % ns.TUBES, presence)
assert tubes_nodes is not None
assert len(tubes_nodes) == 1
diff --git a/tests/twisted/tubes/test-get-available-tubes.py b/tests/twisted/tubes/test-get-available-tubes.py
index 0e86114..e7e9441 100644
--- a/tests/twisted/tubes/test-get-available-tubes.py
+++ b/tests/twisted/tubes/test-get-available-tubes.py
@@ -4,6 +4,7 @@ import dbus
from servicetest import call_async, EventPattern, tp_name_prefix
from gabbletest import exec_test, make_result_iq, acknowledge_iq
+import ns
from twisted.words.xish import domish
@@ -14,14 +15,6 @@ sample_parameters = dbus.Dictionary({
'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 test(q, bus, conn, stream):
conn.Connect()
--
1.5.6.5
More information about the telepathy-commits
mailing list