← Back to branch summary

~free.ekanayaka/landscape-client/lucid-1.5.4-0ubuntu0.10.04.0

« back to all changes in this revision

Viewing changes to landscape/tests/test_deployment.py

  • Committer: Bazaar Package Importer
  • Author(s): Free Ekanayaka
  • Date: 2010-06-28 18:07:18 UTC
  • mfrom: (1.2.5 upstream)
  • Revision ID: james.westby@ubuntu.com-20100628180718-vytyqgbtkiirv5sb
Tags: 1.5.2.1-0ubuntu0.10.04.0
https://launchpad.net/bugs/597000
Filter duplicate network interfaces in get_active_interfaces (LP: #597000)

Show diffs side-by-side

added added

removed removed

Lines of Context:
landscape/tests/test_deployment.py
1
1
import sys
2
2
import os
3
3
from optparse import OptionParser
4
 
import logging
5
 
import signal
6
 
 
7
 
from landscape.lib.dbus_util import Object
8
 
from landscape.deployment import (
9
 
    LandscapeService, Configuration, get_versioned_persist,
10
 
    assert_unowned_bus_name, run_landscape_service)
11
 
from landscape.tests.helpers import (
12
 
    LandscapeTest, LandscapeIsolatedTest, DBusHelper)
 
4
 
 
5
from landscape.deployment import Configuration, get_versioned_persist
 
6
 
 
7
from landscape.tests.helpers import LandscapeTest
13
8
from landscape.tests.mocker import ANY
14
9
 
15
10
 
16
11
class BabbleConfiguration(Configuration):
17
12
    config_section = "babble"
18
13
    default_config_filenames = []
 
14
 
19
15
    def make_parser(self):
20
16
        parser = super(BabbleConfiguration, self).make_parser()
21
17
        parser.add_option("--whatever", metavar="STUFF")
30
26
 
31
27
    def reset_config(self, configuration_class=None):
32
28
        if not configuration_class:
 
29
 
33
30
            class MyConfiguration(Configuration):
34
31
                default_config_filenames = []
35
32
            configuration_class = MyConfiguration
 
33
 
36
34
        self.config_class = configuration_class
37
35
        self.config = configuration_class()
38
36
        self.parser = self.config.make_parser()
56
54
        self.assertEquals(self.config.log_level, "command line")
57
55
 
58
56
    def test_command_line_option_without_default(self):
 
57
 
59
58
        class MyConfiguration(Configuration):
 
59
 
60
60
            def make_parser(self):
61
61
                parser = OptionParser()
62
62
                # Keep the dash in the option name to ensure it works.
63
63
                parser.add_option("--foo-bar")
64
64
                return parser
 
65
 
65
66
        self.assertEquals(MyConfiguration().foo_bar, None)
66
67
 
67
68
    def test_command_line_with_required_options(self):
 
69
 
68
70
        class MyConfiguration(Configuration):
69
71
            required_options = ("foo_bar",)
70
72
            config = None
 
73
 
71
74
            def make_parser(self):
72
75
                parser = super(MyConfiguration, self).make_parser()
73
76
                # Keep the dash in the option name to ensure it works.
86
89
        self.assertEquals(self.config.foo_bar, "ooga")
87
90
 
88
91
    def test_command_line_with_unsaved_options(self):
 
92
 
89
93
        class MyConfiguration(Configuration):
90
94
            unsaved_options = ("foo_bar",)
91
95
            config = None
 
96
 
92
97
            def make_parser(self):
93
98
                parser = super(MyConfiguration, self).make_parser()
94
99
                # Keep the dash in the option name to ensure it works.
95
100
                parser.add_option("--foo-bar", metavar="NAME")
96
101
                return parser
 
102
 
97
103
        self.reset_config(configuration_class=MyConfiguration)
98
104
        self.write_config_file()
99
105
 
117
123
 
118
124
    def test_no_section_available(self):
119
125
        config_filename = self.makeFile("")
 
126
 
120
127
        class MyConfiguration(Configuration):
121
128
            config_section = "nonexistent"
122
129
            default_config_filenames = (config_filename,)
 
130
 
123
131
        self.reset_config(configuration_class=MyConfiguration)
124
132
        self.config.load([])
125
133
 
225
233
        data = open(filename).read()
226
234
        self.assertEquals(data.strip(), "[client]\nlog_level = error")
227
235
 
228
 
    def test_bus_option(self):
229
 
        """The bus option must be specified as 'system' or 'session'."""
230
 
        self.assertRaises(SystemExit,
231
 
                          self.config.load,
232
 
                          ["--bus", "foobar"])
233
 
        self.config.load(["--bus", "session"])
234
 
        self.assertEquals(self.config.bus, "session")
235
 
        self.config.load(["--bus", "system"])
236
 
        self.assertEquals(self.config.bus, "system")
237
 
 
238
236
    def test_config_option(self):
239
237
        opts = self.parser.parse_args(["--config", "hello.cfg"])[0]
240
238
        self.assertEquals(opts.config, "hello.cfg")
245
243
        self.assertEquals(self.config.hello, "world")
246
244
 
247
245
    def test_load_typed_option_from_file(self):
 
246
 
248
247
        class MyConfiguration(self.config_class):
 
248
 
249
249
            def make_parser(self):
250
250
                parser = super(MyConfiguration, self).make_parser()
251
251
                parser.add_option("--year", default=1, type="int")
252
252
                return parser
 
253
 
253
254
        filename = self.makeFile("[client]\nyear = 2008\n")
254
255
        config = MyConfiguration()
255
256
        config.load(["--config", filename])
256
257
        self.assertEquals(config.year, 2008)
257
258
 
258
259
    def test_load_typed_option_from_command_line(self):
 
260
 
259
261
        class MyConfiguration(self.config_class):
 
262
 
260
263
            def make_parser(self):
261
264
                parser = super(MyConfiguration, self).make_parser()
262
265
                parser.add_option("--year", default=1, type="int")
263
266
                return parser
 
267
 
264
268
        config = MyConfiguration()
265
269
        config.load(["--year", "2008"])
266
270
        self.assertEquals(config.year, 2008)
320
324
        os.chmod(default_filename1, 0)
321
325
        self.assertEquals(self.config.get_config_filename(),
322
326
                          default_filename2)
323
 
        
 
327
 
324
328
        # If is is readable, than return the first default configuration file.
325
329
        os.chmod(default_filename1, 0644)
326
330
        self.assertEquals(self.config.get_config_filename(),
339
343
        self.assertEquals(self.config.get_config_filename(),
340
344
                          explicit_filename)
341
345
 
 
346
    def test_sockets_path(self):
 
347
        """
 
348
        The L{Configuration.socket_path} property returns the path to the
 
349
        socket directory.
 
350
        """
 
351
        self.assertEquals(self.config.sockets_path,
 
352
                          "/var/lib/landscape/client/sockets")
342
353
 
343
354
 
344
355
class GetVersionedPersistTest(LandscapeTest):
345
356
 
346
357
    def test_upgrade_service(self):
 
358
 
347
359
        class FakeService(object):
348
360
            persist_filename = self.makePersistFile(content="")
349
361
            service_name = "monitor"
359
371
 
360
372
        persist = get_versioned_persist(FakeService())
361
373
        self.assertEquals(stash[0], persist)
362
 
 
363
 
 
364
 
class LandscapeServiceTest(LandscapeTest):
365
 
 
366
 
    def setUp(self):
367
 
        super(LandscapeServiceTest, self).setUp()
368
 
        signal.signal(signal.SIGUSR1, signal.SIG_DFL)
369
 
 
370
 
    def tearDown(self):
371
 
        super(LandscapeServiceTest, self).tearDown()
372
 
        signal.signal(signal.SIGUSR1, signal.SIG_DFL)
373
 
 
374
 
    def test_create_persist(self):
375
 
        class FakeService(LandscapeService):
376
 
            persist_filename = self.makePersistFile(content="")
377
 
            service_name = "monitor"
378
 
        service = FakeService(None)
379
 
        self.assertEquals(service.persist.filename, service.persist_filename)
380
 
 
381
 
    def test_no_persist_without_filename(self):
382
 
        class FakeService(LandscapeService):
383
 
            service_name = "monitor"
384
 
        service = FakeService(None)
385
 
        self.assertFalse(hasattr(service, "persist"))
386
 
 
387
 
    def test_usr1_rotates_logs(self):
388
 
        """
389
 
        SIGUSR1 should cause logs to be reopened.
390
 
        """
391
 
        logging.getLogger().addHandler(logging.FileHandler(self.makeFile()))
392
 
        # Store the initial set of handlers
393
 
        original_streams = [handler.stream for handler in
394
 
                            logging.getLogger().handlers if
395
 
                            isinstance(handler, logging.FileHandler)]
396
 
 
397
 
        # Instantiating LandscapeService should register the handler
398
 
        LandscapeService(None)
399
 
        # We'll call it directly
400
 
        handler = signal.getsignal(signal.SIGUSR1)
401
 
        self.assertTrue(handler)
402
 
        handler(None, None)
403
 
        new_streams = [handler.stream for handler in
404
 
                       logging.getLogger().handlers if
405
 
                       isinstance(handler, logging.FileHandler)]
406
 
 
407
 
        for stream in new_streams:
408
 
            self.assertTrue(stream not in original_streams)
409
 
 
410
 
    def test_ignore_sigusr1(self):
411
 
        """
412
 
        SIGUSR1 is ignored if we so request.
413
 
        """
414
 
        class Configuration:
415
 
            ignore_sigusr1 = True
416
 
 
417
 
        # Instantiating LandscapeService should not register the
418
 
        # handler if we request to ignore it.
419
 
        config = Configuration()
420
 
        LandscapeService(config)
421
 
 
422
 
        handler = signal.getsignal(signal.SIGUSR1)
423
 
        self.assertFalse(handler)
424
 
 
425
 
 
426
 
class AssertUnownedBusNameTest(LandscapeIsolatedTest):
427
 
 
428
 
    helpers = [DBusHelper]
429
 
 
430
 
    class BoringService(Object):
431
 
        bus_name = "com.example.BoringService"
432
 
        object_path = "/com/example/BoringService"
433
 
 
434
 
    def test_raises_sysexit_when_owned(self):
435
 
        service = self.BoringService(self.bus)
436
 
        self.assertRaises(SystemExit, assert_unowned_bus_name,
437
 
                          self.bus, self.BoringService.bus_name)
438
 
 
439
 
    def test_do_nothing_when_unowned(self):
440
 
        assert_unowned_bus_name(self.bus, self.BoringService.bus_name)
441
 
 
442
 
 
443
 
class RunLandscapeServiceTests(LandscapeTest):
444
 
    def test_wrong_user(self):
445
 
        getuid_mock = self.mocker.replace("os.getuid")
446
 
        reactor_install_mock = self.mocker.replace("landscape.reactor.install")
447
 
        reactor_install_mock()
448
 
        getuid_mock()
449
 
        self.mocker.result(1)
450
 
        self.mocker.replay()
451
 
 
452
 
        class MyService(LandscapeService):
453
 
            service_name = "broker"
454
 
 
455
 
        sys_exit = self.assertRaises(
456
 
            SystemExit, run_landscape_service, Configuration,
457
 
            MyService, [], "whatever")
458
 
        self.assertIn("landscape-broker must be run as landscape",
459
 
                      str(sys_exit))