[Telepathy-commits] [telepathy-gabble/master] Rename Ensuring-text-channels test, add to twisted Makefile
Will Thompson
will.thompson at collabora.co.uk
Wed Sep 24 08:22:43 PDT 2008
---
tests/twisted/Makefile.am | 1 +
tests/twisted/text/ensure-requestotron.py | 193 -----------------------------
tests/twisted/text/ensure.py | 193 +++++++++++++++++++++++++++++
3 files changed, 194 insertions(+), 193 deletions(-)
delete mode 100644 tests/twisted/text/ensure-requestotron.py
create mode 100644 tests/twisted/text/ensure.py
diff --git a/tests/twisted/Makefile.am b/tests/twisted/Makefile.am
index e3cd901..851a72c 100644
--- a/tests/twisted/Makefile.am
+++ b/tests/twisted/Makefile.am
@@ -20,6 +20,7 @@ TWISTED_TESTS = \
roster/test-save-alias-to-roster.py \
text/initiate.py \
text/initiate-requestotron.py \
+ text/ensure.py \
text/respawn.py \
text/test-text-delayed.py \
text/test-text-no-body.py \
diff --git a/tests/twisted/text/ensure-requestotron.py b/tests/twisted/text/ensure-requestotron.py
deleted file mode 100644
index cb68c16..0000000
--- a/tests/twisted/text/ensure-requestotron.py
+++ /dev/null
@@ -1,193 +0,0 @@
-"""
-Test text channel initiated by me, using Requests.EnsureChannel
-"""
-
-import dbus
-
-from twisted.words.xish import domish
-
-from gabbletest import exec_test
-from servicetest import call_async, EventPattern
-
-def test(q, bus, conn, stream):
- conn.Connect()
- q.expect('dbus-signal', signal='StatusChanged', args=[0, 1])
-
- self_handle = conn.GetSelfHandle()
-
- jids = ['foo at bar.com', 'truc at cafe.fr']
- call_async(q, conn, 'RequestHandles', 1, jids)
-
- event = q.expect('dbus-return', method='RequestHandles')
- handles = event.value[0]
-
- properties = conn.GetAll(
- 'org.freedesktop.Telepathy.Connection.Interface.Requests',
- dbus_interface='org.freedesktop.DBus.Properties')
- assert properties.get('Channels') == [], properties['Channels']
- assert ({'org.freedesktop.Telepathy.Channel.ChannelType':
- 'org.freedesktop.Telepathy.Channel.Type.Text',
- 'org.freedesktop.Telepathy.Channel.TargetHandleType': 1,
- },
- ['org.freedesktop.Telepathy.Channel.TargetHandle',
- 'org.freedesktop.Telepathy.Channel.TargetID'
- ],
- ) in properties.get('RequestableChannelClasses'),\
- properties['RequestableChannelClasses']
-
- requestotron = dbus.Interface(conn,
- 'org.freedesktop.Telepathy.Connection.Interface.Requests')
-
- test_ensure_ensure(q, requestotron, conn, self_handle, jids[0], handles[0])
- test_request_ensure(q, requestotron, conn, self_handle, jids[1], handles[1])
-
- conn.Disconnect()
- q.expect('dbus-signal', signal='StatusChanged', args=[2, 1])
-
-
-def test_ensure_ensure(q, requestotron, conn, self_handle, jid, handle):
- """
- Test ensuring a non-existant channel twice. The first call should succeed
- with Yours=True; the subsequent call should succeed with Yours=False
- """
-
- # Check that Ensuring a channel that doesn't exist succeeds
- call_async(q, requestotron, 'EnsureChannel', request_props (handle))
-
- ret, old_sig, new_sig = q.expect_many(
- EventPattern('dbus-return', method='EnsureChannel'),
- EventPattern('dbus-signal', signal='NewChannel'),
- EventPattern('dbus-signal', signal='NewChannels'),
- )
-
- assert len(ret.value) == 3
- yours, path, emitted_props = ret.value
-
- # The channel was created in response to the call, and we were the only
- # requestor, so we should get Yours=True
- assert yours, ret.value
-
- check_props(emitted_props, self_handle, handle, jid)
-
- assert len(old_sig.args) == 5
- old_path, old_ct, old_ht, old_h, old_sh = old_sig.args
-
- assert old_path == path
- assert old_ct == u'org.freedesktop.Telepathy.Channel.Type.Text'
- # check that handle type == contact handle
- assert old_ht == 1
- assert old_h == handle
- assert old_sh == True # suppress handler
-
- assert len(new_sig.args) == 1
- assert len(new_sig.args[0]) == 1 # one channel
- assert len(new_sig.args[0][0]) == 2 # two struct members
- assert new_sig.args[0][0][0] == path
- assert new_sig.args[0][0][1] == emitted_props
-
- properties = conn.GetAll(
- 'org.freedesktop.Telepathy.Connection.Interface.Requests',
- dbus_interface='org.freedesktop.DBus.Properties')
-
- assert new_sig.args[0][0] in properties['Channels'], \
- (new_sig.args[0][0], properties['Channels'])
-
-
- # Now try Ensuring a channel which already exists
- call_async(q, requestotron, 'EnsureChannel', request_props (handle))
- ret_ = q.expect('dbus-return', method='EnsureChannel')
-
- assert len(ret_.value) == 3
- yours_, path_, emitted_props_ = ret_.value
-
- # Someone's already responsible for this channel, so we should get
- # Yours=False
- assert not yours_, ret_.value
- assert path == path_, (path, path_)
- assert emitted_props == emitted_props_, (emitted_props, emitted_props_)
-
-
-def test_request_ensure(q, requestotron, conn, self_handle, jid, handle):
- """
- Test Creating a non-existant channel, then Ensuring the same channel.
- The call to Ensure should succeed with Yours=False.
- """
-
- call_async(q, requestotron, 'CreateChannel', request_props (handle))
-
- ret, old_sig, new_sig = q.expect_many(
- EventPattern('dbus-return', method='CreateChannel'),
- EventPattern('dbus-signal', signal='NewChannel'),
- EventPattern('dbus-signal', signal='NewChannels'),
- )
-
- assert len(ret.value) == 2
- path, emitted_props = ret.value
-
- check_props(emitted_props, self_handle, handle, jid)
-
- assert len(old_sig.args) == 5
- old_path, old_ct, old_ht, old_h, old_sh = old_sig.args
-
- assert old_path == path
- assert old_ct == u'org.freedesktop.Telepathy.Channel.Type.Text'
- # check that handle type == contact handle
- assert old_ht == 1
- assert old_h == handle
- assert old_sh == True # suppress handler
-
- assert len(new_sig.args) == 1
- assert len(new_sig.args[0]) == 1 # one channel
- assert len(new_sig.args[0][0]) == 2 # two struct members
- assert new_sig.args[0][0][0] == path
- assert new_sig.args[0][0][1] == emitted_props
-
- properties = conn.GetAll(
- 'org.freedesktop.Telepathy.Connection.Interface.Requests',
- dbus_interface='org.freedesktop.DBus.Properties')
-
- assert new_sig.args[0][0] in properties['Channels'], \
- (new_sig.args[0][0], properties['Channels'])
-
-
- # Now try Ensuring that same channel.
- call_async(q, requestotron, 'EnsureChannel', request_props (handle))
- ret_ = q.expect('dbus-return', method='EnsureChannel')
-
- assert len(ret_.value) == 3
- yours_, path_, emitted_props_ = ret_.value
-
- # Someone's already responsible for this channel, so we should get
- # Yours=False
- assert not yours_, ret_.value
- assert path == path_, (path, path_)
- assert emitted_props == emitted_props_, (emitted_props, emitted_props_)
-
-
-def check_props(props, self_handle, handle, jid):
- assert props['org.freedesktop.Telepathy.Channel.ChannelType'] ==\
- 'org.freedesktop.Telepathy.Channel.Type.Text'
- assert props['org.freedesktop.Telepathy.Channel.'
- 'TargetHandleType'] == 1
- assert props['org.freedesktop.Telepathy.Channel.TargetHandle'] ==\
- handle
- assert props['org.freedesktop.Telepathy.Channel.TargetID'] == jid
- assert props['org.freedesktop.Telepathy.Channel.FUTURE.'
- 'Requested'] == True
- assert props['org.freedesktop.Telepathy.Channel.FUTURE.'
- 'InitiatorHandle'] == self_handle
- assert props['org.freedesktop.Telepathy.Channel.FUTURE.'
- 'InitiatorID'] == 'test at localhost'
-
-
-def request_props(handle):
- return { 'org.freedesktop.Telepathy.Channel.ChannelType':
- 'org.freedesktop.Telepathy.Channel.Type.Text',
- 'org.freedesktop.Telepathy.Channel.TargetHandleType': 1,
- 'org.freedesktop.Telepathy.Channel.TargetHandle': handle,
- }
-
-
-if __name__ == '__main__':
- exec_test(test)
-
diff --git a/tests/twisted/text/ensure.py b/tests/twisted/text/ensure.py
new file mode 100644
index 0000000..cb68c16
--- /dev/null
+++ b/tests/twisted/text/ensure.py
@@ -0,0 +1,193 @@
+"""
+Test text channel initiated by me, using Requests.EnsureChannel
+"""
+
+import dbus
+
+from twisted.words.xish import domish
+
+from gabbletest import exec_test
+from servicetest import call_async, EventPattern
+
+def test(q, bus, conn, stream):
+ conn.Connect()
+ q.expect('dbus-signal', signal='StatusChanged', args=[0, 1])
+
+ self_handle = conn.GetSelfHandle()
+
+ jids = ['foo at bar.com', 'truc at cafe.fr']
+ call_async(q, conn, 'RequestHandles', 1, jids)
+
+ event = q.expect('dbus-return', method='RequestHandles')
+ handles = event.value[0]
+
+ properties = conn.GetAll(
+ 'org.freedesktop.Telepathy.Connection.Interface.Requests',
+ dbus_interface='org.freedesktop.DBus.Properties')
+ assert properties.get('Channels') == [], properties['Channels']
+ assert ({'org.freedesktop.Telepathy.Channel.ChannelType':
+ 'org.freedesktop.Telepathy.Channel.Type.Text',
+ 'org.freedesktop.Telepathy.Channel.TargetHandleType': 1,
+ },
+ ['org.freedesktop.Telepathy.Channel.TargetHandle',
+ 'org.freedesktop.Telepathy.Channel.TargetID'
+ ],
+ ) in properties.get('RequestableChannelClasses'),\
+ properties['RequestableChannelClasses']
+
+ requestotron = dbus.Interface(conn,
+ 'org.freedesktop.Telepathy.Connection.Interface.Requests')
+
+ test_ensure_ensure(q, requestotron, conn, self_handle, jids[0], handles[0])
+ test_request_ensure(q, requestotron, conn, self_handle, jids[1], handles[1])
+
+ conn.Disconnect()
+ q.expect('dbus-signal', signal='StatusChanged', args=[2, 1])
+
+
+def test_ensure_ensure(q, requestotron, conn, self_handle, jid, handle):
+ """
+ Test ensuring a non-existant channel twice. The first call should succeed
+ with Yours=True; the subsequent call should succeed with Yours=False
+ """
+
+ # Check that Ensuring a channel that doesn't exist succeeds
+ call_async(q, requestotron, 'EnsureChannel', request_props (handle))
+
+ ret, old_sig, new_sig = q.expect_many(
+ EventPattern('dbus-return', method='EnsureChannel'),
+ EventPattern('dbus-signal', signal='NewChannel'),
+ EventPattern('dbus-signal', signal='NewChannels'),
+ )
+
+ assert len(ret.value) == 3
+ yours, path, emitted_props = ret.value
+
+ # The channel was created in response to the call, and we were the only
+ # requestor, so we should get Yours=True
+ assert yours, ret.value
+
+ check_props(emitted_props, self_handle, handle, jid)
+
+ assert len(old_sig.args) == 5
+ old_path, old_ct, old_ht, old_h, old_sh = old_sig.args
+
+ assert old_path == path
+ assert old_ct == u'org.freedesktop.Telepathy.Channel.Type.Text'
+ # check that handle type == contact handle
+ assert old_ht == 1
+ assert old_h == handle
+ assert old_sh == True # suppress handler
+
+ assert len(new_sig.args) == 1
+ assert len(new_sig.args[0]) == 1 # one channel
+ assert len(new_sig.args[0][0]) == 2 # two struct members
+ assert new_sig.args[0][0][0] == path
+ assert new_sig.args[0][0][1] == emitted_props
+
+ properties = conn.GetAll(
+ 'org.freedesktop.Telepathy.Connection.Interface.Requests',
+ dbus_interface='org.freedesktop.DBus.Properties')
+
+ assert new_sig.args[0][0] in properties['Channels'], \
+ (new_sig.args[0][0], properties['Channels'])
+
+
+ # Now try Ensuring a channel which already exists
+ call_async(q, requestotron, 'EnsureChannel', request_props (handle))
+ ret_ = q.expect('dbus-return', method='EnsureChannel')
+
+ assert len(ret_.value) == 3
+ yours_, path_, emitted_props_ = ret_.value
+
+ # Someone's already responsible for this channel, so we should get
+ # Yours=False
+ assert not yours_, ret_.value
+ assert path == path_, (path, path_)
+ assert emitted_props == emitted_props_, (emitted_props, emitted_props_)
+
+
+def test_request_ensure(q, requestotron, conn, self_handle, jid, handle):
+ """
+ Test Creating a non-existant channel, then Ensuring the same channel.
+ The call to Ensure should succeed with Yours=False.
+ """
+
+ call_async(q, requestotron, 'CreateChannel', request_props (handle))
+
+ ret, old_sig, new_sig = q.expect_many(
+ EventPattern('dbus-return', method='CreateChannel'),
+ EventPattern('dbus-signal', signal='NewChannel'),
+ EventPattern('dbus-signal', signal='NewChannels'),
+ )
+
+ assert len(ret.value) == 2
+ path, emitted_props = ret.value
+
+ check_props(emitted_props, self_handle, handle, jid)
+
+ assert len(old_sig.args) == 5
+ old_path, old_ct, old_ht, old_h, old_sh = old_sig.args
+
+ assert old_path == path
+ assert old_ct == u'org.freedesktop.Telepathy.Channel.Type.Text'
+ # check that handle type == contact handle
+ assert old_ht == 1
+ assert old_h == handle
+ assert old_sh == True # suppress handler
+
+ assert len(new_sig.args) == 1
+ assert len(new_sig.args[0]) == 1 # one channel
+ assert len(new_sig.args[0][0]) == 2 # two struct members
+ assert new_sig.args[0][0][0] == path
+ assert new_sig.args[0][0][1] == emitted_props
+
+ properties = conn.GetAll(
+ 'org.freedesktop.Telepathy.Connection.Interface.Requests',
+ dbus_interface='org.freedesktop.DBus.Properties')
+
+ assert new_sig.args[0][0] in properties['Channels'], \
+ (new_sig.args[0][0], properties['Channels'])
+
+
+ # Now try Ensuring that same channel.
+ call_async(q, requestotron, 'EnsureChannel', request_props (handle))
+ ret_ = q.expect('dbus-return', method='EnsureChannel')
+
+ assert len(ret_.value) == 3
+ yours_, path_, emitted_props_ = ret_.value
+
+ # Someone's already responsible for this channel, so we should get
+ # Yours=False
+ assert not yours_, ret_.value
+ assert path == path_, (path, path_)
+ assert emitted_props == emitted_props_, (emitted_props, emitted_props_)
+
+
+def check_props(props, self_handle, handle, jid):
+ assert props['org.freedesktop.Telepathy.Channel.ChannelType'] ==\
+ 'org.freedesktop.Telepathy.Channel.Type.Text'
+ assert props['org.freedesktop.Telepathy.Channel.'
+ 'TargetHandleType'] == 1
+ assert props['org.freedesktop.Telepathy.Channel.TargetHandle'] ==\
+ handle
+ assert props['org.freedesktop.Telepathy.Channel.TargetID'] == jid
+ assert props['org.freedesktop.Telepathy.Channel.FUTURE.'
+ 'Requested'] == True
+ assert props['org.freedesktop.Telepathy.Channel.FUTURE.'
+ 'InitiatorHandle'] == self_handle
+ assert props['org.freedesktop.Telepathy.Channel.FUTURE.'
+ 'InitiatorID'] == 'test at localhost'
+
+
+def request_props(handle):
+ return { 'org.freedesktop.Telepathy.Channel.ChannelType':
+ 'org.freedesktop.Telepathy.Channel.Type.Text',
+ 'org.freedesktop.Telepathy.Channel.TargetHandleType': 1,
+ 'org.freedesktop.Telepathy.Channel.TargetHandle': handle,
+ }
+
+
+if __name__ == '__main__':
+ exec_test(test)
+
--
1.5.6.5
More information about the Telepathy-commits
mailing list