← Back to branch summary

~free.ekanayaka/landscape-client/karmic-updates-1.4.4-0ubuntu0.9.10

« back to all changes in this revision

Viewing changes to landscape/sysinfo/tests/test_deployment.py

  • Committer: Bazaar Package Importer
  • Author(s): Rick Clark
  • Date: 2008-09-08 16:35:57 UTC
  • mfrom: (1.1.1 upstream)
  • Revision ID: james.westby@ubuntu.com-20080908163557-l3ixzj5dxz37wnw2
Tags: 1.0.18-0ubuntu1
New upstream release 

Show diffs side-by-side

added added

removed removed

Lines of Context:
landscape/sysinfo/tests/test_deployment.py
 
1
import os
 
2
 
 
3
from logging.handlers import RotatingFileHandler
 
4
from logging import getLogger
 
5
 
 
6
from twisted.internet.defer import Deferred
 
7
 
 
8
from landscape.sysinfo.deployment import (
 
9
    SysInfoConfiguration, ALL_PLUGINS, run, setup_logging)
 
10
from landscape.sysinfo.testplugin import TestPlugin
 
11
from landscape.sysinfo.sysinfo import SysInfoPluginRegistry
 
12
from landscape.sysinfo.load import Load
 
13
 
 
14
from landscape.tests.helpers import LandscapeTest, StandardIOHelper
 
15
from landscape.tests.mocker import ARGS, KWARGS
 
16
 
 
17
 
 
18
class DeploymentTest(LandscapeTest):
 
19
    def setUp(self):
 
20
        super(DeploymentTest, self).setUp()
 
21
        self.configuration = SysInfoConfiguration()
 
22
        self.configuration.default_config_filenames = []
 
23
 
 
24
    def test_get_plugins(self):
 
25
        self.configuration.load(["--sysinfo-plugins", "Load,TestPlugin",
 
26
                                 "-d", self.make_path()])
 
27
        plugins = self.configuration.get_plugins()
 
28
        self.assertEquals(len(plugins), 2)
 
29
        self.assertTrue(isinstance(plugins[0], Load))
 
30
        self.assertTrue(isinstance(plugins[1], TestPlugin))
 
31
 
 
32
    def test_get_all_plugins(self):
 
33
        self.configuration.load(["-d", self.make_path()])
 
34
        plugins = self.configuration.get_plugins()
 
35
        self.assertEquals(len(plugins), len(ALL_PLUGINS))
 
36
 
 
37
    def test_exclude_plugins(self):
 
38
        exclude = ",".join(x for x in ALL_PLUGINS if x != "Load")
 
39
        self.configuration.load(["--exclude-sysinfo-plugins", exclude,
 
40
                                 "-d", self.make_path()])
 
41
        plugins = self.configuration.get_plugins()
 
42
        self.assertEquals(len(plugins), 1)
 
43
        self.assertTrue(isinstance(plugins[0], Load))
 
44
 
 
45
    def test_config_file(self):
 
46
        filename = self.make_path()
 
47
        f = open(filename, "w")
 
48
        f.write("[sysinfo]\nsysinfo_plugins = TestPlugin\n")
 
49
        f.close()
 
50
        self.configuration.load(["--config", filename, "-d", self.make_path()])
 
51
        plugins = self.configuration.get_plugins()
 
52
        self.assertEquals(len(plugins), 1)
 
53
        self.assertTrue(isinstance(plugins[0], TestPlugin))
 
54
 
 
55
 
 
56
class FakeReactor(object):
 
57
    """
 
58
    Something that's easier to understand and more reusable than a bunch of
 
59
    mocker
 
60
    """
 
61
    def __init__(self):
 
62
        self.queued_calls = []
 
63
        self.scheduled_calls = []
 
64
        self.running = False
 
65
    def callWhenRunning(self, callable):
 
66
        self.queued_calls.append(callable)
 
67
    def run(self):
 
68
        self.running = True
 
69
    def callLater(self, seconds, callable, *args, **kwargs):
 
70
        self.scheduled_calls.append((seconds, callable, args, kwargs))
 
71
    def stop(self):
 
72
        self.running = False
 
73
 
 
74
 
 
75
class RunTest(LandscapeTest):
 
76
 
 
77
    helpers = [StandardIOHelper]
 
78
 
 
79
    def tearDown(self):
 
80
        super(RunTest, self).tearDown()
 
81
        logger = getLogger("landscape-sysinfo")
 
82
        for handler in logger.handlers[:]:
 
83
            logger.removeHandler(handler)
 
84
 
 
85
    def test_registry_runs_plugin_and_gets_correct_information(self):
 
86
        run(["--sysinfo-plugins", "TestPlugin"])
 
87
 
 
88
        from landscape.sysinfo.testplugin import current_instance
 
89
 
 
90
        self.assertEquals(current_instance.has_run, True)
 
91
        sysinfo = current_instance.sysinfo
 
92
        self.assertEquals(sysinfo.get_headers(),
 
93
                          [("Test header", "Test value")])
 
94
        self.assertEquals(sysinfo.get_notes(), ["Test note"])
 
95
        self.assertEquals(sysinfo.get_footnotes(), ["Test footnote"])
 
96
 
 
97
    def test_format_sysinfo_gets_correct_information(self):
 
98
        format_sysinfo = self.mocker.replace("landscape.sysinfo.sysinfo."
 
99
                                             "format_sysinfo")
 
100
        format_sysinfo([("Test header", "Test value")],
 
101
                       ["Test note"], ["Test footnote"],
 
102
                       indent="  ")
 
