[Telepathy-commits] [telepathy-gabble/master] test-si-accept-tubes.py: Remove dead code, clean up
Alban Crequy
alban.crequy at collabora.co.uk
Tue Nov 11 10:35:57 PST 2008
---
tests/twisted/tubes/test-si-accept-tubes.py | 169 +--------------------------
1 files changed, 3 insertions(+), 166 deletions(-)
diff --git a/tests/twisted/tubes/test-si-accept-tubes.py b/tests/twisted/tubes/test-si-accept-tubes.py
index 1594563..62b4c0b 100644
--- a/tests/twisted/tubes/test-si-accept-tubes.py
+++ b/tests/twisted/tubes/test-si-accept-tubes.py
@@ -1,160 +1,19 @@
-"""Test 1-1 tubes support."""
-
-import base64
-import errno
-import os
+"""Test to accept a 1-1 stream tube."""
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, EventProtocolClientFactory
-from gabbletest import exec_test, acknowledge_iq, sync_stream
+ EventProtocolClientFactory
+from gabbletest import exec_test, acknowledge_iq
from twisted.words.xish import domish, xpath
-from twisted.internet.protocol import Factory, Protocol
from twisted.internet import reactor
-from twisted.words.protocols.jabber.client import IQ
-
-from gabbleconfig import HAVE_DBUS_TUBES
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'),
- 'u': dbus.UInt32(123),
- 'i': dbus.Int32(-123),
- }, signature='sv')
-
-new_sample_parameters = dbus.Dictionary({
- 's': 'newhello',
- 'ay': dbus.ByteArray('newhello'),
- 'u': dbus.UInt32(123),
- 'i': dbus.Int32(-123),
- }, signature='sv')
-
-
-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')
-
- if channel_list == None:
- assert properties.get('Channels') == [], properties['Channels']
- else:
- for i in channel_list:
- 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,
- },
- ['org.freedesktop.Telepathy.Channel.TargetHandle',
- 'org.freedesktop.Telepathy.Channel.TargetID',
- ]
- ) 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,
- },
- ['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',
- ]
- ) in properties.get('RequestableChannelClasses'),\
- properties['RequestableChannelClasses']
-
-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')
- 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='org.freedesktop.DBus.Properties')
- 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'), \
- channel_props['Interfaces']
-
- stream_tube_props = channel.GetAll(
- 'org.freedesktop.Telepathy.Channel.Type.StreamTube.DRAFT',
- dbus_interface='org.freedesktop.DBus.Properties')
- 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_NewChannel_signal(old_sig, channel_type, chan_path, contact_handle):
- 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[3] == contact_handle
- assert old_sig[4] == True # suppress handler
-
-def check_NewChannels_signal(new_sig, channel_type, chan_path, contact_handle,
- contact_id, initiator_handle):
- assert len(new_sig) == 1
- assert len(new_sig[0]) == 1 # one channel
- assert len(new_sig[0][0]) == 2 # two struct members
- 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'
-
def test(q, bus, conn, stream):
- check_conn_properties(q, bus, conn, stream)
-
conn.Connect()
_, vcard_event, roster_event = q.expect_many(
@@ -170,30 +29,8 @@ def test(q, bus, conn, stream):
item = roster_event.query.addElement('item')
item['jid'] = 'bob at localhost' # Bob can do tubes
item['subscription'] = 'both'
- item = roster_event.query.addElement('item')
- item['jid'] = 'joe at localhost' # Joe cannot do tubes
- item['subscription'] = 'both'
stream.send(roster)
- # Send Joe presence is without caps
- presence = domish.Element(('jabber:client', 'presence'))
- presence['from'] = 'joe at localhost/Joe'
- presence['to'] = 'test at localhost/Resource'
- c = presence.addElement('c')
- c['xmlns'] = 'http://jabber.org/protocol/caps'
- c['node'] = 'http://example.com/IDontSupportTubes'
- c['ver'] = '1.0'
- stream.send(presence)
-
- event = q.expect('stream-iq', iq_type='get',
- query_ns='http://jabber.org/protocol/disco#info',
- to='joe at localhost/Joe')
- result = event.stanza
- result['type'] = 'result'
- assert event.query['node'] == \
- 'http://example.com/IDontSupportTubes#1.0'
- stream.send(result)
-
# Send Bob presence and his caps
presence = domish.Element(('jabber:client', 'presence'))
presence['from'] = 'bob at localhost/Bob'
--
1.5.6.5
More information about the Telepathy-commits
mailing list