From 3a5a98bee241f63be9008a900c484cb427cad51f Mon Sep 17 00:00:00 2001 From: Jonny Lamb Date: Tue, 30 Jul 2013 16:43:13 +0200 Subject: split phoenix-test into phoenix-test-text and -call --- .gitignore | 6 +- src/Makefile.am | 21 ++- src/phoenix-test-call.in | 3 + src/phoenix-test-call.py | 65 +++++++++ src/phoenix-test-text.in | 3 + src/phoenix-test-text.py | 47 +++++++ src/phoenix-test.in | 3 - src/phoenix-test.py | 348 ----------------------------------------------- src/test.py | 285 ++++++++++++++++++++++++++++++++++++++ 9 files changed, 422 insertions(+), 359 deletions(-) create mode 100644 src/phoenix-test-call.in create mode 100755 src/phoenix-test-call.py create mode 100644 src/phoenix-test-text.in create mode 100755 src/phoenix-test-text.py delete mode 100644 src/phoenix-test.in delete mode 100755 src/phoenix-test.py create mode 100644 src/test.py diff --git a/.gitignore b/.gitignore index cc63ba1..c17e29b 100644 --- a/.gitignore +++ b/.gitignore @@ -37,6 +37,8 @@ src/phoenix-authenticator src/phoenix-echo-call src/phoenix src/phoenix-uninstalled -src/phoenix-test -src/phoenix-test-uninstalled +src/phoenix-test-text +src/phoenix-test-text-uninstalled +src/phoenix-test-call +src/phoenix-test-call-uninstalled *.o diff --git a/src/Makefile.am b/src/Makefile.am index 6bbc9c3..3733329 100644 --- a/src/Makefile.am +++ b/src/Makefile.am @@ -4,12 +4,12 @@ EXTRA_DIST = \ $(python_DATA) \ $(libexec_SCRIPTS) -bin_SCRIPTS=phoenix phoenix-test -bin_SCRIPTS_IN=phoenix.in phoenix-test.in -BUILT_SOURCES=phoenix-uninstalled phoenix-test-uninstalled +bin_SCRIPTS=phoenix phoenix-test-text phoenix-test-call +bin_SCRIPTS_IN=phoenix.in phoenix-test-text.in phoenix-test-call.in +BUILT_SOURCES=phoenix-uninstalled phoenix-test-text-uninstalled phoenix-test-call-uninstalled pythondir = $(pkgdatadir) -python_DATA=phoenix.py phoenix-test.py util.py +python_DATA=phoenix.py phoenix-test-text.py phoenix-test-call test.py util.py libexec_PROGRAMS = phoenix-authenticator phoenix-echo-call libexec_SCRIPTS = phoenix-echo-text phoenix-approver @@ -36,11 +36,20 @@ phoenix-uninstalled: phoenix.in -e "s,[@]PATH[@],@abs_top_builddir@/src," < $< > $@ @chmod +x $@ -phoenix-test: phoenix-test.in +phoenix-test-text: phoenix-test-text.in sed -e "s,[@]DATADIR[@],$(pkgdatadir)/data," \ -e "s,[@]PATH[@],$(pkgdatadir)," < $< > $@ -phoenix-test-uninstalled: phoenix-test.in +phoenix-test-text-uninstalled: phoenix-test-text.in + sed -e "s,[@]DATADIR[@],@abs_top_builddir@/data/uninstalled," \ + -e "s,[@]PATH[@],@abs_top_builddir@/src," < $< > $@ + @chmod +x $@ + +phoenix-test-call: phoenix-test-call.in + sed -e "s,[@]DATADIR[@],$(pkgdatadir)/data," \ + -e "s,[@]PATH[@],$(pkgdatadir)," < $< > $@ + +phoenix-test-call-uninstalled: phoenix-test-call.in sed -e "s,[@]DATADIR[@],@abs_top_builddir@/data/uninstalled," \ -e "s,[@]PATH[@],@abs_top_builddir@/src," < $< > $@ @chmod +x $@ diff --git a/src/phoenix-test-call.in b/src/phoenix-test-call.in new file mode 100644 index 0000000..05e2a79 --- /dev/null +++ b/src/phoenix-test-call.in @@ -0,0 +1,3 @@ +#!/bin/sh + +exec python "@PATH@/phoenix-test-call.py" --datadir=@DATADIR@ $@ diff --git a/src/phoenix-test-call.py b/src/phoenix-test-call.py new file mode 100755 index 0000000..6342eed --- /dev/null +++ b/src/phoenix-test-call.py @@ -0,0 +1,65 @@ +#!/usr/bin/env python +# -*- coding: utf-8 -*- + +import sys + +from gi.repository import GObject, Gio, GLib +from gi.repository import TelepathyGLib as Tp + +from test import main, TestConnection + + +class CallConnection(TestConnection): + def __init__(self, t, connection): + TestConnection.__init__(self, t, connection) + self.call_success = False + + def hangup(self): + self.channel.hangup_async(Tp.CallStateChangeReason.USER_REQUESTED, + '', '', None, None) + + def check_call_status(self, *args): + (state, flags, details, reason) = self.channel.get_state() + got_audio = self.proxy.get_cached_property('ReceivingAudio') + got_video = self.proxy.get_cached_property('ReceivingVideo') + + if state == Tp.CallState.ACTIVE and got_audio and got_video: + self.t.write('Successful call, letting it run for 5 seconds') + self.call_success = True + GLib.timeout_add_seconds(5, self.hangup) + + if state == Tp.CallState.ENDED: + self.t.assertEqual(True, self.call_success) + self.t.assertEqual(Tp.CallStateChangeReason.USER_REQUESTED, + reason.reason) + self.t.assertEqual(self.connection.get_self_handle(), + reason.actor) + self.t.done() + + def create_channel_finished(self, req, r, u): + TestConnection.create_channel_finished(self, req, r, u) + + d = Gio.bus_get_sync(Gio.BusType.SESSION, None) + self.proxy = proxy = Gio.DBusProxy.new_sync(d, 0, None, + 'org.freedesktop.Telepathy.Phoenix.Calls', + '/org/freedesktop/Telepathy/Phoenix/Calls' + + self.channel.get_object_path(), + 'org.freedesktop.Telepathy.Phoenix.CallInfo', + None) + proxy.connect('g-properties-changed', + self.check_call_status, None) + self.channel.connect('notify::state', + self.check_call_status, None) + + def handle_capabilities(self): + account = self.contact.get_account() + + req = Tp.AccountChannelRequest.new_audio_call(account, 0) + req.set_hint('call-mode', GLib.Variant('s', 'test-inputs')) + req.set_target_contact(self.contact) + req.create_and_observe_channel_async('', None, + self.create_channel_finished, None) + + +if __name__ == '__main__': + main(sys.argv, 'VOIP', CallConnection) diff --git a/src/phoenix-test-text.in b/src/phoenix-test-text.in new file mode 100644 index 0000000..92aac21 --- /dev/null +++ b/src/phoenix-test-text.in @@ -0,0 +1,3 @@ +#!/bin/sh + +exec python "@PATH@/phoenix-test-text.py" --datadir=@DATADIR@ $@ diff --git a/src/phoenix-test-text.py b/src/phoenix-test-text.py new file mode 100755 index 0000000..56b3b63 --- /dev/null +++ b/src/phoenix-test-text.py @@ -0,0 +1,47 @@ +#!/usr/bin/env python +# -*- coding: utf-8 -*- + +import sys + +from gi.repository import TelepathyGLib as Tp + +from test import main, TestConnection + +MESSAGE = 'Test from telepathy-phoenix' + + +class TextConnection(TestConnection): + def message_sent(self, channel, message, flags, token, data): + (message, flags) = message.to_text() + + self.t.assertEqual(MESSAGE, message) + + self.channel.close_async(None, None) + self.t.done() + + def create_channel_finished(self, req, r, u): + TestConnection.create_channel_finished(self, req, r, u) + + self.channel.connect('message-sent', + self.message_sent, None) + + m = Tp.ClientMessage.new_text(Tp.ChannelTextMessageType.NORMAL, MESSAGE) + self.channel.send_message_async(m, 0, None, None) + + def handle_capabilities(self): + caps = self.contact.get_capabilities() + account = self.contact.get_account() + + if not caps.supports_text_chats(): + print 'no caps for text chats' + self.t.done(False) + return + + req = Tp.AccountChannelRequest.new_text(account, 0) + req.set_target_contact(self.contact) + req.create_and_observe_channel_async('', None, + self.create_channel_finished, None) + + +if __name__ == '__main__': + main(sys.argv, 'Text', TextConnection) diff --git a/src/phoenix-test.in b/src/phoenix-test.in deleted file mode 100644 index ae9a869..0000000 --- a/src/phoenix-test.in +++ /dev/null @@ -1,3 +0,0 @@ -#!/bin/sh - -exec python "@PATH@/phoenix-test.py" --datadir=@DATADIR@ $@ diff --git a/src/phoenix-test.py b/src/phoenix-test.py deleted file mode 100755 index d803fbe..0000000 --- a/src/phoenix-test.py +++ /dev/null @@ -1,348 +0,0 @@ -#!/usr/bin/env python -# -*- coding: utf-8 -*- - -import os -import sys -import getopt -import tempfile -import shutil -import time -import signal -from util import spawnbus, setup_data_dir -from util import setup_run_dir, create_account, scrub_env - -from gi.repository import GObject, Gio, GLib -from gi.repository import TelepathyGLib as Tp - -MESSAGE = 'Test from TpRing' - - -class TestCase: - def __init__(self, loop, test_contact, quiet=False): - self.loop = loop - self.quiet = quiet - self.test_contact = test_contact - - def assertNotEqual(self, expected, value): - if expected == value: - self.done(False) - raise AssertionError("didn't expect: %s, got: %s", - str(expected), str(value)) - - def assertEqual(self, expected, value): - if expected != value: - self.done(False) - raise AssertionError('expected: %s, got: %s', - str(expected), str(value)) - - def assertNotNone(self, value): - if value == None: - self.done(False) - raise AssertionError('value was None') - - def done(self, success=True): - self.loop.quit() - if not success: - print 'VOIP Test: FAILED' - else: - print 'VOIP Test: PASS' - - def bite(self): - # Bitten by the watchdog - print 'Timeout hit' - self.done(False) - - def set_timeout(self, timeout=10): - GLib.timeout_add_seconds(timeout, self.bite) - - def write(self, string): - if not self.quiet: - print string - - def set_account(self, cm, protocol, *settings): - self.cm = cm - self.protocol = protocol - self.settings = {} - self.password = None - - for s in settings: - (k, v) = s.split('=', 2) - (t, v) = v.split(':', 2) - - if k == 'password': - self.password = v - continue - - if t == 'b': - variant = GLib.Variant(t, - v in [ '1', 'True', 'true', 't']) - else: - variant = GLib.Variant(t, v) - - self.settings[k] = variant - - -# Watch one Telepathy connection -class TestConnection: - def __init__(self, t, connection): - self.t = t - self.connection = connection - self.contact = None - self.fully_subscribed = False - self.call_success = False - - self.contact_list = self.connection.has_interface( - Tp.IFACE_CONNECTION_INTERFACE_CONTACT_LIST) - - if self.contact_list: - self.connection.connect('notify::contact-list-state', - self.contact_list_state_cb, None) - - if connection.get_contact_list_state() \ - == Tp.ContactListState.SUCCESS: - self.start_test() - else: - self.start_test() - - def contact_list_state_cb(self, connection, spec, data): - if connection.get_contact_list_state() \ - == Tp.ContactListState.SUCCESS: - self.start_test() - - def teardown_cb(self, contact, result, u): - contact.remove_finish(result) - - def teardown(self): - self.contact.remove_async(self.teardown_cb, None) - - def start_test(self): - self.connection.dup_contact_by_id_async(self.t.test_contact, - [Tp.ContactFeature.SUBSCRIPTION_STATES, - Tp.ContactFeature.CAPABILITIES], - self.got_test_contact, None) - - def hangup(self): - self.channel.hangup_async(Tp.CallStateChangeReason.USER_REQUESTED, - '', '', None, None) - - def check_call_status(self, *args): - (state, flags, details, reason) = self.channel.get_state() - got_audio = self.proxy.get_cached_property('ReceivingAudio') - got_video = self.proxy.get_cached_property('ReceivingVideo') - - if state == Tp.CallState.ACTIVE and got_audio and got_video: - self.t.write('Successful call, letting it run for 5 seconds') - self.call_success = True - GLib.timeout_add_seconds(5, self.hangup) - - if state == Tp.CallState.ENDED: - self.t.assertEqual(True, self.call_success) - self.t.assertEqual(Tp.CallStateChangeReason.USER_REQUESTED, - reason.reason) - self.t.assertEqual(self.connection.get_self_handle(), - reason.actor) - self.t.done() - - def message_sent(self, channel, message, flags, token, data): - (message, flags) = message.to_text() - - self.t.assertEqual(MESSAGE, message) - - self.channel.close_async(None, None) - self.t.done() - - def create_channel_finished(self, req, r, u): - try: - self.channel = channel = \ - req.create_and_observe_channel_finish(r) - except Exception, e: - print e - self.t.done(False) - - if isinstance(self.channel, Tp.CallChannel): - d = Gio.bus_get_sync(Gio.BusType.SESSION, None) - self.proxy = proxy = Gio.DBusProxy.new_sync(d, 0, None, - 'org.freedesktop.Telepathy.Phoenix.Calls', - '/org/freedesktop/Telepathy/Phoenix/Calls' + - channel.get_object_path(), - 'org.freedesktop.Telepathy.Phoenix.CallInfo', - None) - proxy.connect('g-properties-changed', - self.check_call_status, None) - channel.connect('notify::state', - self.check_call_status, None) - elif isinstance(self.channel, Tp.TextChannel): - channel.connect('message-sent', - self.message_sent, None) - - m = Tp.ClientMessage.new_text(Tp.ChannelTextMessageType.NORMAL, MESSAGE) - self.channel.send_message_async(m, 0, None, None) - else: - print 'unknown channel type created:', self.channel - self.t.done(Fail) - - def handle_capabilities(self): - caps = self.contact.get_capabilities() - account = self.contact.get_account() - - if caps.supports_audio_video_call(Tp.HandleType.CONTACT): - req = Tp.AccountChannelRequest.new_audio_video_call(account, 0) - req.set_hint('call-mode', GLib.Variant('s', 'test-inputs')) - elif caps.supports_text_chats(): - req = Tp.AccountChannelRequest.new_text(account, 0) - else: - print 'no caps for call or text chat' - self.t.done(False) - return - - req.set_target_contact(self.contact) - req.create_and_observe_channel_async('', None, - self.create_channel_finished, None) - - def handle_test_contact_states(self): - p = self.contact.get_publish_state() - s = self.contact.get_subscribe_state() - - if p == Tp.SubscriptionState.YES and s == Tp.SubscriptionState.YES: - if not self.fully_subscribed: - self.fully_subscribed = True - self.t.write('Subscription complete') - elif p == Tp.SubscriptionState.ASK: - self.contact.authorize_publication_async( - lambda c, r, u: c.authorize_publication_finish(r), None) - - def contact_states_cb(self, c, spec, d): - self.handle_test_contact_states() - - def got_test_contact(self, c, r, u): - self.contact = c.dup_contact_by_id_finish(r) - - if self.contact_list: - s = self.contact.get_subscribe_state() - - # If we're not subscribed, subscribe, needed to make calls for example - if s != Tp.SubscriptionState.YES: - print 'Asking for subscription' - self.contact.request_subscription_async('Test subscription', - lambda c, r, u: c.request_subscription_finish(r) , - None) - - self.contact.connect('notify::publish-state', - self.contact_states_cb, None) - self.contact.connect('notify::subscribe-state', - self.contact_states_cb, None) - - self.contact.connect('notify::capabilities', - lambda *x: self.handle_capabilities(), None) - self.handle_capabilities() - self.handle_test_contactf_states() - - -class TestAccount: - def __init__(self, t, account): - self.t = t - self.account = account - - self.t.assertEqual(False, account.is_enabled()) - - self.connection_notify_id = \ - self.account.connect('notify::connection', - self.connection_cb, None) - - # Enable & Request availability, causing our connection to be setup - self.account.set_enabled_async(True, None, None) - self.account.request_presence_async( - Tp.ConnectionPresenceType.AVAILABLE, - '', '', - None, None) - - def connection_test_if_possible(self): - s = self.account.get_property('connection-status') - self.t.assertNotEqual(Tp.ConnectionStatus.DISCONNECTED, s) - - self.account.disconnect(self.connection_notify_id) - - c = self.account.get_property('connection') - if s == Tp.ConnectionStatus.CONNECTED and c != None: - TestConnection(self.t, c) - - def connection_cb(self, account, spec, data): - self.connection_test_if_possible() - - -# Watch the account manager -class TestManager: - def __init__(self, t): - self.t = t - self.am = am = Tp.AccountManager.dup() - factory = am.get_factory() - factory.add_account_features([Tp.Account.get_feature_quark_connection()]) - factory.add_connection_features( - [Tp.Connection.get_feature_quark_contact_list()]) - self.accounts = {} - - am.prepare_async(None, self.prepared, None) - - def got_account(self, account): - self.t.assertNotNone(account) - TestAccount(self.t, account) - - def prepared(self, am, result, data): - # We start in a temporary session without existing accounts - self.t.assertEqual([], am.get_valid_accounts()) - - # Create a fresh acocunt - create_account(self.am, - self.t.cm, self.t.protocol, - 'Phoenix Test Account', - self.t.settings, - self.t.password, - self.got_account) - -if __name__ == '__main__': - quiet = False - datadir = None - testcontact = None - try: - (opts, args) = getopt.getopt(sys.argv[1:], - 'q', ['datadir=', 'quiet', 'testcontact=']) - for (o, a) in opts: - if o == '--datadir': - datadir = a - elif o in [ '-q', '--quiet']: - quiet = True - elif o == '--testcontact': - testcontact = a - if testcontact == None: - print 'Testcontact option is required' - sys.exit(2) - if len(args) < 2: - print 'usage: %s connection-manager protocol [settings]' % sys.argv[0] - sys.exit(2) - except getopt.GetoptError, err: - print str(err) - sys.exit(2) - - # Use a temporary directory for the testrun - tempdir = tempfile.mkdtemp(prefix='phoenix-test') - setup_run_dir(tempdir, quiet=quiet) - if datadir != None: - setup_data_dir(datadir, quiet) - - scrub_env() - p = spawnbus(quiet) - loop = GObject.MainLoop() - t = TestCase(loop, testcontact, quiet) - t.set_timeout(30) - t.set_account(*args) - - Tp.debug_set_flags(os.getenv('PHOENIX_TEST_DEBUG', '')) - - m = TestManager(t) - try: - loop.run() - finally: - os.kill(p.pid, signal.SIGKILL) - # Sleep 2 seconds so everything can die a nice death - time.sleep(2) - shutil.rmtree(tempdir) diff --git a/src/test.py b/src/test.py new file mode 100644 index 0000000..dd32478 --- /dev/null +++ b/src/test.py @@ -0,0 +1,285 @@ +# -*- coding: utf-8 -*- + +import os +import sys +import getopt +import tempfile +import shutil +import time +import signal +from util import spawnbus, setup_data_dir +from util import setup_run_dir, create_account, scrub_env + +from gi.repository import GObject, Gio, GLib +from gi.repository import TelepathyGLib as Tp + + +class TestCase: + def __init__(self, loop, test_contact, name, connection_type, quiet=False): + self.loop = loop + self.quiet = quiet + self.test_contact = test_contact + self.name = name + self.connection_type = connection_type + + def assertNotEqual(self, expected, value): + if expected == value: + self.done(False) + raise AssertionError("didn't expect: %s, got: %s", + str(expected), str(value)) + + def assertEqual(self, expected, value): + if expected != value: + self.done(False) + raise AssertionError('expected: %s, got: %s', + str(expected), str(value)) + + def assertNotNone(self, value): + if value == None: + self.done(False) + raise AssertionError('value was None') + + def done(self, success=True): + self.loop.quit() + if not success: + print '%s Test: FAILED' % self.name + else: + print '%s Test: PASS' % self.name + + def bite(self): + # Bitten by the watchdog + print 'Timeout hit' + self.done(False) + + def set_timeout(self, timeout=10): + GLib.timeout_add_seconds(timeout, self.bite) + + def write(self, string): + if not self.quiet: + print string + + def set_account(self, cm, protocol, *settings): + self.cm = cm + self.protocol = protocol + self.settings = {} + self.password = None + + for s in settings: + (k, v) = s.split('=', 2) + (t, v) = v.split(':', 2) + + if k == 'password': + self.password = v + continue + + if t == 'b': + variant = GLib.Variant(t, + v in [ '1', 'True', 'true', 't']) + else: + variant = GLib.Variant(t, v) + + self.settings[k] = variant + + +# Watch one Telepathy connection +class TestConnection: + def __init__(self, t, connection): + self.t = t + self.connection = connection + self.contact = None + self.fully_subscribed = False + + self.contact_list = self.connection.has_interface( + Tp.IFACE_CONNECTION_INTERFACE_CONTACT_LIST) + + if self.contact_list: + self.connection.connect('notify::contact-list-state', + self.contact_list_state_cb, None) + + if connection.get_contact_list_state() \ + == Tp.ContactListState.SUCCESS: + self.start_test() + else: + self.start_test() + + def contact_list_state_cb(self, connection, spec, data): + if connection.get_contact_list_state() \ + == Tp.ContactListState.SUCCESS: + self.start_test() + + def teardown_cb(self, contact, result, u): + contact.remove_finish(result) + + def teardown(self): + self.contact.remove_async(self.teardown_cb, None) + + def start_test(self): + self.connection.dup_contact_by_id_async(self.t.test_contact, + [Tp.ContactFeature.SUBSCRIPTION_STATES, + Tp.ContactFeature.CAPABILITIES], + self.got_test_contact, None) + + def create_channel_finished(self, req, r, u): + try: + self.channel = channel = \ + req.create_and_observe_channel_finish(r) + except Exception, e: + print e + self.t.done(False) + + def handle_capabilities(self): + raise NotImplemented('implement handle_capabilities') + + def handle_test_contact_states(self): + p = self.contact.get_publish_state() + s = self.contact.get_subscribe_state() + + if p == Tp.SubscriptionState.YES and s == Tp.SubscriptionState.YES: + if not self.fully_subscribed: + self.fully_subscribed = True + self.t.write('Subscription complete') + elif p == Tp.SubscriptionState.ASK: + self.contact.authorize_publication_async( + lambda c, r, u: c.authorize_publication_finish(r), None) + + def contact_states_cb(self, c, spec, d): + self.handle_test_contact_states() + + def got_test_contact(self, c, r, u): + self.contact = c.dup_contact_by_id_finish(r) + + if self.contact_list: + s = self.contact.get_subscribe_state() + + # If we're not subscribed, subscribe, needed to make calls for example + if s != Tp.SubscriptionState.YES: + print 'Asking for subscription' + self.contact.request_subscription_async('Test subscription', + lambda c, r, u: c.request_subscription_finish(r) , + None) + + self.contact.connect('notify::publish-state', + self.contact_states_cb, None) + self.contact.connect('notify::subscribe-state', + self.contact_states_cb, None) + + self.contact.connect('notify::capabilities', + lambda *x: self.handle_capabilities(), None) + self.handle_capabilities() + self.handle_test_contact_states() + +class TestAccount: + def __init__(self, t, account): + self.t = t + self.account = account + + self.t.assertEqual(False, account.is_enabled()) + + self.connection_notify_id = \ + self.account.connect('notify::connection', + self.connection_cb, None) + self.connection_status_id = \ + self.account.connect("notify::connection-status", + self.connection_status_cb, None) + + + # Enable & Request availability, causing our connection to be setup + self.account.set_enabled_async(True, None, None) + self.account.request_presence_async( + Tp.ConnectionPresenceType.AVAILABLE, + '', '', + None, None) + + def connection_test_if_possible(self): + s = self.account.get_property('connection-status') + self.t.assertNotEqual(Tp.ConnectionStatus.DISCONNECTED, s) + + c = self.account.get_property('connection') + if s == Tp.ConnectionStatus.CONNECTED and c != None: + self.t.connection_type(self.t, c) + + self.account.disconnect(self.connection_notify_id) + self.account.disconnect(self.connection_status_id) + + def connection_cb(self, account, spec, data): + self.connection_test_if_possible() + + def connection_status_cb(self, account, spec, data): + self.connection_test_if_possible() + +# Watch the account manager +class TestManager: + def __init__(self, t): + self.t = t + self.am = am = Tp.AccountManager.dup() + factory = am.get_factory() + factory.add_account_features([Tp.Account.get_feature_quark_connection()]) + factory.add_connection_features( + [Tp.Connection.get_feature_quark_contact_list()]) + self.accounts = {} + + am.prepare_async(None, self.prepared, None) + + def got_account(self, account): + self.t.assertNotNone(account) + TestAccount(self.t, account) + + def prepared(self, am, result, data): + # We start in a temporary session without existing accounts + self.t.assertEqual([], am.get_valid_accounts()) + + # Create a fresh acocunt + create_account(self.am, + self.t.cm, self.t.protocol, + 'Phoenix Test Account', + self.t.settings, + self.t.password, + self.got_account) + +def main(argv, name, connection_type): + quiet = False + datadir = None + testcontact = None + try: + (opts, args) = getopt.getopt(argv[1:], + 'q', ['datadir=', 'quiet', 'testcontact=']) + for (o, a) in opts: + if o == '--datadir': + datadir = a + elif o in [ '-q', '--quiet']: + quiet = True + elif o == '--testcontact': + testcontact = a + if testcontact == None: + print 'Testcontact option is required' + sys.exit(2) + if len(args) < 2: + print 'usage: %s connection-manager protocol [settings]' % argv[0] + sys.exit(2) + except getopt.GetoptError, err: + print str(err) + sys.exit(2) + + # Use a temporary directory for the testrun + tempdir = tempfile.mkdtemp(prefix='phoenix-test') + setup_run_dir(tempdir, quiet=quiet) + if datadir != None: + setup_data_dir(datadir, quiet) + + scrub_env() + p = spawnbus(quiet) + loop = GObject.MainLoop() + t = TestCase(loop, testcontact, name, connection_type, quiet) + t.set_timeout(30) + t.set_account(*args) + + Tp.debug_set_flags(os.getenv('PHOENIX_TEST_DEBUG', '')) + + m = TestManager(t) + try: + loop.run() + finally: + os.kill(p.pid, signal.SIGKILL) + # Sleep 2 seconds so everything can die a nice death + time.sleep(2) + shutil.rmtree(tempdir) -- cgit v1.2.3