103
        format_sysinfo(ARGS, KWARGS)
 
104
        self.mocker.count(0)
 
105
        self.mocker.replay()
 
106
 
 
107
        run(["--sysinfo-plugins", "TestPlugin"])
 
108
 
 
109
    def test_format_sysinfo_output_is_printed(self):
 
110
        format_sysinfo = self.mocker.replace("landscape.sysinfo.sysinfo."
 
111
                                             "format_sysinfo")
 
112
        format_sysinfo(ARGS, KWARGS)
 
113
        self.mocker.result("Hello there!")
 
114
        self.mocker.replay()
 
115
 
 
116
        run(["--sysinfo-plugins", "TestPlugin"])
 
117
 
 
118
        self.assertEquals(self.stdout.getvalue(), "Hello there!\n")
 
119
 
 
120
    def test_output_is_only_displayed_once_deferred_fires(self):
 
121
        deferred = Deferred()
 
122
        sysinfo = self.mocker.patch(SysInfoPluginRegistry)
 
123
        sysinfo.run()
 
124
        self.mocker.passthrough()
 
125
        self.mocker.result(deferred)
 
126
        self.mocker.replay()
 
127
 
 
128
        run(["--sysinfo-plugins", "TestPlugin"])
 
129
 
 
130
        self.assertNotIn("Test note", self.stdout.getvalue())
 
131
        deferred.callback(None)
 
132
        self.assertIn("Test note", self.stdout.getvalue())
 
133
 
 
134
    def test_default_arguments_load_default_plugins(self):
 
135
        result = run([])
 
136
        def check_result(result):
 
137
            self.assertIn("System load", self.stdout.getvalue())
 
138
            self.assertNotIn("Test note", self.stdout.getvalue())
 
139
        return result.addCallback(check_result)
 
140
 
 
141
    def test_plugins_called_after_reactor_starts(self):
 
142
        """
 
143
        Plugins are invoked after the reactor has started, so that they can
 
144
        spawn processes without concern for race conditions.
 
145
        """
 
146
        reactor = FakeReactor()
 
147
        d = run(["--sysinfo-plugins", "TestPlugin"], reactor=reactor)
 
148
        self.assertEquals(self.stdout.getvalue(), "")
 
149
 
 
150
        self.assertTrue(reactor.running)
 
151
        for x in reactor.queued_calls:
 
152
            x()
 
153
 
 
154
        self.assertEquals(
 
155
            self.stdout.getvalue(),
 
156
            "  Test header: Test value\n\n  => Test note\n\n  Test footnote\n")
 
157
        return d
 
158
 
 
159
    def test_stop_scheduled_in_callback(self):
 
160
        """
 
161
        Because of tm:3011, reactor.stop() must be called in a scheduled call.
 
162
        """
 
163
        reactor = FakeReactor()
 
164
        d = run(["--sysinfo-plugins", "TestPlugin"], reactor=reactor)
 
165
        for x in reactor.queued_calls:
 
166
            x()
 
167
        self.assertEquals(reactor.scheduled_calls, [(0, reactor.stop, (), {})])
 
168
        return d
 
169
 
 
170
    def test_stop_reactor_even_when_sync_exception_from_sysinfo_run(self):
 
171
        """
 
172
        Even when there's a synchronous exception from run_sysinfo, the reactor
 
173
        should be stopped.
 
174
        """
 
175
        self.log_helper.ignore_errors(ZeroDivisionError)
 
176
        reactor = FakeReactor()
 
177
        sysinfo = SysInfoPluginRegistry()
 
178
        sysinfo.run = lambda: 1 / 0
 
179
        d = run(["--sysinfo-plugins", "TestPlugin"], reactor=reactor,
 
180
                sysinfo=sysinfo)
 
181
 
 
182
        for x in reactor.queued_calls:
 
183
            x()
 
184
 
 
185
        self.assertEquals(reactor.scheduled_calls, [(0, reactor.stop, (), {})])
 
186
        return self.assertFailure(d, ZeroDivisionError)
 
187
 
 
188
    def test_wb_logging_setup(self):
 
189
        """
 
190
        setup_logging sets up a "landscape-sysinfo" logger which rotates every
 
191
        week and does not propagate logs to higher-level handlers.
 
192
        """
 
193
        # This hecka whiteboxes but there aren't any underscores!
 
194
        logger = getLogger("landscape-sysinfo")
 
195
        self.assertEquals(logger.handlers, [])
 
196
        setup_logging()
 
197
        logger = getLogger("landscape-sysinfo")
 
198
        self.assertEquals(len(logger.handlers), 1)
 
199
        handler = logger.handlers[0]
 
200
        self.assertTrue(isinstance(handler, RotatingFileHandler))
 
201
        self.assertEquals(handler.maxBytes, 500*1024)
 
202
        self.assertEquals(handler.backupCount, 1)
 
203
        self.assertFalse(logger.propagate)
 
204
 
 
205
    def test_create_log_dir(self):
 
206
        log_dir = self.make_path()
 
207
        self.assertFalse(os.path.exists(log_dir))
 
208
        setup_logging(landscape_dir=log_dir)
 
209
        self.assertTrue(os.path.exists(log_dir))
 
210
        
 
211
 
 
212
    def test_run_sets_up_logging(self):
 
213
        setup_logging_mock = self.mocker.replace(
 
214
            "landscape.sysinfo.deployment.setup_logging")
 
215
        setup_logging_mock()
 
216
        self.mocker.replay()
 
217
 
 
218
        run(["--sysinfo-plugins", "TestPlugin"])