[Telepathy-commits] [telepathy-gabble/master] Extract checking ListTubes() to a helper function
Will Thompson
will.thompson at collabora.co.uk
Tue Feb 3 03:36:29 PST 2009
---
tests/twisted/Makefile.am | 3 +-
tests/twisted/tubes/accept-muc-stream-tube-ibb.py | 7 +++
tests/twisted/tubes/close-muc-with-closed-tube.py | 15 +++---
.../offer-accept-private-dbus-stream-tube-ibb.py | 48 +++++--------------
...offer-accept-private-dbus-stream-tube-socks5.py | 35 +++++----------
tests/twisted/tubes/offer-muc-dbus-tube.py | 13 ++---
tests/twisted/tubes/offer-muc-stream-tube-ibb.py | 6 +++
tests/twisted/tubes/offer-private-dbus-tube-ibb.py | 13 ++---
tests/twisted/tubes/tubetestutil.py | 29 ++++++++++++
9 files changed, 85 insertions(+), 84 deletions(-)
create mode 100644 tests/twisted/tubes/tubetestutil.py
diff --git a/tests/twisted/Makefile.am b/tests/twisted/Makefile.am
index e8eddfa..ad65846 100644
--- a/tests/twisted/Makefile.am
+++ b/tests/twisted/Makefile.am
@@ -143,7 +143,8 @@ EXTRA_DIST = \
jingle/jingletest2.py \
ns.py \
olpc/util.py \
- tubes/muctubeutil.py
+ tubes/muctubeutil.py \
+ tubes/tubetestutil.py
noinst_PROGRAMS = \
telepathy-gabble-debug
diff --git a/tests/twisted/tubes/accept-muc-stream-tube-ibb.py b/tests/twisted/tubes/accept-muc-stream-tube-ibb.py
index a30c660..f98b62d 100644
--- a/tests/twisted/tubes/accept-muc-stream-tube-ibb.py
+++ b/tests/twisted/tubes/accept-muc-stream-tube-ibb.py
@@ -5,6 +5,8 @@ import dbus
from servicetest import call_async, EventPattern, EventProtocolClientFactory
from gabbletest import exec_test, make_result_iq, acknowledge_iq
+from constants import *
+from tubetestutil import *
from twisted.words.xish import domish, xpath
from twisted.internet import reactor
@@ -158,6 +160,8 @@ def test(q, bus, conn, stream):
q.expect('dbus-signal', signal='NewTube',
args=[stream_tube_id, bob_handle, 1, 'echo', sample_parameters, 0])
+ expected_tube = (stream_tube_id, bob_handle, TUBE_TYPE_STREAM, 'echo',
+ sample_parameters, TUBE_STATE_LOCAL_PENDING)
tubes = tubes_iface.ListTubes(byte_arrays=True)
assert tubes == [(
stream_tube_id,
@@ -168,6 +172,9 @@ def test(q, bus, conn, stream):
TUBE_CHANNEL_STATE_LOCAL_PENDING
)]
+ assert len(tubes) == 1, unwrap(tubes)
+ check_tube_in_tubes(expected_tube, tubes)
+
# tube channel is also announced (new API)
new_event = q.expect('dbus-signal', signal='NewChannels')
diff --git a/tests/twisted/tubes/close-muc-with-closed-tube.py b/tests/twisted/tubes/close-muc-with-closed-tube.py
index 128a9c5..ee09641 100644
--- a/tests/twisted/tubes/close-muc-with-closed-tube.py
+++ b/tests/twisted/tubes/close-muc-with-closed-tube.py
@@ -5,6 +5,8 @@ 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 *
+from tubetestutil import *
from twisted.words.xish import domish, xpath
from twisted.internet import reactor
@@ -151,15 +153,12 @@ def test(q, bus, conn, stream):
q.expect('dbus-signal', signal='NewTube',
args=[tube_id, bob_handle, 0, 'org.telepathy.freedesktop.test', sample_parameters, 0])
+ expected_tube = (tube_id, bob_handle, TUBE_TYPE_DBUS,
+ 'org.telepathy.freedesktop.test', sample_parameters,
+ TUBE_STATE_LOCAL_PENDING)
tubes = tubes_iface.ListTubes(byte_arrays=True)
- assert tubes == [(
- tube_id,
- bob_handle,
- 0, # D-Bus
- 'org.telepathy.freedesktop.test',
- sample_parameters,
- 0, # local pending
- )]
+ assert len(tubes) == 1, unwrap(tubes)
+ check_tube_in_tubes(expected_tube, tubes)
# reject the tube
tubes_iface.CloseTube(tube_id)
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 12f534e..910e635 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
@@ -11,6 +11,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 *
+from tubetestutil import *
from dbus import PROPERTIES_IFACE
@@ -146,7 +147,6 @@ def check_NewChannels_signal(new_sig, channel_type, chan_path, contact_handle,
assert emitted_props[INITIATOR_HANDLE] == initiator_handle
assert emitted_props[INITIATOR_ID] == 'test at localhost'
-
def test(q, bus, conn, stream):
set_up_echo("")
set_up_echo("2")
@@ -534,15 +534,9 @@ def test(q, bus, conn, stream):
q.expect('dbus-signal', signal='StreamTubeNewConnection',
args=[stream_tube_id, bob_handle])
- tubes = tubes_iface.ListTubes(byte_arrays=True)
- assert (
- stream_tube_id,
- self_handle,
- 1, # Unix stream
- 'echo',
- sample_parameters,
- 2, # OPEN
- ) in tubes
+ expected_tube = (stream_tube_id, self_handle, TUBE_TYPE_STREAM, 'echo',
+ sample_parameters, TUBE_STATE_OPEN)
+ check_tube_in_tubes(expected_tube, tubes_iface.ListTubes(byte_arrays=True))
# The CM is the server, so fake a client wanting to talk to it
# New API tube
@@ -583,15 +577,9 @@ def test(q, bus, conn, stream):
q.expect('dbus-signal', signal='StreamTubeNewConnection',
args=[bob_handle])
- tubes = tubes_iface.ListTubes(byte_arrays=True)
- assert (
- new_stream_tube_id,
- self_handle,
- 1, # Unix stream
- 'newecho',
- new_sample_parameters,
- 2, # OPEN
- ) in tubes, tubes
+ expected_tube = (new_stream_tube_id, self_handle, TUBE_TYPE_STREAM,
+ 'newecho', new_sample_parameters, TUBE_STATE_OPEN)
+ check_tube_in_tubes (expected_tube, tubes_iface.ListTubes(byte_arrays=True))
# have the fake client open the stream
# Old tube API
@@ -733,22 +721,12 @@ def test(q, bus, conn, stream):
args=[dbus_tube_id, 2]) # 2 == OPEN
tubes = tubes_iface.ListTubes(byte_arrays=True)
- assert (
- dbus_tube_id,
- self_handle,
- 0, # DBUS
- 'com.example.TestCase',
- sample_parameters,
- 2, # OPEN
- ) in tubes
- assert (
- stream_tube_id,
- self_handle,
- 1, # stream
- 'echo',
- sample_parameters,
- 2, # OPEN
- ) in tubes
+ expected_dtube = (dbus_tube_id, self_handle, TUBE_TYPE_DBUS,
+ 'com.example.TestCase', sample_parameters, TUBE_STATE_OPEN)
+ expected_stube = (stream_tube_id, self_handle, TUBE_TYPE_STREAM,
+ 'echo', sample_parameters, TUBE_STATE_OPEN)
+ check_tube_in_tubes(expected_dtube, tubes)
+ check_tube_in_tubes(expected_stube, tubes)
dbus_tube_adr = tubes_iface.GetDBusTubeAddress(dbus_tube_id)
dbus_tube_conn = Connection(dbus_tube_adr)
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 d2e8185..1b275fc 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
@@ -12,6 +12,8 @@ from dbus.lowlevel import SignalMessage
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 *
+from tubetestutil import *
from twisted.words.xish import domish, xpath
from twisted.internet.protocol import Factory, Protocol
@@ -571,15 +573,10 @@ def test(q, bus, conn, stream):
q.expect('dbus-signal', signal='StreamTubeNewConnection',
args=[stream_tube_id, bob_handle])
+ expected_tube = (stream_tube_id, self_handle, TUBE_TYPE_STREAM, 'echo',
+ sample_parameters, TUBE_STATE_OPEN)
tubes = tubes_iface.ListTubes(byte_arrays=True)
- assert (
- stream_tube_id,
- self_handle,
- 1, # Unix stream
- 'echo',
- sample_parameters,
- 2, # OPEN
- ) in tubes
+ check_tube_in_tubes(expected_tube, tubes)
# The CM is the server, so fake a client wanting to talk to it
# New API tube
@@ -841,22 +838,12 @@ def test(q, bus, conn, stream):
args=[dbus_tube_id, 2]) # 2 == OPEN
tubes = tubes_iface.ListTubes(byte_arrays=True)
- assert (
- dbus_tube_id,
- self_handle,
- 0, # DBUS
- 'com.example.TestCase',
- sample_parameters,
- 2, # OPEN
- ) in tubes
- assert (
- stream_tube_id,
- self_handle,
- 1, # stream
- 'echo',
- sample_parameters,
- 2, # OPEN
- ) in tubes
+ expected_dtube = (dbus_tube_id, self_handle, TUBE_TYPE_DBUS,
+ 'com.example.TestCase', sample_parameters, TUBE_STATE_OPEN)
+ expected_stube = (stream_tube_id, self_handle, TUBE_TYPE_STREAM,
+ 'echo', sample_parameters, TUBE_STATE_OPEN)
+ check_tube_in_tubes(expected_dtube, tubes)
+ check_tube_in_tubes(expected_stube, tubes)
dbus_tube_adr = tubes_iface.GetDBusTubeAddress(dbus_tube_id)
dbus_tube_conn = Connection(dbus_tube_adr)
diff --git a/tests/twisted/tubes/offer-muc-dbus-tube.py b/tests/twisted/tubes/offer-muc-dbus-tube.py
index 9ffa237..33b6a3f 100644
--- a/tests/twisted/tubes/offer-muc-dbus-tube.py
+++ b/tests/twisted/tubes/offer-muc-dbus-tube.py
@@ -8,6 +8,8 @@ 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 *
+from tubetestutil import *
from twisted.words.xish import domish, xpath
@@ -142,14 +144,9 @@ def test(q, bus, conn, stream):
tubes = tubes_iface.ListTubes(byte_arrays=True)
assert len(tubes) == 1
- sorted(tubes) == sorted([(
- dbus_tube_id,
- tubes_self_handle,
- 0, # DBUS
- 'com.example.TestCase',
- sample_parameters,
- 2, # OPEN
- )])
+ expected_tube = (dbus_tube_id, tubes_self_handle, TUBE_TYPE_DBUS,
+ 'com.example.TestCase', sample_parameters, TUBE_STATE_OPEN)
+ check_tube_in_tubes(expected_tube, tubes)
dbus_tube_adr = tubes_iface.GetDBusTubeAddress(dbus_tube_id)
tube = Connection(dbus_tube_adr)
diff --git a/tests/twisted/tubes/offer-muc-stream-tube-ibb.py b/tests/twisted/tubes/offer-muc-stream-tube-ibb.py
index 2dc3701..c8b1ab0 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 *
+from tubetestutil import *
from twisted.words.xish import domish, xpath
from twisted.internet import reactor
@@ -211,6 +212,11 @@ def test(q, bus, conn, stream):
TUBE_CHANNEL_STATE_OPEN
)]
+ assert len(tubes) == 1, unwrap(tubes)
+ expected_tube = (stream_tube_id, tubes_self_handle, TUBE_TYPE_STREAM,
+ 'echo', sample_parameters, TUBE_STATE_OPEN)
+ check_tube_in_tubes(expected_tube, tubes)
+
# FIXME: if we use an unknown JID here, everything fails
# (the code uses lookup where it should use ensure)
diff --git a/tests/twisted/tubes/offer-private-dbus-tube-ibb.py b/tests/twisted/tubes/offer-private-dbus-tube-ibb.py
index 16ba97e..3e61b65 100644
--- a/tests/twisted/tubes/offer-private-dbus-tube-ibb.py
+++ b/tests/twisted/tubes/offer-private-dbus-tube-ibb.py
@@ -8,6 +8,8 @@ from dbus.lowlevel import SignalMessage
from servicetest import call_async, EventPattern, tp_name_prefix
from gabbletest import exec_test, make_result_iq, acknowledge_iq, sync_stream
+from constants import *
+from tubetestutil import *
from twisted.words.xish import domish, xpath
from ns import DISCO_INFO, TUBES, SI, FEATURE_NEG, IBB, MUC_BYTESTREAM, X_DATA
@@ -150,14 +152,9 @@ def test(q, bus, conn, stream):
tubes = tubes_iface.ListTubes(byte_arrays=True)
assert len(tubes) == 1
- sorted(tubes) == sorted([(
- dbus_tube_id,
- self_handle,
- 0, # DBUS
- 'com.example.TestCase',
- sample_parameters,
- 1, # Remote Pending
- )])
+ expected_tube = (dbus_tube_id, self_handle, TUBE_TYPE_DBUS,
+ 'com.example.TestCase', sample_parameters, TUBE_STATE_REMOTE_PENDING)
+ check_tube_in_tubes(expected_tube, tubes)
# Alice accepts the tube and wants to use IBB
result = make_result_iq(stream, iq)
diff --git a/tests/twisted/tubes/tubetestutil.py b/tests/twisted/tubes/tubetestutil.py
new file mode 100644
index 0000000..fff393c
--- /dev/null
+++ b/tests/twisted/tubes/tubetestutil.py
@@ -0,0 +1,29 @@
+"""
+Helper functions for writing tubes tests
+"""
+
+from servicetest import unwrap
+
+def check_tube_in_tubes(tube, tubes):
+ """
+ Check that 'tube' is in 'tubes', which should be the return value of
+ ListTubes()
+ """
+
+ utube = unwrap(tube)
+ for t in tubes:
+ if tube[0] != t[0]:
+ continue
+
+ pair = "\n %s,\n %s" % (utube, unwrap(t))
+
+ assert tube[1] == t[1], "self handles don't match: %s" % pair
+ assert tube[2] == t[2], "tube types don't match: %s" % pair
+ assert tube[3] == t[3], "services don't match: %s " % pair
+ assert tube[4] == t[4], "parameters don't match: %s" % pair
+ assert tube[5] == t[5], "states don't match: %s" % pair
+
+ return
+
+ assert False, "tube %s not in %s" % (unwrap (tube), unwrap (tubes))
+
--
1.5.6.5
More information about the telepathy-commits
mailing list