[telepathy-mission-control/master] Add a regression test for requesting channels on offline accounts
Simon McVittie
simon.mcvittie at collabora.co.uk
Thu Jun 25 05:57:53 PDT 2009
This correctly puts the account online.
---
test/twisted/Makefile.am | 1 +
test/twisted/dispatcher/connect-for-request.py | 146 ++++++++++++++++++++++++
2 files changed, 147 insertions(+), 0 deletions(-)
create mode 100644 test/twisted/dispatcher/connect-for-request.py
diff --git a/test/twisted/Makefile.am b/test/twisted/Makefile.am
index e1339c9..4997b58 100644
--- a/test/twisted/Makefile.am
+++ b/test/twisted/Makefile.am
@@ -21,6 +21,7 @@ TWISTED_BASIC_TESTS = \
dispatcher/bypass-approval.py \
dispatcher/cancel.py \
dispatcher/capture-bundle.py \
+ dispatcher/connect-for-request.py \
dispatcher/create-handler-fails.py \
dispatcher/create-lowmem.py \
dispatcher/create-no-preferred-handler.py \
diff --git a/test/twisted/dispatcher/connect-for-request.py b/test/twisted/dispatcher/connect-for-request.py
new file mode 100644
index 0000000..4f17d71
--- /dev/null
+++ b/test/twisted/dispatcher/connect-for-request.py
@@ -0,0 +1,146 @@
+import dbus
+import dbus.service
+
+from servicetest import EventPattern, tp_name_prefix, tp_path_prefix, \
+ call_async, sync_dbus
+from mctest import exec_test, create_fakecm_account, SimulatedConnection, \
+ SimulatedChannel
+import constants as cs
+
+def test(q, bus, mc):
+ account_manager = bus.get_object(cs.AM, cs.AM_PATH)
+ account_manager_iface = dbus.Interface(account_manager, cs.AM)
+
+ params = dbus.Dictionary({"account": "smcv at example.com",
+ "password": "secrecy"}, signature='sv')
+ (cm_name_ref, account) = create_fakecm_account(q, bus, mc, params)
+
+ account_iface = dbus.Interface(account, cs.ACCOUNT)
+ account_props = dbus.Interface(account, cs.PROPERTIES_IFACE)
+
+ # Ensure that it's enabled but has offline RP and doesn't connect
+ # automatically
+
+ call_async(q, account_props, 'Set', cs.ACCOUNT, 'RequestedPresence',
+ (dbus.UInt32(cs.PRESENCE_TYPE_OFFLINE), 'offline', ''))
+ q.expect('dbus-return', method='Set')
+ q.expect('dbus-signal', signal='AccountPropertyChanged',
+ predicate=lambda e:
+ e.args[0].get('RequestedPresence', (None, None, None))[1]
+ == 'offline')
+
+ call_async(q, account_props, 'Set', cs.ACCOUNT, 'AutomaticPresence',
+ (dbus.UInt32(cs.PRESENCE_TYPE_BUSY), 'busy',
+ 'Testing automatic presence'))
+ q.expect('dbus-return', method='Set')
+ q.expect('dbus-signal', signal='AccountPropertyChanged',
+ predicate=lambda e:
+ e.args[0].get('AutomaticPresence', (None, None, None))[1]
+ == 'busy')
+
+ call_async(q, account_props, 'Set', cs.ACCOUNT, 'ConnectAutomatically',
+ False)
+ q.expect('dbus-return', method='Set')
+
+ call_async(q, account_props, 'Set', cs.ACCOUNT, 'Enabled', True)
+ q.expect('dbus-return', method='Set')
+ q.expect('dbus-signal', signal='AccountPropertyChanged',
+ predicate=lambda e: e.args[0].get('Enabled'))
+
+ # Requesting a channel will put us online
+
+ user_action_time = dbus.Int64(1238582606)
+
+ cd = bus.get_object(cs.CD, cs.CD_PATH)
+
+ request = dbus.Dictionary({
+ cs.CHANNEL + '.ChannelType': cs.CHANNEL_TYPE_TEXT,
+ cs.CHANNEL + '.TargetHandleType': cs.HT_CONTACT,
+ cs.CHANNEL + '.TargetID': 'juliet',
+ }, signature='sv')
+ account_requests = dbus.Interface(account,
+ cs.ACCOUNT_IFACE_NOKIA_REQUESTS)
+ call_async(q, cd, 'CreateChannel',
+ account.object_path, request, user_action_time, "",
+ dbus_interface=cs.CD)
+ ret = q.expect('dbus-return', method='CreateChannel')
+ request_path = ret.value[0]
+
+ cr = bus.get_object(cs.AM, request_path)
+ request_props = cr.GetAll(cs.CR, dbus_interface=cs.PROPERTIES_IFACE)
+ assert request_props['Account'] == account.object_path
+ assert request_props['Requests'] == [request]
+ assert request_props['UserActionTime'] == user_action_time
+ assert request_props['PreferredHandler'] == ""
+ assert request_props['Interfaces'] == []
+
+ # make sure RequestConnection doesn't get called until we Proceed
+ events = [EventPattern('dbus-method-call', method='RequestConnection')]
+ q.forbid_events(events)
+
+ sync_dbus(bus, q, mc)
+
+ q.unforbid_events(events)
+
+ cr.Proceed(dbus_interface=cs.CR)
+
+ e = q.expect('dbus-method-call', method='RequestConnection',
+ args=['fakeprotocol', params],
+ destination=cs.tp_name_prefix + '.ConnectionManager.fakecm',
+ path=cs.tp_path_prefix + '/ConnectionManager/fakecm',
+ interface=cs.tp_name_prefix + '.ConnectionManager',
+ handled=False)
+
+ conn = SimulatedConnection(q, bus, 'fakecm', 'fakeprotocol', '_',
+ 'myself', has_requests=True, has_presence=True)
+
+ q.dbus_return(e.message, conn.bus_name, conn.object_path, signature='so')
+
+ q.expect('dbus-method-call', method='Connect',
+ path=conn.object_path, handled=True)
+ conn.StatusChanged(cs.CONN_STATUS_CONNECTED, cs.CONN_STATUS_REASON_NONE)
+ conn.presence = dbus.Struct((cs.PRESENCE_TYPE_AVAILABLE, 'available', ''),
+ signature='uss')
+
+ _, _, cm_request_call = q.expect_many(
+ EventPattern('dbus-method-call', path=conn.object_path,
+ interface=cs.PROPERTIES_IFACE, method='GetAll',
+ args=[cs.CONN_IFACE_REQUESTS], handled=True),
+ EventPattern('dbus-method-call', path=conn.object_path,
+ interface=cs.CONN_IFACE_SIMPLE_PRESENCE, method='SetPresence',
+ args=['busy', 'Testing automatic presence'], handled=True),
+ EventPattern('dbus-method-call', path=conn.object_path,
+ interface=cs.CONN_IFACE_REQUESTS, method='CreateChannel',
+ args=[request], handled=False),
+ )
+
+ q.dbus_emit(conn.object_path, cs.CONN_IFACE_SIMPLE_PRESENCE,
+ 'PresencesChanged',
+ {conn.self_handle: (dbus.UInt32(cs.PRESENCE_TYPE_BUSY), 'busy',
+ 'Testing automatic presence')},
+ signature='a{u(uss)}')
+
+ # Time passes. A channel is returned.
+
+ channel_immutable = dbus.Dictionary(request)
+ channel_immutable[cs.CHANNEL + '.InitiatorID'] = conn.self_ident
+ channel_immutable[cs.CHANNEL + '.InitiatorHandle'] = conn.self_handle
+ channel_immutable[cs.CHANNEL + '.Requested'] = True
+ channel_immutable[cs.CHANNEL + '.Interfaces'] = \
+ dbus.Array([], signature='s')
+ channel_immutable[cs.CHANNEL + '.TargetHandle'] = \
+ conn.ensure_handle(cs.HT_CONTACT, 'juliet')
+ channel = SimulatedChannel(conn, channel_immutable)
+
+ # this order of events is guaranteed by telepathy-spec (since 0.17.14)
+ q.dbus_return(cm_request_call.message,
+ channel.object_path, channel.immutable, signature='oa{sv}')
+ channel.announce()
+
+ # there's no handler, so it gets shot down
+
+ q.expect('dbus-method-call', path=channel.object_path, method='Close',
+ handled=True)
+
+if __name__ == '__main__':
+ exec_test(test, {})
--
1.5.6.5
More information about the telepathy-commits
mailing list