← Back to branch summary

~ubuntu-branches/ubuntu/lucid/landscape-client/lucid-updates-201411191716

« back to all changes in this revision

Viewing changes to landscape/tests/helpers.py

  • Committer: Bazaar Package Importer
  • Author(s): Free Ekanayaka
  • Date: 2009-12-16 10:50:05 UTC
  • mto: This revision was merged to the branch mainline in revision 25.
  • Revision ID: james.westby@ubuntu.com-20091216105005-9lgmwl5zklvvp6e2
Import upstream version 1.4.0

Show diffs side-by-side

added added

removed removed

Lines of Context:
landscape/tests/helpers.py
6
6
import os
7
7
import tempfile
8
8
import sys
 
9
import unittest
9
10
 
10
11
import dbus
11
12
 
37
38
    "set-intervals", "unknown-id"]
38
39
 
39
40
 
40
 
class LandscapeTest(MockerTestCase, TestCase):
 
41
class HelperTestCase(unittest.TestCase):
41
42
 
42
43
    helpers = []
43
44
 
44
45
    def setUp(self):
45
 
        super(LandscapeTest, self).setUp()
46
 
 
47
 
        self._old_config_filenames = BaseConfiguration.default_config_filenames
48
 
        BaseConfiguration.default_config_filenames = []
49
 
        # make_path-related stuff
50
 
        self.dirname = tempfile.mkdtemp()
51
 
        self.counter = 0
52
 
 
53
46
        self._helper_instances = []
54
47
        if LogKeeperHelper not in self.helpers:
55
48
            self.helpers.insert(0, LogKeeperHelper)
59
52
            self._helper_instances.append(helper)
60
53
 
61
54
    def tearDown(self):
62
 
        BaseConfiguration.default_config_filenames = self._old_config_filenames
63
55
        for helper in reversed(self._helper_instances):
64
56
            helper.tear_down(self)
65
 
        shutil.rmtree(self.dirname)
66
 
        super(LandscapeTest, self).tearDown()
 
57
 
 
58
 
 
59
class MessageTestCase(unittest.TestCase):
67
60
 
68
61
    def assertMessage(self, obtained, expected):
69
62
        obtained = obtained.copy()
93
86
            raise self.failureException("Got %d more messages than expected:\n"
94
87
                                        "%s" % (diff, extra))
95
88
 
 
89
 
 
90
class LandscapeTest(MessageTestCase, MockerTestCase,
 
91
                    HelperTestCase, TestCase):
 
92
 
 
93
    def setUp(self):
 
94
        self._old_config_filenames = BaseConfiguration.default_config_filenames
 
95
        BaseConfiguration.default_config_filenames = []
 
96
        MockerTestCase.setUp(self)
 
97
        HelperTestCase.setUp(self)
 
98
        TestCase.setUp(self)
 
99
 
 
100
    def tearDown(self):
 
101
        BaseConfiguration.default_config_filenames = self._old_config_filenames
 
102
        TestCase.tearDown(self)
 
103
        HelperTestCase.tearDown(self)
 
104
        MockerTestCase.tearDown(self)
 
105
 
96
106
    def assertDeferredSucceeded(self, deferred):
97
107
        self.assertTrue(isinstance(deferred, Deferred))
98
108
        called = []
101
111
        deferred.addCallback(callback)
102
112
        self.assertTrue(called)
103
113
 
104
 
    def make_dir(self):
105
 
        path = self.make_path()
106
 
        os.mkdir(path)
107
 
        return path
108
 
 
109
 
    def make_path(self, content=None, path=None):
110
 
        if path is None:
111
 
            self.counter += 1
112
 
            path = "%s/%03d" % (self.dirname, self.counter)
113
 
        if content is not None:
114
 
            file = open(path, "w")
 
114
    def assertFileContent(self, filename, expected_content):
 
115
        fd = open(filename)
 
116
        actual_content = fd.read()
 
117
        fd.close()
 
118
        self.assertEquals(expected_content, actual_content)
 
119
 
 
120
    def makePersistFile(self, *args, **kwargs):
 
121
        """Return a temporary filename to be used by a L{Persist} object.
 
122
 
 
123
        The possible .old persist file is cleaned up after the test.
 
124
 
 
125
        @see: L{MockerTestCase.makeFile}
 
126
        """
 
127
        persist_filename = self.makeFile(*args, **kwargs)
 
128
 
 
129
        def remove_saved_persist():
115
130
            try:
116
 
                file.write(content)
117
 
            finally:
118
 
                file.close()
119
 
        return path
 
131
                os.remove(persist_filename + ".old")
 
