← Back to branch summary

~jderose/ubuntu/precise/dbus-python/oneiric-backport

« back to all changes in this revision

Viewing changes to .pc/python3-support.patch/test/test-signals.py

  • Committer: Package Import Robot
  • Author(s): Barry Warsaw
  • Date: 2012-01-12 14:47:33 UTC
  • Revision ID: package-import@ubuntu.com-20120112144733-xtfbmgw30h0j40d2
Tags: 0.84.0-2ubuntu1
https://launchpad.net/bugs/893091
* debian/patches:
  - since-0.84.0.patch: Upstream unreleased changes from git tag
    dbus-python-0.84.0 to HEAD.  This is a precursor to the following.
  - python3-support.patch: Upstream unreleased changes from git
    `python3` branch for supporting Python 3. (LP: #893091)
* debian/rules: Enable the test suite.

Show diffs side-by-side

added added

removed removed

Lines of Context:
.pc/python3-support.patch/test/test-signals.py
 
1
#!/usr/bin/env python
 
2
 
 
3
# Copyright (C) 2004 Red Hat Inc. <http://www.redhat.com/>
 
4
# Copyright (C) 2005-2007 Collabora Ltd. <http://www.collabora.co.uk/>
 
5
#
 
6
# Permission is hereby granted, free of charge, to any person
 
7
# obtaining a copy of this software and associated documentation
 
8
# files (the "Software"), to deal in the Software without
 
9
# restriction, including without limitation the rights to use, copy,
 
10
# modify, merge, publish, distribute, sublicense, and/or sell copies
 
11
# of the Software, and to permit persons to whom the Software is
 
12
# furnished to do so, subject to the following conditions:
 
13
#
 
14
# The above copyright notice and this permission notice shall be
 
15
# included in all copies or substantial portions of the Software.
 
16
#
 
17
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
 
18
# EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
 
19
# MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND
 
20
# NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT
 
21
# HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY,
 
22
# WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
 
23
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
 
24
# DEALINGS IN THE SOFTWARE.
 
25
 
 
26
import sys
 
27
import os
 
28
import unittest
 
29
import time
 
30
import logging
 
31
 
 
32
builddir = os.path.normpath(os.environ["DBUS_TOP_BUILDDIR"])
 
33
pydir = os.path.normpath(os.environ["DBUS_TOP_SRCDIR"])
 
34
 
 
35
import dbus
 
36
import _dbus_bindings
 
37
import gobject
 
38
import dbus.glib
 
39
import dbus.service
 
40
 
 
41
 
 
42
logging.basicConfig()
 
43
logging.getLogger().setLevel(1)
 
44
logger = logging.getLogger('test-signals')
 
45
 
 
46
 
 
47
pkg = dbus.__file__
 
48
if not pkg.startswith(pydir):
 
49
    raise Exception("DBus modules (%s) are not being picked up from the package"%pkg)
 
50
 
 
51
if not _dbus_bindings.__file__.startswith(builddir):
 
52
    raise Exception("DBus modules (%s) are not being picked up from the package"%_dbus_bindings.__file__)
 
53
 
 
54
 
 
55
NAME = "org.freedesktop.DBus.TestSuitePythonService"
 
56
IFACE = "org.freedesktop.DBus.TestSuiteInterface"
 
57
OBJECT = "/org/freedesktop/DBus/TestSuitePythonObject"
 
58
 
 
59
 
 
60
class TestSignals(unittest.TestCase):
 
61
    def setUp(self):
 
62
        logger.info('setUp()')
 
63
        self.bus = dbus.SessionBus()
 
64
        self.remote_object = self.bus.get_object(NAME, OBJECT)
 
65
        self.remote_object_fallback_trivial = self.bus.get_object(NAME,
 
66
                OBJECT + '/Fallback')
 
67
        self.remote_object_fallback = self.bus.get_object(NAME,
 
68
                OBJECT + '/Fallback/Badger')
 
69
        self.remote_object_follow = self.bus.get_object(NAME, OBJECT,
 
70
                follow_name_owner_changes=True)
 
71
        self.iface = dbus.Interface(self.remote_object, IFACE)
 
72
        self.iface_follow = dbus.Interface(self.remote_object_follow, IFACE)
 
73
        self.fallback_iface = dbus.Interface(self.remote_object_fallback, IFACE)
 
74
        self.fallback_trivial_iface = dbus.Interface(
 
75
                self.remote_object_fallback_trivial, IFACE)
 
76
        self.in_test = None
 
77
 
 
78
    def signal_test_impl(self, iface, name, test_removal=False):
 
79
        self.in_test = name
 
80
        # using append rather than assignment here to avoid scoping issues
 
81
        result = []
 
82
 
 
83
        def _timeout_handler():
 
84
            logger.debug('_timeout_handler for %s: current state %s', name, self.in_test)
 
85
            if self.in_test == name:
 
86
                main_loop.quit()
 
87
        def _signal_handler(s, sender, path):
 
88
            logger.debug('_signal_handler for %s: current state %s', name, self.in_test)
 
89
            if self.in_test not in (name, name + '+removed'):
 
90
                return
 
91
            logger.info('Received signal from %s:%s, argument is %r',
 
92
                        sender, path, s)
 
93
            result.append('received')
 
94
            main_loop.quit()
 
95
        def _rm_timeout_handler():
 
96
            logger.debug('_timeout_handler for %s: current state %s', name, self.in_test)
 
97
            if self.in_test == name + '+removed':
 
98
                main_loop.quit()
 
99
 
 
100
        logger.info('Testing %s', name)
 
101
        match = iface.connect_to_signal('SignalOneString', _signal_handler,
 
102
                                        sender_keyword='sender',
 
103
                                        path_keyword='path')
 
104
        logger.info('Waiting for signal...')
 
105
        iface.EmitSignal('SignalOneString', 0)
 
106
        source_id = gobject.timeout_add(1000, _timeout_handler)
 
107
        main_loop.run()
 
108
        if not result:
 
109
            raise AssertionError('Signal did not arrive within 1 second')
 
110
        logger.debug('Removing match')
 
111
        match.remove()
 
112
        gobject.source_remove(source_id)
 
113
        if test_removal:
 
114
            self.in_test = name + '+removed'
 
115
            logger.info('Testing %s', name)
 
116
            result = []
 
117
            iface.EmitSignal('SignalOneString', 0)
 
118
            source_id = gobject.timeout_add(1000, _rm_timeout_handler)
 
119
            main_loop.run()
 
120
            if result:
 
121
                raise AssertionError('Signal should not have arrived, but did')
 
122
            gobject.source_remove(source_id)
 
123
 
 
124
    def testFallback(self):
 
125
        self.signal_test_impl(self.fallback_iface, 'Fallback')
 
126
 
 
127
    def testFallbackTrivial(self):
 
128
        self.signal_test_impl(self.fallback_trivial_iface, 'FallbackTrivial')
 
129
 
 
130
    def testSignal(self):
 
131
        self.signal_test_impl(self.iface, 'Signal')
 
132
 
 
133
    def testRemoval(self):
 
134
        self.signal_test_impl(self.iface, 'Removal', True)
 
135
 
 
136
    def testSignalAgain(self):
 
137
        self.signal_test_impl(self.iface, 'SignalAgain')
 
138
 
 
139
    def testRemovalAgain(self):
 
140
        self.signal_test_impl(self.iface, 'RemovalAgain', True)
 
141
 
 
142
    def testSignalF(self):
 
143
        self.signal_test_impl(self.iface_follow, 'Signal')
 
144
 
 
145
    def testRemovalF(self):
 
146
        self.signal_test_impl(self.iface_follow, 'Removal', True)
 
147
 
 
148
    def testSignalAgainF(self):
 
149
        self.signal_test_impl(self.iface_follow, 'SignalAgain')
 
150
 
 
151
    def testRemovalAgainF(self):
 
152
        self.signal_test_impl(self.iface_follow, 'RemovalAgain', True)
 
153
 
 
154
if __name__ == '__main__':
 
155
    main_loop = gobject.MainLoop()
 
156
    gobject.threads_init()
 
157
    dbus.glib.init_threads()
 
158
 
 
159
    logger.info('Starting unit test')
 
160
    unittest.main()