← Back to branch summary

~ubuntuone-control-tower/ubuntu-sso-client/trunk

« back to all changes in this revision

Viewing changes to contrib/testing/testcase.py

  • Committer: Tarmac
  • Author(s): Natalia B. Bidart
  • Date: 2010-12-20 15:49:02 UTC
  • mfrom: (659.2.6 use-devtools)
  • Revision ID: tarmac-20101220154902-ofu0qcb3xucvg4ry
* Use ubuntuone-dev-tools to run the tests and the lint checker (LP: #686606).

Show diffs side-by-side

added added

removed removed

Lines of Context:
contrib/testing/testcase.py
1
 
#
2
 
# Author: Guillermo Gonzalez <guillermo.gonzalez@canonical.com>
3
 
# Author: Natalia B. Bidart <natalia.bidart@canonical.com>
4
 
#
5
 
# Copyright 2009-2010 Canonical Ltd.
6
 
#
7
 
# This program is free software: you can redistribute it and/or modify it
8
 
# under the terms of the GNU General Public License version 3, as published
9
 
# by the Free Software Foundation.
10
 
#
11
 
# This program is distributed in the hope that it will be useful, but
12
 
# WITHOUT ANY WARRANTY; without even the implied warranties of
13
 
# MERCHANTABILITY, SATISFACTORY QUALITY, or FITNESS FOR A PARTICULAR
14
 
# PURPOSE.  See the GNU General Public License for more details.
15
 
#
16
 
# You should have received a copy of the GNU General Public License along
17
 
# with this program.  If not, see <http://www.gnu.org/licenses/>.
18
 
"""Base tests cases and test utilities."""
19
 
 
20
 
import logging
21
 
 
22
 
import dbus
23
 
 
24
 
from dbus.mainloop.glib import DBusGMainLoop
25
 
from twisted.internet import defer
26
 
from twisted.python import failure
27
 
from twisted.trial.unittest import TestCase
28
 
 
29
 
 
30
 
class DBusTestCase(TestCase):
31
 
    """Test the DBus event handling."""
32
 
 
33
 
    def setUp(self):
34
 
        """Setup the infrastructure fo the test (dbus service)."""
35
 
        self.loop = DBusGMainLoop(set_as_default=True)
36
 
        self.bus = dbus.bus.BusConnection(mainloop=self.loop)
37
 
        # monkeypatch busName.__del__ to avoid errors on gc
38
 
        # we take care of releasing the name in shutdown
39
 
        dbus.service.BusName.__del__ = lambda _: None
40
 
        self.bus.set_exit_on_disconnect(False)
41
 
        self.signal_receivers = set()
42
 
 
43
 
    def tearDown(self):
44
 
        """Cleanup the test."""
45
 
        d = self.cleanup_signal_receivers(self.signal_receivers)
46
 
        d.addBoth(self._tear_down)
47
 
        return d
48
 
 
49
 
    def _tear_down(self, _):
50
 
        """Shutdown."""
51
 
        self.bus.flush()
52
 
        self.bus.close()
53
 
 
54
 
    def error_handler(self, error):
55
 
        """Default error handler for DBus calls."""
56
 
        if isinstance(error, failure.Failure):
57
 
            self.fail(error.getErrorMessage())
58
 
 
59
 
    def cleanup_signal_receivers(self, signal_receivers):
60
 
        """Cleanup self.signal_receivers and returns a deferred."""
61
 
        deferreds = []
62
 
        for match in signal_receivers:
63
 
            d = defer.Deferred()
64
 
 
65
 
            def callback(*args):
66
 
                """Callback that accepts *args."""
67
 
                if not d.called:
68
 
                    d.callback(args)
69
 
 
70
 
            self.bus.call_async(dbus.bus.BUS_DAEMON_NAME,
71
 
                                dbus.bus.BUS_DAEMON_PATH,
72
 
                                dbus.bus.BUS_DAEMON_IFACE, 'RemoveMatch', 's',
73
 
                                (str(match),), callback, self.error_handler)
74
 
            deferreds.append(d)
75
 
        if deferreds:
76
 
            return defer.DeferredList(deferreds)
77
 
        else:
78
 
            return defer.succeed(True)
79
 
 
80
 
 
81
 
class MementoHandler(logging.Handler):
82
 
    """A handler class which store logging records in a list."""
83
 
 
84
 
    def __init__(self, *args, **kwargs):
85
 
        """Create the instance, and add a records attribute."""
86
 
        logging.Handler.__init__(self, *args, **kwargs)
87
 
        self.records = []
88
 
 
89
 
    def emit(self, record):
90
 
        """Just add the record to self.records."""
91
 
        self.records.append(record)
92
 
 
93
 
    def check(self, level, *msgs):
94
 
        """Verifies that the msgs are logged in the specified level."""
95
 
        for rec in self.records:
96
 
            if rec.levelno == level and all(m in rec.message for m in msgs):
97
 
                return True
98
 
        return False
99
 
 
100
 
    def check_debug(self, *msgs):
101
 
        """Shortcut for checking in DEBUG."""
102
 
        return self.check(logging.DEBUG, *msgs)
103
 
 
104
 
    def check_info(self, *msgs):
105
 
        """Shortcut for checking in INFO."""
106
 
        return self.check(logging.INFO, *msgs)
107
 
 
108
 
    def check_warning(self, *msgs):
109
 
        """Shortcut for checking in WARNING."""
110
 
        return self.check(logging.WARNING, *msgs)
111
 
 
112
 
    def check_error(self, *msgs):
113
 
        """Shortcut for checking in ERROR."""
114
 
        return self.check(logging.ERROR, *msgs)
115
 
 
116
 
    def check_exception(self, exception_class, *msgs):
117
 
        """Shortcut for checking exceptions."""
118
 
        for rec in self.records:
119
 
            if rec.levelno == logging.ERROR and \
120
 
               all(m in rec.exc_text for m in msgs) and \
121
 
               exception_class == rec.exc_info[0]:
122
 
                return True
123
 
        return False