[Telepathy-commits] [telepathy-gabble/master] offer-accept-private-dbus-stream-tube-ibb.py: use constants.py
Guillaume Desmottes
guillaume.desmottes at collabora.co.uk
Tue Jan 13 07:47:13 PST 2009
---
.../offer-accept-private-dbus-stream-tube-ibb.py | 206 ++++++++------------
1 files changed, 78 insertions(+), 128 deletions(-)
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 ffb7456..00bb073 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
@@ -8,9 +8,11 @@ import dbus
from dbus.connection import Connection
from dbus.lowlevel import SignalMessage
-from servicetest import call_async, EventPattern, tp_name_prefix, \
- watch_tube_signals, sync_dbus
+from servicetest import call_async, EventPattern, watch_tube_signals
from gabbletest import exec_test, acknowledge_iq, sync_stream
+from constants import *
+
+from dbus import PROPERTIES_IFACE
from twisted.words.xish import domish, xpath
from twisted.internet.protocol import Factory, Protocol
@@ -38,11 +40,6 @@ new_sample_parameters = dbus.Dictionary({
'i': dbus.Int32(-123),
}, signature='sv')
-TUBE_CHANNEL_STATE_LOCAL_PENDING = 0
-TUBE_CHANNEL_STATE_REMOTE_PENDING = 1
-TUBE_CHANNEL_STATE_OPEN = 2
-TUBE_CHANNEL_STATE_NOT_OFFERED = 3
-
class Echo(Protocol):
def dataReceived(self, data):
self.transport.write(data)
@@ -59,8 +56,8 @@ def set_up_echo(name):
def check_conn_properties(q, bus, conn, stream, channel_list=None):
properties = conn.GetAll(
- 'org.freedesktop.Telepathy.Connection.Interface.Requests',
- dbus_interface='org.freedesktop.DBus.Properties')
+ CONN_IFACE_REQUESTS,
+ dbus_interface=PROPERTIES_IFACE)
if channel_list == None:
assert properties.get('Channels') == [], properties['Channels']
@@ -69,24 +66,17 @@ def check_conn_properties(q, bus, conn, stream, channel_list=None):
assert i in properties['Channels'], \
(i, properties['Channels'])
- assert ({'org.freedesktop.Telepathy.Channel.ChannelType':
- 'org.freedesktop.Telepathy.Channel.Type.Tubes',
- 'org.freedesktop.Telepathy.Channel.TargetHandleType': 1,
+ assert ({CHANNEL_TYPE: CHANNEL_TYPE_TUBES,
+ TARGET_HANDLE_TYPE: HT_CONTACT,
},
- ['org.freedesktop.Telepathy.Channel.TargetHandle',
- 'org.freedesktop.Telepathy.Channel.TargetID',
+ [TARGET_HANDLE, TARGET_ID
]
) in properties.get('RequestableChannelClasses'),\
properties['RequestableChannelClasses']
- assert ({'org.freedesktop.Telepathy.Channel.ChannelType':
- 'org.freedesktop.Telepathy.Channel.Type.StreamTube.DRAFT',
- 'org.freedesktop.Telepathy.Channel.TargetHandleType': 1,
+ assert ({CHANNEL_TYPE: CHANNEL_TYPE_STREAM_TUBE,
+ TARGET_HANDLE_TYPE: HT_CONTACT
},
- ['org.freedesktop.Telepathy.Channel.TargetHandle',
- 'org.freedesktop.Telepathy.Channel.TargetID',
- 'org.freedesktop.Telepathy.Channel.Interface.Tube.DRAFT.Parameters',
- 'org.freedesktop.Telepathy.Channel.Type.StreamTube.DRAFT.Service',
- ]
+ [TARGET_HANDLE, TARGET_ID, TUBE_PARAMETERS, STREAM_TUBE_SERVICE]
) in properties.get('RequestableChannelClasses'),\
properties['RequestableChannelClasses']
@@ -94,18 +84,14 @@ 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='org.freedesktop.DBus.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') == 1,\
channel_props.get('TargetHandleType')
- assert channel_props.get('ChannelType') == \
- 'org.freedesktop.Telepathy.Channel.Type.' + channel_type,\
- channel_props.get('ChannelType')
+ assert channel_props.get('ChannelType') == channel_type , channel_props.get('ChannelType')
assert 'Interfaces' in channel_props, channel_props
- assert 'org.freedesktop.Telepathy.Channel.Interface.Group' not in \
+ assert CHANNEL_IFACE_GROUP not in \
channel_props['Interfaces'], \
channel_props['Interfaces']
assert channel_props['TargetID'] == contact_id
@@ -113,27 +99,23 @@ def check_channel_properties(q, bus, conn, stream, channel, channel_type,
assert channel_props['InitiatorID'] == 'test at localhost'
assert channel_props['InitiatorHandle'] == conn.GetSelfHandle()
-
- if channel_type == "Tubes":
+ 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(
- 'org.freedesktop.Telepathy.Channel.Interface.Tube.DRAFT',
- dbus_interface='org.freedesktop.DBus.Properties')
+ tube_props = channel.GetAll(CHANNEL_IFACE_TUBE,
+ dbus_interface=PROPERTIES_IFACE)
assert tube_props['Status'] == 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'), \
+ dbus.Array([CHANNEL_IFACE_TUBE], signature='s'), \
channel_props['Interfaces']
- stream_tube_props = channel.GetAll(
- 'org.freedesktop.Telepathy.Channel.Type.StreamTube.DRAFT',
- dbus_interface='org.freedesktop.DBus.Properties')
+ 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
@@ -143,8 +125,8 @@ def check_channel_properties(q, bus, conn, stream, channel, channel_type,
def check_NewChannel_signal(old_sig, channel_type, chan_path, contact_handle, suppress_handler):
if chan_path is not None:
assert old_sig[0] == chan_path
- assert old_sig[1] == tp_name_prefix + '.Channel.Type.' + channel_type
- assert old_sig[2] == 1 # contact handle
+ assert old_sig[1] == channel_type
+ assert old_sig[2] == HT_CONTACT
assert old_sig[3] == contact_handle
assert old_sig[4] == suppress_handler # suppress handler
@@ -156,18 +138,13 @@ def check_NewChannels_signal(new_sig, channel_type, chan_path, contact_handle,
assert new_sig[0][0][0] == chan_path
emitted_props = new_sig[0][0][1]
- assert emitted_props[tp_name_prefix + '.Channel.ChannelType'] ==\
- tp_name_prefix + '.Channel.Type.' + channel_type
- assert emitted_props[tp_name_prefix + '.Channel.TargetHandleType'] == 1
- assert emitted_props[tp_name_prefix + '.Channel.TargetHandle'] ==\
- contact_handle
- assert emitted_props[tp_name_prefix + '.Channel.TargetID'] == \
- contact_id
- assert emitted_props[tp_name_prefix + '.Channel.Requested'] == True
- assert emitted_props[tp_name_prefix + '.Channel.InitiatorHandle'] \
- == initiator_handle
- assert emitted_props[tp_name_prefix + '.Channel.InitiatorID'] == \
- 'test at localhost'
+ assert emitted_props[CHANNEL_TYPE] == channel_type
+ assert emitted_props[TARGET_HANDLE_TYPE] == 1
+ assert emitted_props[TARGET_HANDLE] == contact_handle
+ assert emitted_props[TARGET_ID] == contact_id
+ assert emitted_props[REQUESTED] == True
+ assert emitted_props[INITIATOR_HANDLE] == initiator_handle
+ assert emitted_props[INITIATOR_ID] == 'test at localhost'
def test(q, bus, conn, stream):
@@ -242,14 +219,12 @@ def test(q, bus, conn, stream):
sync_stream(q, stream)
# new requestotron
- requestotron = dbus.Interface(conn,
- 'org.freedesktop.Telepathy.Connection.Interface.Requests')
+ requestotron = dbus.Interface(conn, CONN_IFACE_REQUESTS)
# Test tubes with Joe. Joe does not have tube capabilities.
# Gabble does not allow to offer a tube to him.
joe_handle = conn.RequestHandles(1, ['joe at localhost'])[0]
- call_async(q, conn, 'RequestChannel',
- tp_name_prefix + '.Channel.Type.Tubes', 1, joe_handle, True);
+ call_async(q, conn, 'RequestChannel', CHANNEL_TYPE_TUBES, HT_CONTACT, joe_handle, True);
ret, old_sig, new_sig = q.expect_many(
EventPattern('dbus-return', method='RequestChannel'),
@@ -259,8 +234,7 @@ def test(q, bus, conn, stream):
joe_chan_path = ret.value[0]
joe_tubes_chan = bus.get_object(conn.bus_name, joe_chan_path)
- joe_tubes_iface = dbus.Interface(joe_tubes_chan,
- tp_name_prefix + '.Channel.Type.Tubes')
+ joe_tubes_iface = dbus.Interface(joe_tubes_chan, CHANNEL_TYPE_TUBES)
path = os.getcwd() + '/stream'
call_async(q, joe_tubes_iface, 'OfferStreamTube',
'echo', sample_parameters, 0, dbus.ByteArray(path), 0, "")
@@ -272,8 +246,7 @@ def test(q, bus, conn, stream):
bob_handle = conn.RequestHandles(1, ['bob at localhost'])[0]
# old requestotron
- call_async(q, conn, 'RequestChannel',
- tp_name_prefix + '.Channel.Type.Tubes', 1, bob_handle, True);
+ call_async(q, conn, 'RequestChannel', CHANNEL_TYPE_TUBES, 1, bob_handle, True)
ret, old_sig, new_sig = q.expect_many(
EventPattern('dbus-return', method='RequestChannel'),
@@ -284,8 +257,8 @@ def test(q, bus, conn, stream):
assert len(ret.value) == 1
chan_path = ret.value[0]
- check_NewChannel_signal(old_sig.args, "Tubes", chan_path, bob_handle, True)
- check_NewChannels_signal(new_sig.args, "Tubes", chan_path,
+ check_NewChannel_signal(old_sig.args, CHANNEL_TYPE_TUBES, chan_path, bob_handle, True)
+ check_NewChannels_signal(new_sig.args, CHANNEL_TYPE_TUBES, chan_path,
bob_handle, 'bob at localhost', conn.GetSelfHandle())
old_tubes_channel_properties = new_sig.args[0][0]
@@ -294,12 +267,9 @@ def test(q, bus, conn, stream):
# Try to CreateChannel with unknown properties
# Gabble must return an error
call_async(q, requestotron, 'CreateChannel',
- {'org.freedesktop.Telepathy.Channel.ChannelType':
- 'org.freedesktop.Telepathy.Channel.Type.StreamTube.DRAFT',
- 'org.freedesktop.Telepathy.Channel.TargetHandleType':
- 1,
- 'org.freedesktop.Telepathy.Channel.TargetHandle':
- bob_handle,
+ {CHANNEL_TYPE: CHANNEL_TYPE_STREAM_TUBE,
+ TARGET_HANDLE_TYPE: HT_CONTACT,
+ TARGET_HANDLE: bob_handle,
'this.property.does.not.exist':
'this.value.should.not.exist'
});
@@ -310,12 +280,9 @@ def test(q, bus, conn, stream):
# Try to CreateChannel with missing properties ("Service")
# Gabble must return an error
call_async(q, requestotron, 'CreateChannel',
- {'org.freedesktop.Telepathy.Channel.ChannelType':
- 'org.freedesktop.Telepathy.Channel.Type.StreamTube.DRAFT',
- 'org.freedesktop.Telepathy.Channel.TargetHandleType':
- 1,
- 'org.freedesktop.Telepathy.Channel.TargetHandle':
- bob_handle
+ {CHANNEL_TYPE: CHANNEL_TYPE_STREAM_TUBE,
+ TARGET_HANDLE_TYPE: HT_CONTACT,
+ TARGET_HANDLE: bob_handle
});
ret = q.expect_many(EventPattern('dbus-error', method='CreateChannel'))
# CreateChannel failed, we expect no new channel
@@ -324,16 +291,11 @@ def test(q, bus, conn, stream):
# Try to CreateChannel with correct properties
# Gabble must succeed
call_async(q, requestotron, 'CreateChannel',
- {'org.freedesktop.Telepathy.Channel.ChannelType':
- 'org.freedesktop.Telepathy.Channel.Type.StreamTube.DRAFT',
- 'org.freedesktop.Telepathy.Channel.TargetHandleType':
- 1,
- 'org.freedesktop.Telepathy.Channel.TargetHandle':
- bob_handle,
- 'org.freedesktop.Telepathy.Channel.Type.StreamTube.DRAFT.Service':
- "newecho",
- 'org.freedesktop.Telepathy.Channel.Interface.Tube.DRAFT.Parameters':
- dbus.Dictionary({'foo': 'bar'}, signature='sv'),
+ {CHANNEL_TYPE: CHANNEL_TYPE_STREAM_TUBE,
+ TARGET_HANDLE_TYPE: HT_CONTACT,
+ TARGET_HANDLE: bob_handle,
+ STREAM_TUBE_SERVICE: 'newecho',
+ TUBE_PARAMETERS: dbus.Dictionary({'foo': 'bar'}, signature='sv'),
});
# the NewTube signal (old API) is not fired now as the tube wasn't offered
@@ -348,8 +310,8 @@ def test(q, bus, conn, stream):
new_chan_path = ret.value[0]
new_chan_prop_asv = ret.value[1]
# Status and Parameters are mutables so not announced
- assert (tp_name_prefix + '.Channel.Interface.Tube.DRAFT.Status') not in new_chan_prop_asv
- assert (tp_name_prefix + '.Channel.Interface.Tube.DRAFT.Parameters') not in new_chan_prop_asv
+ assert (TUBE_STATUS) not in new_chan_prop_asv
+ assert (TUBE_PARAMETERS) not in new_chan_prop_asv
assert new_chan_path.find("StreamTube") != -1, new_chan_path
assert new_chan_path.find("SITubesChannel") == -1, new_chan_path
# The path of the Channel.Type.Tubes object MUST be different to the path
@@ -357,21 +319,19 @@ def test(q, bus, conn, stream):
assert chan_path != new_chan_path
new_tube_chan = bus.get_object(conn.bus_name, new_chan_path)
- new_tube_iface = dbus.Interface(new_tube_chan,
- tp_name_prefix + '.Channel.Type.StreamTube.DRAFT')
+ new_tube_iface = dbus.Interface(new_tube_chan, CHANNEL_TYPE_STREAM_TUBE)
# check Status and Parameters
- new_tube_props = new_tube_chan.GetAll(
- 'org.freedesktop.Telepathy.Channel.Interface.Tube.DRAFT',
- dbus_interface='org.freedesktop.DBus.Properties')
+ new_tube_props = new_tube_chan.GetAll(CHANNEL_IFACE_TUBE,
+ dbus_interface=PROPERTIES_IFACE)
# the tube created using the old API is in the "not offered" state
assert new_tube_props['Status'] == TUBE_CHANNEL_STATE_NOT_OFFERED
assert new_tube_props['Parameters'] == {'foo': 'bar'}
- check_NewChannel_signal(old_sig.args, "StreamTube.DRAFT", \
+ check_NewChannel_signal(old_sig.args, CHANNEL_TYPE_STREAM_TUBE,
new_chan_path, bob_handle, True)
- check_NewChannels_signal(new_sig.args, "StreamTube.DRAFT", new_chan_path, \
+ check_NewChannels_signal(new_sig.args, CHANNEL_TYPE_STREAM_TUBE, new_chan_path, \
bob_handle, 'bob at localhost', conn.GetSelfHandle())
stream_tube_channel_properties = new_sig.args[0][0]
@@ -380,10 +340,9 @@ def test(q, bus, conn, stream):
tubes_chan = bus.get_object(conn.bus_name, chan_path)
- tubes_iface = dbus.Interface(tubes_chan,
- tp_name_prefix + '.Channel.Type.Tubes')
+ tubes_iface = dbus.Interface(tubes_chan, CHANNEL_TYPE_TUBES)
- check_channel_properties(q, bus, conn, stream, tubes_chan, "Tubes",
+ check_channel_properties(q, bus, conn, stream, tubes_chan, CHANNEL_TYPE_TUBES,
bob_handle, "bob at localhost")
# Create another tube using old API
@@ -421,9 +380,9 @@ def test(q, bus, conn, stream):
EventPattern('dbus-signal', signal='NewChannels'))
# the tube channel (new API) is announced
- check_NewChannel_signal(new_chan.args, "StreamTube.DRAFT", \
+ check_NewChannel_signal(new_chan.args, CHANNEL_TYPE_STREAM_TUBE,
None, bob_handle, False)
- check_NewChannels_signal(new_chans.args, "StreamTube.DRAFT", new_chan.args[0],
+ check_NewChannels_signal(new_chans.args, CHANNEL_TYPE_STREAM_TUBE, new_chan.args[0],
bob_handle, "bob at localhost", self_handle)
props = new_chans.args[0][0][1]
@@ -431,10 +390,10 @@ def test(q, bus, conn, stream):
# We offered a tube using the old tube API and created one with the new
# API, so there are 2 tubes. Check the new tube API works
assert len(filter(lambda x:
- x[1] == "org.freedesktop.Telepathy.Channel.Type.Tubes",
+ x[1] == CHANNEL_TYPE_TUBES,
conn.ListChannels())) == 1
channels = filter(lambda x:
- x[1] == "org.freedesktop.Telepathy.Channel.Type.StreamTube.DRAFT" and
+ x[1] == CHANNEL_TYPE_STREAM_TUBE and
x[0] == new_chan_path,
conn.ListChannels())
assert len(channels) == 1
@@ -442,38 +401,33 @@ def test(q, bus, conn, stream):
old_tube_chan = bus.get_object(conn.bus_name, new_chan.args[0])
- tube_basic_props = old_tube_chan.GetAll(
- 'org.freedesktop.Telepathy.Channel',
- dbus_interface='org.freedesktop.DBus.Properties')
+ tube_basic_props = old_tube_chan.GetAll(CHANNEL,
+ dbus_interface=PROPERTIES_IFACE)
assert tube_basic_props.get("InitiatorHandle") == self_handle
- stream_tube_props = old_tube_chan.GetAll(
- 'org.freedesktop.Telepathy.Channel.Type.StreamTube.DRAFT',
- dbus_interface='org.freedesktop.DBus.Properties')
+ stream_tube_props = old_tube_chan.GetAll(CHANNEL_TYPE_STREAM_TUBE,
+ dbus_interface=PROPERTIES_IFACE)
assert stream_tube_props.get("Service") == "echo", stream_tube_props
- old_tube_props = old_tube_chan.GetAll(
- 'org.freedesktop.Telepathy.Channel.Interface.Tube.DRAFT',
- dbus_interface='org.freedesktop.DBus.Properties', byte_arrays=True)
+ old_tube_props = old_tube_chan.GetAll(CHANNEL_IFACE_TUBE,
+ dbus_interface=PROPERTIES_IFACE, byte_arrays=True)
assert old_tube_props.get("Parameters") == dbus.Dictionary(sample_parameters)
# Tube have been created using the old API and so is already offered
assert old_tube_props['Status'] == TUBE_CHANNEL_STATE_REMOTE_PENDING
- check_channel_properties(q, bus, conn, stream, tubes_chan, "Tubes",
+ check_channel_properties(q, bus, conn, stream, tubes_chan, CHANNEL_TYPE_TUBES,
bob_handle, "bob at localhost")
check_channel_properties(q, bus, conn, stream, old_tube_chan,
- "StreamTube.DRAFT", bob_handle, "bob at localhost", TUBE_CHANNEL_STATE_REMOTE_PENDING)
+ CHANNEL_TYPE_STREAM_TUBE, bob_handle, "bob at localhost", TUBE_CHANNEL_STATE_REMOTE_PENDING)
# change the parameters of the not offered tube (the one created using the
# new API)
- new_tube_chan.Set('org.freedesktop.Telepathy.Channel.Interface.Tube.DRAFT',
- 'Parameters', new_sample_parameters,
- dbus_interface='org.freedesktop.DBus.Properties')
+ new_tube_chan.Set(CHANNEL_IFACE_TUBE, 'Parameters', new_sample_parameters,
+ dbus_interface=PROPERTIES_IFACE)
# check it is correctly changed
- new_tube_props = new_tube_chan.GetAll(
- 'org.freedesktop.Telepathy.Channel.Interface.Tube.DRAFT',
- dbus_interface='org.freedesktop.DBus.Properties', byte_arrays=True)
+ new_tube_props = new_tube_chan.GetAll(CHANNEL_IFACE_TUBE,
+ dbus_interface=PROPERTIES_IFACE, byte_arrays=True)
assert new_tube_props.get("Parameters") == new_sample_parameters, \
new_tube_props.get("Parameters")
@@ -520,19 +474,16 @@ def test(q, bus, conn, stream):
# The new tube has been offered, the parameters cannot be changed anymore
# We need to use call_async to check the error
- tube_prop_iface = dbus.Interface(old_tube_chan,
- 'org.freedesktop.DBus.Properties')
- call_async(q, tube_prop_iface, 'Set',
- 'org.freedesktop.Telepathy.Channel.Interface.Tube.DRAFT',
+ tube_prop_iface = dbus.Interface(old_tube_chan, PROPERTIES_IFACE)
+ call_async(q, tube_prop_iface, 'Set', CHANNEL_IFACE_TUBE,
'Parameters', dbus.Dictionary(
{dbus.String(u'foo2'): dbus.String(u'bar2')},
signature=dbus.Signature('sv')),
- dbus_interface='org.freedesktop.DBus.Properties')
+ dbus_interface=PROPERTIES_IFACE)
set_error = q.expect('dbus-error')
# check it is *not* correctly changed
- new_tube_props = new_tube_chan.GetAll(
- 'org.freedesktop.Telepathy.Channel.Interface.Tube.DRAFT',
- dbus_interface='org.freedesktop.DBus.Properties', byte_arrays=True)
+ new_tube_props = new_tube_chan.GetAll(CHANNEL_IFACE_TUBE,
+ dbus_interface=PROPERTIES_IFACE, byte_arrays=True)
assert new_tube_props.get("Parameters") == new_sample_parameters, \
new_tube_props.get("Parameters")
@@ -701,8 +652,7 @@ def test(q, bus, conn, stream):
assert binary == 'hello, new world'
# OK, how about D-Bus?
- call_async(q, tubes_iface, 'OfferDBusTube',
- 'com.example.TestCase', sample_parameters)
+ call_async(q, tubes_iface, 'OfferDBusTube', 'com.example.TestCase', sample_parameters)
event = q.expect('stream-iq', iq_type='set', to='bob at localhost/Bob')
iq = event.stanza
--
1.5.6.5
More information about the Telepathy-commits
mailing list