132
            except OSError:
 
133
                pass
 
134
        self.addCleanup(remove_saved_persist)
 
135
        return persist_filename
120
136
 
121
137
 
122
138
class LandscapeIsolatedTest(LandscapeTest):
123
139
    """TestCase that also runs all test methods in a subprocess."""
124
140
 
125
141
    def run(self, result):
 
142
        if not getattr(LandscapeTest, "_cleanup_patch", False):
 
143
            run_method = LandscapeTest.run
 
144
            def run_wrapper(oself, *args, **kwargs):
 
145
                try:
 
146
                    return run_method(oself, *args, **kwargs)
 
147
                finally:
 
148
                    MockerTestCase._MockerTestCase__cleanup(oself)
 
149
            LandscapeTest.run = run_wrapper
 
150
            LandscapeTest._cleanup_patch = True
126
151
        run_isolated(LandscapeTest, self, result)
127
152
 
128
153
 
210
235
            self.ignored_exception_types.append(type_or_regex)
211
236
 
212
237
 
213
 
class MakePathHelper(object):
214
 
 
215
 
    def set_up(self, test_case):
216
 
        pass
217
 
 
218
 
    def tear_down(self, test_case):
219
 
        pass
220
 
 
221
 
 
222
238
class EnvironSnapshot(object):
223
239
 
224
240
    def __init__(self):
249
265
      - data_path: The data path that the broker will use.
250
266
    """
251
267
 
 
268
    reactor_factory = FakeReactor
 
269
    transport_factory = FakeTransport
 
270
    needs_bpickle_dbus = True
 
271
 
252
272
    def set_up(self, test_case):
253
 
 
254
 
        bpickle_dbus.install()
255
 
 
256
 
        test_case.config_filename = test_case.make_path(
 
273
        if self.needs_bpickle_dbus:
 
274
            bpickle_dbus.install()
 
275
 
 
276
        test_case.config_filename = test_case.makeFile(
257
277
            "[client]\n"
258
278
            "url = http://localhost:91919\n"
259
279
            "computer_title = Default Computer Title\n"
260
280
            "account_name = default_account_name\n"
261
281
            "ping_url = http://localhost:91910/\n")
262
282
 
263
 
        test_case.data_path = test_case.make_dir()
264
 
        test_case.log_dir = test_case.make_dir()
 
283
        test_case.data_path = test_case.makeDir()
 
284
        test_case.log_dir = test_case.makeDir()
265
285
 
266
286
        bootstrap_list.bootstrap(data_path=test_case.data_path,
267
287
                                 log_dir=test_case.log_dir)
270
290
            default_config_filenames = [test_case.config_filename]
271
291
 
272
292
        config = MyBrokerConfiguration()
273
 
        config.load(["--bus", "session", "--data-path", test_case.data_path])
 
293
        config.load(["--bus", "session",
 
294
                     "--data-path", test_case.data_path,
 
295
                     "--ignore-sigusr1"])
274
296
 
275
297
        class FakeBrokerService(BrokerService):
276
298
            """A broker which uses a fake reactor and fake transport."""
277
 
            reactor_factory = FakeReactor
278
 
            transport_factory = FakeTransport
 
299
            reactor_factory = self.reactor_factory
 
300
            transport_factory = self.transport_factory
279
301
 
280
302
        test_case.broker_service = service = FakeBrokerService(config)
281
303
        test_case.remote = FakeRemoteBroker(service.exchanger,
282
304
                                            service.message_store)
283
305
 
284
306
    def tear_down(self, test_case):
285
 
        bpickle_dbus.uninstall()
 
307
        if self.needs_bpickle_dbus:
 
308
            bpickle_dbus.uninstall()
286
309
 
287
310
 
288
311
class RemoteBrokerHelper(FakeRemoteBrokerHelper):
345
368
    def set_up(self, test_case):
346
369
        super(MonitorHelper, self).set_up(test_case)
347
370
        persist = Persist()
348
 
        persist_filename = test_case.make_path()
 
371
        persist_filename = test_case.makePersistFile()
349
372
        test_case.monitor = MonitorPluginRegistry(
350
373
            test_case.remote, test_case.broker_service.reactor,
351
374
            test_case.broker_service.config,
365
388
        class MyManagerConfiguration(ManagerConfiguration):
366
389
            default_config_filenames = [test_case.config_filename]
367
390
        config = MyManagerConfiguration()
 
391
        config.load(["--data-path", test_case.data_path])
368
392
        test_case.manager = ManagerPluginRegistry(
369
393
            test_case.remote, test_case.broker_service.reactor,
370
394
            config)