[telepathy-mission-control/master] mctest: add an improved SimulatedConnection class

[Simon McVittie]

---
 test/twisted/mctest.py |   89 ++++++++++++++++++++++++++++++++++++++++++++++++
 1 files changed, 89 insertions(+), 0 deletions(-)

diff --git a/test/twisted/mctest.py b/test/twisted/mctest.py
index 1c2e065..7fba030 100644
--- a/test/twisted/mctest.py
+++ b/test/twisted/mctest.py
@@ -8,11 +8,13 @@ import os
 import sha
 import sys
 
+import constants as cs
 import servicetest
 import twisted
 from twisted.internet import reactor
 
 import dbus
+import dbus.service
 
 def make_mc(bus, event_func, params=None):
     default_params = {
@@ -87,3 +89,90 @@ def exec_test_deferred (fun, params, protocol=None, timeout=None):
 def exec_test(fun, params=None, protocol=None, timeout=None):
   reactor.callWhenRunning (exec_test_deferred, fun, params, protocol, timeout)
   reactor.run()
+
+class SimulatedConnection(object):
+
+    def ensure_handle(self, type, identifier):
+        if (type, identifier) in self._handles:
+            return self._handles[identifier]
+
+        self._last_handle += 1
+        self._handles[(type, identifier)] = self._last_handle
+        self._identifiers[(type, self._last_handle)] = identifier
+        return self._last_handle
+
+    def __init__(self, q, bus, cmname, protocol, account_part, self_ident):
+        self.q = q
+        self.bus = bus
+
+        self.bus_name = '.'.join([cs.tp_name_prefix, 'Connection',
+                cmname, protocol.replace('-', '_'), account_part])
+        self._bus_name_ref = dbus.service.BusName(self.bus_name, self.bus)
+        self.object_path = '/' + self.bus_name.replace('.', '/')
+
+        self._last_handle = 41
+        self._handles = {}
+        self._identifiers = {}
+        self.status = cs.CONN_STATUS_DISCONNECTED
+        self.reason = cs.CONN_STATUS_CONNECTING
+        self.self_ident = self_ident
+        self.self_handle = self.ensure_handle(cs.HT_CONTACT, self_ident)
+
+        q.add_dbus_method_impl(self.Connect,
+                path=self.object_path, interface=cs.CONN, method='Connect')
+        q.add_dbus_method_impl(self.GetSelfHandle,
+                path=self.object_path,
+                interface=cs.CONN, method='GetSelfHandle')
+        q.add_dbus_method_impl(self.GetStatus,
+                path=self.object_path, interface=cs.CONN, method='GetStatus')
+        q.add_dbus_method_impl(self.GetInterfaces,
+                path=self.object_path, interface=cs.CONN,
+                method='GetInterfaces')
+        q.add_dbus_method_impl(self.InspectHandles,
+                path=self.object_path, interface=cs.CONN,
+                method='InspectHandles')
+        q.add_dbus_method_impl(self.GetAll_Requests,
+                path=self.object_path,
+                interface=cs.PROPERTIES_IFACE, method='GetAll',
+                args=[cs.CONN_IFACE_REQUESTS])
+
+    def GetInterfaces(self, e):
+        # FIXME: when needed, allow altering this return somehow
+        self.q.dbus_return(e.message, [cs.CONN_IFACE_REQUESTS], signature='as')
+
+    def Connect(self, e):
+        self.q.dbus_emit(self.object_path, cs.CONN, 'StatusChanged',
+                cs.CONN_STATUS_CONNECTING, cs.CONN_STATUS_REASON_NONE,
+                signature='uu')
+        self.q.dbus_return(e.message, signature='')
+
+    def InspectHandles(self, e):
+        htype, hs = e.args
+        ret = []
+
+        for h in hs:
+            if (htype, h) in self._identifiers:
+                ret.append(self._identifiers[(htype, h)])
+            else:
+                self.q.dbus_raise(e.message, INVALID_HANDLE, str(h))
+                return
+
+        self.q.dbus_return(e.message, ret, signature='as')
+
+    def GetStatus(self, e):
+        self.q.dbus_return(e.message, self.status, signature='u')
+
+    def StatusChanged(self, status, reason):
+        self.status = status
+        self.reason = reason
+        self.q.dbus_emit(self.object_path, cs.CONN, 'StatusChanged',
+                status, reason, signature='uu')
+
+    def GetAll_Requests(self, e):
+        # FIXME: stub: assumes no channels
+        self.q.dbus_return(e.message, {
+            'Channels': dbus.Array(signature='(oa{sv})'),
+        }, signature='a{sv}')
+
+    def GetSelfHandle(self, e):
+        self.q.dbus_return(e.message, self.self_handle, signature='u')
-- 
1.5.6.5