Add a manual test for ConnectionManager and Protocol introspectabilityintrospectable

3 files changed, 441 insertions, 0 deletions

diff --git a/configure.ac b/configure.ac
index 2b29ff775..407545151 100644
--- a/configure.ac
+++ b/configure.ac
@@ -331,6 +331,7 @@ AC_OUTPUT( Makefile \
 	   tests/Makefile \
 	   tests/lib/Makefile \
 	   tests/dbus/Makefile \
+	   tests/python/Makefile \
 	   tests/tools/Makefile \
 	   tools/Makefile \
 	   m4/Makefile \
diff --git a/tests/python/Makefile.am b/tests/python/Makefile.am
new file mode 100644
index 000000000..6f8e7bc71
--- /dev/null
+++ b/tests/python/Makefile.am
@@ -0,0 +1,8 @@
+# for now these are just manual tests
+
+EXTRA_DIST = \
+	cm.py \
+	$(NULL)
+
+clean-local:
+	rm -f *.pyc *.pyo
diff --git a/tests/python/cm.py b/tests/python/cm.py
new file mode 100644
index 000000000..f9e09dfae
--- /dev/null
+++ b/tests/python/cm.py
@@ -0,0 +1,432 @@
+#!/usr/bin/env python3
+# encoding: utf-8
+
+# Copyright © 2013 Collabora Ltd. <http://www.collabora.co.uk/>
+#
+# This library is free software; you can redistribute it and/or
+# modify it under the terms of the GNU Lesser General Public
+# License as published by the Free Software Foundation; either
+# version 2.1 of the License, or (at your option) any later version.
+#
+# This library is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+# Lesser General Public License for more details.
+#
+# You should have received a copy of the GNU Lesser General Public
+# License along with this library; if not, write to the Free Software
+# Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA  02110-1301  USA
+
+import os
+import unittest
+
+from gi.repository import GLib
+from gi.repository import TelepathyGLib as Tp
+
+class ExampleProtocol(Tp.BaseProtocol,
+        Tp.IntrospectableBaseProtocol,
+        Tp.ProtocolAddressing):
+    def __init__(self, name):
+        super(ExampleProtocol, self).__init__(name=name)
+
+    def do_dup_manager_file_content(self):
+        return '\n'.join([
+            '[Protocol %s]' % self.props.name,
+            'Interfaces=%s;'
+                % ';'.join(['com.example.FooProto', 'org.example.BarProto',
+                    Tp.IFACE_PROTOCOL_INTERFACE_AVATARS,
+                    Tp.IFACE_PROTOCOL_INTERFACE_PRESENCE,
+                    Tp.IFACE_PROTOCOL_INTERFACE_ADDRESSING]),
+            'ConnectionInterfaces=com.example.FooConn;org.example.BarConn;',
+            'param-account=s required',
+            'param-server=s',
+            'param-port=u',
+            'default-port=6667',
+            'RequestableChannelClasses=text;',
+            'VCardField=x-exampleproto',
+            'Icon=im-exemplary',
+            'EnglishName=Example protocol',
+            'AuthenticationTypes=com.example.FooAuth;org.example.BarAuth;',
+            # Presence
+            'status-chatty=%d settable message' %
+                int(Tp.ConnectionPresenceType.AVAILABLE),
+            'status-available=%d settable' %
+                int(Tp.ConnectionPresenceType.AVAILABLE),
+            'status-away=%d settable message' %
+                int(Tp.ConnectionPresenceType.AWAY),
+            'status-offline=%d' % int(Tp.ConnectionPresenceType.OFFLINE),
+            'status-unknown=%d' % int(Tp.ConnectionPresenceType.UNKNOWN),
+            'status-error=%d' % int(Tp.ConnectionPresenceType.ERROR),
+            # Avatars
+            'SupportedAvatarMIMETypes=image/png;image/jpeg;',
+            'MinimumAvatarHeight=32',
+            'MinimumAvatarWidth=64',
+            'RecommendedAvatarHeight=48',
+            'RecommendedAvatarWidth=96',
+            'MaximumAvatarHeight=64',
+            'MaximumAvatarWidth=128',
+            'MaximumAvatarBytes=123456',
+            # Addressing
+            'AddressableVCardFields=tel;x-sip;',
+            'AddressableURISchemes=tel;sip;',
+            '',
+            '[text]',
+            '%s s=%s' % (Tp.PROP_CHANNEL_CHANNEL_TYPE,
+                Tp.IFACE_CHANNEL_TYPE_TEXT),
+            '%s u=%d' % (Tp.PROP_CHANNEL_TARGET_HANDLE_TYPE,
+                int(Tp.HandleType.CONTACT)),
+            'allowed=%s;%s;' % (Tp.PROP_CHANNEL_TARGET_HANDLE,
+                Tp.PROP_CHANNEL_TARGET_ID),
+            ''])
+
+    # FIXME: test
+    def do_introspectable_new_connection(self):
+        return (None, Tp.error_quark(), Tp.Error.NOT_IMPLEMENTED,
+                'Not implemented yet')
+
+    def do_introspectable_normalize_contact(self, contact):
+        if contact:
+            return (contact.lower(), 0, 0, None)
+        else:
+            return (None, Tp.error_quark(), Tp.Error.INVALID_HANDLE,
+                    'Contact is not valid')
+
+    def do_introspectable_identify_account(self, vardict):
+        return self.do_introspectable_normalize_contact(vardict['account'])
+
+    def do_introspectable_normalize_vcard_address(self, field, value):
+        if field.lower() not in ('x-sip', 'tel', 'x-exampleproto'):
+            return (None, Tp.error_quark(), Tp.Error.NOT_IMPLEMENTED,
+                "I don't know how to normalize %s" % field.lower())
+
+        # this is not really how you normalize SIP URIs or phone numbers,
+        # obviously...
+        return self.do_introspectable_normalize_contact(value)
+
+    def do_introspectable_normalize_contact_uri(self, uri):
+        scheme = GLib.uri_parse_scheme(uri)
+
+        if scheme is None:
+            return (None, Tp.error_quark(), Tp.Error.INVALID_ARGUMENT,
+                'Not a valid URI: %s' % uri)
+        elif scheme not in ('sip', 'tel'):
+            return (None, Tp.error_quark(), Tp.Error.NOT_IMPLEMENTED,
+                'I only implement sip: and tel: URIs, not "%s:"' % scheme)
+        else:
+            # this is not really how you normalize SIP URIs or phone numbers,
+            # obviously...
+            ret = self.do_introspectable_normalize_contact(uri[len(scheme) + 1:])
+
+            if ret[0] is None:
+                return ret
+            return (scheme + ':' + ret[0],) + ret[1:]
+
+class ExampleConnectionManager(Tp.BaseConnectionManager):
+    def __init__(self):
+        super(ExampleConnectionManager, self).__init__()
+        self.add_protocol(ExampleProtocol('example'))
+
+    def do_dup_cm_dbus_name(self):
+        return "example_python_cm"
+
+    def do_get_interfaces(self):
+        return (Tp.BaseConnectionManager.do_get_interfaces(self) +
+            ["com.example.FooCM", "org.example.BarCM"])
+
+class Waiter(object):
+    def __init__(self):
+        self.source = None
+        self.result = None
+
+    def callback(self, source, result, user_data):
+        self.source = source
+        self.result = result
+        assert user_data is self
+
+    def wait(self):
+        assert self.result is None
+        while self.result is None:
+            GLib.MainContext.default().iteration(True)
+
+class TestIntrospectedCM(unittest.TestCase):
+    def setUp(self):
+        os.environ.setdefault('G_MESSAGES_DEBUG', 'all')
+        Tp.debug_set_flags("all")
+
+        def variant_equals(a, b, msg=None):
+            if not a.equal(b):
+                if msg is not None:
+                    raise self.failureException(msg)
+                else:
+                    raise self.failureException('(%s) != (%s)' % (
+                        a.print_(True), b.print_(True)))
+
+        def assert_status_equals(a, b, msg=None):
+            if a.get_name() != b.get_name():
+                if msg is not None:
+                    raise self.failureException(msg)
+                else:
+                    raise self.failureException('%r != %r' % (
+                        a.get_name(), b.get_name()))
+
+            if a.get_presence_type() != b.get_presence_type():
+                if msg is not None:
+                    raise self.failureException(msg)
+                else:
+                    raise self.failureException('%r != %r' % (
+                        a.get_presence_type(), b.get_presence_type()))
+
+            if a.can_set_on_self() != b.can_set_on_self():
+                if msg is not None:
+                    raise self.failureException(msg)
+                else:
+                    raise self.failureException('%r != %r' % (
+                        a.can_set_on_self(), b.can_set_on_self()))
+
+            if a.has_message() != b.has_message():
+                if msg is not None:
+                    raise self.failureException(msg)
+                else:
+                    raise self.failureException('%r != %r' % (
+                        a.has_message(), b.has_message()))
+
+        self.addTypeEqualityFunc(GLib.Variant, variant_equals)
+        self.addTypeEqualityFunc(Tp.PresenceStatusSpec, assert_status_equals)
+
+        self.cm = None
+        self.dbus_daemon = Tp.DBusDaemon.dup()
+
+        cm_client = None
+
+    def assertSameSets(self, a, b, msg=None):
+        self.assertSetEqual(frozenset(a), frozenset(b), msg=msg)
+
+    def test_cm(self):
+        self.cm = ExampleConnectionManager()
+
+        self.assertEqual(self.cm.dup_cm_dbus_name(), "example_python_cm")
+        self.assertEqual(self.cm.get_dbus_daemon(), self.dbus_daemon)
+        self.assertEqual(self.cm.props.dbus_daemon, self.dbus_daemon)
+        self.assertEqual(self.cm.props.interfaces,
+                ["com.example.FooCM", "org.example.BarCM"])
+
+        self.cm.register()
+
+        # for interactive testing
+        while 'TP_PYTHON_TEST_INTERACTIVE' in os.environ:
+            GLib.MainContext.default().iteration(True)
+
+        cm_client = Tp.ConnectionManager.new(self.dbus_daemon,
+                self.cm.dup_cm_dbus_name(), None)
+
+        self.assertEqual(cm_client.get_name(), self.cm.dup_cm_dbus_name())
+
+        waiter = Waiter()
+        cm_client.prepare_async(None, waiter.callback, waiter)
+        waiter.wait()
+        cm_client.prepare_finish(waiter.result)
+
+        self.assertTrue(cm_client.is_running())
+        self.assertTrue(cm_client.has_interface("com.example.FooCM"))
+        self.assertTrue(cm_client.has_interface("org.example.BarCM"))
+        self.assertFalse(cm_client.has_interface("net.example.Bees"))
+
+        self.assertSameSets(cm_client.dup_protocol_names(), ["example"])
+        self.assertTrue(cm_client.has_protocol("example"))
+        self.assertFalse(cm_client.has_protocol("bees"))
+        protocol = cm_client.get_protocol_object("example")
+        protocols = cm_client.dup_protocols()
+        self.assertSameSets(protocols, [protocol])
+
+        self.assertEqual(protocol.get_name(), "example")
+        self.assertEqual(protocol.get_cm_name(), cm_client.get_name())
+        self.assertFalse(protocol.can_register())
+        self.assertTrue(protocol.has_interface("com.example.FooProto"))
+        self.assertTrue(protocol.has_interface("org.example.BarProto"))
+        self.assertFalse(protocol.has_interface("net.example.Bees"))
+
+        self.assertTrue(protocol.has_param("account"))
+        self.assertSameSets(protocol.dup_param_names(),
+                ["account", "server", "port"])
+
+        param = protocol.dup_param("account")
+        self.assertEqual(protocol.get_param("account").get_name(), "account")
+        self.assertEqual(param.get_name(), "account")
+        self.assertEqual(param.get_dbus_signature(), "s")
+        self.assertTrue(param.is_required())
+        self.assertFalse(param.is_required_for_registration())
+        self.assertFalse(param.is_secret())
+        self.assertFalse(param.is_dbus_property())
+        self.assertEqual(param.dup_default_variant(), None)
+
+        param = protocol.dup_param("server")
+        self.assertEqual(protocol.get_param("server").get_name(), "server")
+        self.assertEqual(param.get_name(), "server")
+        self.assertEqual(param.get_dbus_signature(), "s")
+        self.assertFalse(param.is_required())
+        self.assertFalse(param.is_required_for_registration())
+        self.assertFalse(param.is_secret())
+        self.assertFalse(param.is_dbus_property())
+        self.assertEqual(param.dup_default_variant(), None)
+
+        param = protocol.dup_param("port")
+        self.assertEqual(protocol.get_param("port").get_name(), "port")
+        self.assertEqual(param.get_name(), "port")
+        self.assertEqual(param.get_dbus_signature(), "u")
+        self.assertFalse(param.is_required())
+        self.assertFalse(param.is_required_for_registration())
+        self.assertFalse(param.is_secret())
+        self.assertFalse(param.is_dbus_property())
+        self.assertEqual(param.dup_default_variant(), GLib.Variant('u', 6667))
+
+        params = protocol.dup_params()
+        # a non-empty list is true, so...
+        self.assertTrue([p for p in params if p.get_name() == "account"])
+        # similarly...
+        self.assertFalse([p for p in params if p.get_name() == "bees"])
+
+        self.assertSameSets(protocol.get_authentication_types(),
+                ['com.example.FooAuth', 'org.example.BarAuth'])
+        self.assertEqual(protocol.get_vcard_field(), 'x-exampleproto')
+        self.assertEqual(protocol.get_english_name(), "Example protocol")
+        self.assertEqual(protocol.get_icon_name(), "im-exemplary")
+
+        caps = protocol.get_capabilities()
+        self.assertTrue(caps.supports_text_chats(),
+                caps.dup_channel_classes_variant().print_(True))
+        self.assertFalse(caps.supports_text_chatrooms())
+
+        reqs = protocol.get_avatar_requirements()
+        self.assertSameSets(reqs.supported_mime_types,
+                ['image/png', 'image/jpeg'])
+        self.assertEqual(reqs.minimum_height, 32)
+        self.assertEqual(reqs.minimum_width, 64)
+        self.assertEqual(reqs.recommended_height, 48)
+        self.assertEqual(reqs.recommended_width, 96)
+        self.assertEqual(reqs.maximum_height, 64)
+        self.assertEqual(reqs.maximum_width, 128)
+        self.assertEqual(reqs.maximum_bytes, 123456)
+
+        status_list = protocol.dup_presence_statuses()
+        statuses = {}
+        for s in status_list:
+            statuses[s.get_name()] = s
+        self.assertEqual(statuses['available'], Tp.PresenceStatusSpec.new(
+            'available', Tp.ConnectionPresenceType.AVAILABLE, True, False))
+        self.assertEqual(statuses['chatty'], Tp.PresenceStatusSpec.new(
+            'chatty', Tp.ConnectionPresenceType.AVAILABLE, True, True))
+        self.assertEqual(statuses['away'], Tp.PresenceStatusSpec.new(
+            'away', Tp.ConnectionPresenceType.AWAY, True, True))
+        self.assertEqual(statuses['offline'], Tp.PresenceStatusSpec.new(
+            'offline', Tp.ConnectionPresenceType.OFFLINE, False, False))
+        self.assertEqual(statuses['unknown'], Tp.PresenceStatusSpec.new(
+            'unknown', Tp.ConnectionPresenceType.UNKNOWN, False, False))
+        self.assertEqual(statuses['error'], Tp.PresenceStatusSpec.new(
+            'error', Tp.ConnectionPresenceType.ERROR, False, False))
+
+        waiter = Waiter()
+        protocol.normalize_contact_async("MiXeDcAsE", None,
+                waiter.callback, waiter)
+        waiter.wait()
+        self.assertEqual(protocol.normalize_contact_finish(waiter.result),
+                "mixedcase")
+
+        waiter = Waiter()
+        protocol.normalize_contact_async("", None, waiter.callback, waiter)
+        waiter.wait()
+        with self.assertRaises(GLib.GError) as trap:
+            protocol.normalize_contact_finish(waiter.result)
+        self.assertEqual(trap.exception.code, Tp.Error.INVALID_HANDLE)
+
+        waiter = Waiter()
+        protocol.normalize_vcard_address_async("x-sip", "MiXeDcAsE",
+                None, waiter.callback, waiter)
+        waiter.wait()
+        self.assertEqual(protocol.normalize_vcard_address_finish(waiter.result),
+                "mixedcase")
+
+        waiter = Waiter()
+        protocol.normalize_vcard_address_async("x-sip", "",
+                None, waiter.callback, waiter)
+        waiter.wait()
+        with self.assertRaises(GLib.GError) as trap:
+            protocol.normalize_vcard_address_finish(waiter.result)
+        self.assertEqual(trap.exception.code, Tp.Error.INVALID_HANDLE)
+
+        waiter = Waiter()
+        protocol.normalize_vcard_address_async("argh", "example",
+                None, waiter.callback, waiter)
+        waiter.wait()
+        with self.assertRaises(GLib.GError) as trap:
+            protocol.normalize_vcard_address_finish(waiter.result)
+        self.assertEqual(trap.exception.code, Tp.Error.NOT_IMPLEMENTED)
+
+        waiter = Waiter()
+        protocol.normalize_contact_uri_async("sip:MiXeDcAsE",
+                None, waiter.callback, waiter)
+        waiter.wait()
+        self.assertEqual(protocol.normalize_contact_uri_finish(waiter.result),
+                "sip:mixedcase")
+
+        waiter = Waiter()
+        protocol.normalize_contact_uri_async("sip:",
+                None, waiter.callback, waiter)
+        waiter.wait()
+        with self.assertRaises(GLib.GError) as trap:
+            protocol.normalize_contact_uri_finish(waiter.result)
+        self.assertEqual(trap.exception.code, Tp.Error.INVALID_HANDLE)
+
+        waiter = Waiter()
+        protocol.normalize_contact_uri_async("foo:bar",
+                None, waiter.callback, waiter)
+        waiter.wait()
+        with self.assertRaises(GLib.GError) as trap:
+            protocol.normalize_contact_uri_finish(waiter.result)
+        self.assertEqual(trap.exception.code, Tp.Error.NOT_IMPLEMENTED)
+
+        waiter = Waiter()
+        protocol.identify_account_async(GLib.Variant('a{sv}', {
+                'account': GLib.Variant('s', 'MiXeDcAsE'),
+            }), None ,waiter.callback, waiter)
+        waiter.wait()
+        self.assertEqual(protocol.identify_account_finish(waiter.result),
+                "mixedcase")
+
+        waiter = Waiter()
+        protocol.identify_account_async(GLib.Variant('a{sv}', {
+            }), None, waiter.callback, waiter)
+        waiter.wait()
+        with self.assertRaises(GLib.GError) as trap:
+            protocol.identify_account_finish(waiter.result)
+        self.assertEqual(trap.exception.code, Tp.Error.INVALID_ARGUMENT)
+
+        waiter = Waiter()
+        protocol.identify_account_async(GLib.Variant('a{sv}', {
+            'account': GLib.Variant('s', ''),
+            }), None, waiter.callback, waiter)
+        waiter.wait()
+        with self.assertRaises(GLib.GError) as trap:
+            protocol.identify_account_finish(waiter.result)
+        self.assertEqual(trap.exception.code, Tp.Error.INVALID_HANDLE)
+
+        waiter = Waiter()
+        protocol.identify_account_async(GLib.Variant('a{sv}', {
+            'account': GLib.Variant('s', ''),
+            'foo': GLib.Variant('u', 42),
+            }), None, waiter.callback, waiter)
+        waiter.wait()
+        with self.assertRaises(GLib.GError) as trap:
+            protocol.identify_account_finish(waiter.result)
+        self.assertEqual(trap.exception.code, Tp.Error.INVALID_ARGUMENT)
+
+        # FIXME: test CM.NewConnection (there's no medium-level API,
+        # and the high-level API needs an AccountManager)
+
+    def tearDown(self):
+        if self.cm is not None:
+            # FIXME: there's no unregister function, so delete it and
+            # hope for the best
+            self.cm = None
+
+if __name__ == '__main__':
+    unittest.main()