[Telepathy-commits] [telepathy-gabble/master] Move duplicated tube(s) property checking to tubetestutil
Will Thompson
will.thompson at collabora.co.uk
Wed Feb 18 08:31:05 PST 2009
---
tests/twisted/constants.py | 1 +
.../offer-accept-private-dbus-stream-tube-ibb.py | 51 ++---------------
...offer-accept-private-dbus-stream-tube-socks5.py | 59 ++------------------
tests/twisted/tubes/tubetestutil.py | 50 +++++++++++++++++
4 files changed, 61 insertions(+), 100 deletions(-)
diff --git a/tests/twisted/constants.py b/tests/twisted/constants.py
index 4f07a73..f69c91f 100644
--- a/tests/twisted/constants.py
+++ b/tests/twisted/constants.py
@@ -66,6 +66,7 @@ SOCKET_ACCESS_CONTROL_CREDENTIALS = 3
TUBE_STATE_LOCAL_PENDING = 0
TUBE_STATE_REMOTE_PENDING = 1
TUBE_STATE_OPEN = 2
+TUBE_STATE_NOT_OFFERED = 3
TUBE_TYPE_DBUS = 0
TUBE_TYPE_STREAM = 1
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 58d8dbd..db3107b 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
@@ -33,48 +33,6 @@ new_sample_parameters = dbus.Dictionary({
'i': dbus.Int32(-123),
}, signature='sv')
-def check_channel_properties(q, bus, conn, stream, channel, channel_type,
- contact_handle, contact_id, state=None):
- # Exercise basic Channel Properties from spec 0.17.7
- # on the channel of type channel_type
- channel_props = channel.GetAll(CHANNEL, dbus_interface=PROPERTIES_IFACE)
- assert channel_props.get('TargetHandle') == contact_handle,\
- (channel_props.get('TargetHandle'), contact_handle)
- assert channel_props.get('TargetHandleType') == 1,\
- channel_props.get('TargetHandleType')
- assert channel_props.get('ChannelType') == channel_type , channel_props.get('ChannelType')
- assert 'Interfaces' in channel_props, channel_props
- assert CHANNEL_IFACE_GROUP not in \
- channel_props['Interfaces'], \
- channel_props['Interfaces']
- assert channel_props['TargetID'] == contact_id
- assert channel_props['Requested'] == True
- assert channel_props['InitiatorID'] == 'test at localhost'
- assert channel_props['InitiatorHandle'] == conn.GetSelfHandle()
-
- if channel_type == CHANNEL_TYPE_TUBES:
- assert state is None
- assert len(channel_props['Interfaces']) == 0, channel_props['Interfaces']
- supported_socket_types = channel.GetAvailableStreamTubeTypes()
- else:
- assert state is not None
- tube_props = channel.GetAll(CHANNEL_IFACE_TUBE,
- dbus_interface=PROPERTIES_IFACE)
- assert tube_props['State'] == state
- # no strict check but at least check the properties exist
- assert tube_props['Parameters'] is not None
- assert channel_props['Interfaces'] == \
- dbus.Array([CHANNEL_IFACE_TUBE], signature='s'), \
- channel_props['Interfaces']
-
- stream_tube_props = channel.GetAll(CHANNEL_TYPE_STREAM_TUBE,
- dbus_interface=PROPERTIES_IFACE)
- supported_socket_types = stream_tube_props['SupportedSocketTypes']
-
- # Support for different socket types. no strict check but at least check
- # there is some support.
- assert len(supported_socket_types) == 3
-
def check_NewChannels_signal(new_sig, channel_type, chan_path, contact_handle,
contact_id, initiator_handle):
assert len(new_sig) == 1
@@ -291,7 +249,7 @@ def test(q, bus, conn, stream):
tubes_iface = dbus.Interface(tubes_chan, CHANNEL_TYPE_TUBES)
- check_channel_properties(q, bus, conn, stream, tubes_chan, CHANNEL_TYPE_TUBES,
+ t.check_channel_properties(q, bus, conn, tubes_chan, CHANNEL_TYPE_TUBES,
bob_handle, "bob at localhost")
# Create another tube using old API
@@ -368,10 +326,11 @@ def test(q, bus, conn, stream):
# Tube have been created using the old API and so is already offered
assert old_tube_props['State'] == TUBE_CHANNEL_STATE_REMOTE_PENDING
- check_channel_properties(q, bus, conn, stream, tubes_chan, CHANNEL_TYPE_TUBES,
+ t.check_channel_properties(q, bus, conn, tubes_chan, CHANNEL_TYPE_TUBES,
bob_handle, "bob at localhost")
- check_channel_properties(q, bus, conn, stream, old_tube_chan,
- CHANNEL_TYPE_STREAM_TUBE, bob_handle, "bob at localhost", TUBE_CHANNEL_STATE_REMOTE_PENDING)
+ t.check_channel_properties(q, bus, conn, old_tube_chan,
+ CHANNEL_TYPE_STREAM_TUBE, bob_handle, "bob at localhost",
+ TUBE_CHANNEL_STATE_REMOTE_PENDING)
# Offer the first tube created (new API)
path2 = os.getcwd() + '/stream2'
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 cdd85cc..b517a03 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
@@ -60,56 +60,6 @@ class S5BFactory(Factory):
def clientConnectionLost(self, connector, reason):
pass
-def check_channel_properties(q, bus, conn, stream, channel, channel_type,
- contact_handle, contact_id, state=None):
- # Exercise basic Channel Properties from spec 0.17.7
- # on the channel of type channel_type
- channel_props = channel.GetAll(
- 'org.freedesktop.Telepathy.Channel',
- dbus_interface=dbus.PROPERTIES_IFACE)
- assert channel_props.get('TargetHandle') == contact_handle,\
- (channel_props.get('TargetHandle'), contact_handle)
- assert channel_props.get('TargetHandleType') == 1,\
- channel_props.get('TargetHandleType')
- assert channel_props.get('ChannelType') == \
- 'org.freedesktop.Telepathy.Channel.Type.' + channel_type,\
- channel_props.get('ChannelType')
- assert 'Interfaces' in channel_props, channel_props
- assert 'org.freedesktop.Telepathy.Channel.Interface.Group' not in \
- channel_props['Interfaces'], \
- channel_props['Interfaces']
- assert channel_props['TargetID'] == contact_id
- assert channel_props['Requested'] == True
- assert channel_props['InitiatorID'] == 'test at localhost'
- assert channel_props['InitiatorHandle'] == conn.GetSelfHandle()
-
-
- if channel_type == "Tubes":
- assert state is None
- assert len(channel_props['Interfaces']) == 0, channel_props['Interfaces']
- supported_socket_types = channel.GetAvailableStreamTubeTypes()
- else:
- assert state is not None
- tube_props = channel.GetAll(
- 'org.freedesktop.Telepathy.Channel.Interface.Tube.DRAFT',
- dbus_interface=dbus.PROPERTIES_IFACE)
- assert tube_props['State'] == state
- # no strict check but at least check the properties exist
- assert tube_props['Parameters'] is not None
- assert channel_props['Interfaces'] == \
- dbus.Array(['org.freedesktop.Telepathy.Channel.Interface.Tube.DRAFT'],
- signature='s'), \
- channel_props['Interfaces']
-
- stream_tube_props = channel.GetAll(
- 'org.freedesktop.Telepathy.Channel.Type.StreamTube.DRAFT',
- dbus_interface=dbus.PROPERTIES_IFACE)
- supported_socket_types = stream_tube_props['SupportedSocketTypes']
-
- # Support for different socket types. no strict check but at least check
- # there is some support.
- assert len(supported_socket_types) == 3
-
def check_NewChannels_signal(new_sig, channel_type, chan_path, contact_handle,
contact_id, initiator_handle):
assert len(new_sig) == 1
@@ -292,7 +242,7 @@ def test(q, bus, conn, stream):
tubes_iface = dbus.Interface(tubes_chan,
tp_name_prefix + '.Channel.Type.Tubes')
- check_channel_properties(q, bus, conn, stream, tubes_chan, "Tubes",
+ t.check_channel_properties(q, bus, conn, tubes_chan, CHANNEL_TYPE_TUBES,
bob_handle, "bob at localhost")
# Offer the tube, old API
@@ -357,10 +307,11 @@ def test(q, bus, conn, stream):
# 3 == Tube_Channel_State_Not_Offered
assert tube_props.get("State") == 3, tube_props
- check_channel_properties(q, bus, conn, stream, tubes_chan, "Tubes",
+ t.check_channel_properties(q, bus, conn, tubes_chan, CHANNEL_TYPE_TUBES,
bob_handle, "bob at localhost")
- check_channel_properties(q, bus, conn, stream, tube_chan,
- "StreamTube.DRAFT", bob_handle, "bob at localhost", 3)
+ t.check_channel_properties(q, bus, conn, tube_chan,
+ CHANNEL_TYPE_STREAM_TUBE, bob_handle, "bob at localhost",
+ TUBE_STATE_NOT_OFFERED)
# Offer the tube, new API
path2 = os.getcwd() + '/stream2'
diff --git a/tests/twisted/tubes/tubetestutil.py b/tests/twisted/tubes/tubetestutil.py
index 93f6887..aa65e67 100644
--- a/tests/twisted/tubes/tubetestutil.py
+++ b/tests/twisted/tubes/tubetestutil.py
@@ -5,6 +5,7 @@ Helper functions for writing tubes tests
import errno
import os
+import dbus
from dbus import PROPERTIES_IFACE
from servicetest import unwrap
@@ -124,6 +125,55 @@ def check_NewChannel_signal(args, channel_type, chan_path, contact_handle,
assert args[3] == contact_handle, (args, contact_handle)
assert args[4] == suppress_handler, (args, suppress_handler)
+def check_channel_properties(q, bus, conn, channel, channel_type,
+ contact_handle, contact_id, state=None):
+ """
+ Checks the D-Bus properties of a 1-1 Tubes, StreamTube or DBusTube channel,
+ initiated by the test user
+ """
+
+ # Check o.fd.T.Channel properties.
+ channel_props = channel.GetAll(CHANNEL, dbus_interface=PROPERTIES_IFACE)
+ assert channel_props.get('TargetHandle') == contact_handle, \
+ (channel_props.get('TargetHandle'), contact_handle)
+ assert channel_props.get('TargetHandleType') == HT_CONTACT, \
+ channel_props.get('TargetHandleType')
+ assert channel_props.get('ChannelType') == channel_type, \
+ channel_props.get('ChannelType')
+ assert 'Interfaces' in channel_props, channel_props
+ assert CHANNEL_IFACE_GROUP not in \
+ channel_props['Interfaces'], \
+ channel_props['Interfaces']
+ assert channel_props['TargetID'] == contact_id
+ assert channel_props['Requested'] == True
+ assert channel_props['InitiatorID'] == 'test at localhost'
+ assert channel_props['InitiatorHandle'] == conn.GetSelfHandle()
+
+ if channel_type == CHANNEL_TYPE_TUBES:
+ assert state is None
+ assert len(channel_props['Interfaces']) == 0, channel_props['Interfaces']
+ supported_socket_types = channel.GetAvailableStreamTubeTypes()
+ else:
+ assert state is not None
+ tube_props = channel.GetAll(CHANNEL_IFACE_TUBE,
+ dbus_interface=PROPERTIES_IFACE)
+ assert tube_props['State'] == state
+ # no strict check but at least check the properties exist
+ assert tube_props['Parameters'] is not None
+ assert channel_props['Interfaces'] == \
+ dbus.Array([CHANNEL_IFACE_TUBE], signature='s'), \
+ channel_props['Interfaces']
+
+ if channel_type == CHANNEL_TYPE_STREAM_TUBE:
+ supported_socket_types = channel.Get(CHANNEL_TYPE_STREAM_TUBE,
+ 'SupportedSocketTypes', dbus_interface=PROPERTIES_IFACE)
+ else:
+ supported_socket_types = None
+
+ if supported_socket_types is not None:
+ # FIXME: this should check for particular types, not just a magic length
+ assert len(supported_socket_types) == 3
+
class Echo(Protocol):
"""
A trivial protocol that just echoes back whatever you send it, in lowercase.
--
1.5.6.5
More information about the telepathy-commits
mailing list