1
from landscape.lib.persist import Persist
2
from landscape.reactor import FakeReactor
3
from landscape.monitor.computerinfo import ComputerInfo
4
from landscape.monitor.loadaverage import LoadAverage
5
from landscape.monitor.deployment import MonitorService, MonitorConfiguration
6
from landscape.tests.helpers import (
7
LandscapeTest, LandscapeIsolatedTest, RemoteBrokerHelper)
8
from landscape.broker.tests.test_remote import assertTransmitterActive
9
from landscape.tests.test_plugin import assertReceivesMessages
11
from twisted.internet.defer import Deferred
14
class DeploymentTest(LandscapeTest):
16
def test_get_plugins(self):
17
configuration = MonitorConfiguration()
18
configuration.load(["--monitor-plugins", "ComputerInfo, LoadAverage",
19
"-d", self.makeFile()])
20
monitor = MonitorService(configuration)
21
plugins = monitor.plugins
22
self.assertEquals(len(plugins), 2)
23
self.assertTrue(isinstance(plugins[0], ComputerInfo))
24
self.assertTrue(isinstance(plugins[1], LoadAverage))
26
def test_get_all_plugins(self):
27
configuration = MonitorConfiguration()
28
configuration.load(["--monitor-plugins", "ALL",
29
"-d", self.makeFile()])
30
monitor = MonitorService(configuration)
31
plugins = monitor.plugins
32
self.assertEquals(len(plugins), 12)
35
class MonitorServiceTest(LandscapeIsolatedTest):
37
helpers = [RemoteBrokerHelper]
40
super(MonitorServiceTest, self).setUp()
41
configuration = MonitorConfiguration()
42
configuration.load(["-d", self.broker_service.config.data_path,
44
self.monitor = MonitorService(configuration)
45
self.monitor.reactor = FakeReactor()
46
self.monitor.startService()
49
super(MonitorServiceTest, self).tearDown()
50
self.monitor.stopService()
53
class DeploymentBusTest(MonitorServiceTest):
55
def test_dbus_reactor_transmitter_installed(self):
56
# Set the config to monitor only, because assertTransmitterActive
57
# fires a resynchronize event and the monitor's callback registered
58
# for it would try to get a DBus object published by the manager.
59
self.monitor.registry.config.monitor_only = True
60
return assertTransmitterActive(self, self.broker_service,
63
def test_receives_messages(self):
64
return assertReceivesMessages(self, self.monitor.dbus_service,
65
self.broker_service, self.remote)
67
def test_register_plugin_on_broker_started(self):
69
When the broker is restarted, it fires a "broker-started" signal which
70
makes the Monitor plugin register itself again.
74
def register_plugin(bus_name, object_path):
75
d.callback((bus_name, object_path))
78
self.monitor.remote_broker.register_plugin = register_plugin
79
self.broker_service.dbus_object.broker_started()
82
return self.remote.get_registered_plugins(
84
).addCallback(self.assertEquals,
85
("com.canonical.landscape.Monitor",
86
"/com/canonical/landscape/Monitor"))
88
def test_register_message_on_broker_started(self):
90
When the broker is restarted, it fires a "broker-started" signal which
91
makes the Monitor plugin register all registered messages again.
93
self.monitor.registry.register_message("foo", lambda x: None)
96
def register_client_accepted_message_type(type):
101
self.monitor.remote_broker.register_client_accepted_message_type = \
102
register_client_accepted_message_type
103
self.broker_service.dbus_object.broker_started()
106
return self.remote.get_registered_plugins(
108
).addCallback(self.assertEquals, "foo")
111
class MonitorTest(MonitorServiceTest):
113
def test_monitor_flushes_on_flush_event(self):
114
"""L{MonitorService.flush} saves the persist."""
115
self.monitor.registry.persist.set("a", 1)
116
self.monitor.registry.flush()
119
persist.load(self.monitor.registry.persist_filename)
120
self.assertEquals(persist.get("a"), 1)
122
def test_monitor_flushes_on_flush_interval(self):
124
The monitor is flushed every C{flush_interval} seconds, after
125
the monitor service is started.
127
self.monitor.registry.persist.set("a", 1)
128
self.monitor.reactor.advance(self.monitor.config.flush_interval)
131
persist.load(self.monitor.registry.persist_filename)
132
self.assertEquals(persist.get("a"), 1)
134
def test_monitor_flushes_on_service_stop(self):
135
"""The monitor is flushed when the service stops."""
136
self.monitor.registry.persist.set("a", 1)
137
self.monitor.stopService()
140
persist.load(self.monitor.registry.persist_filename)
141
self.assertEquals(persist.get("a"), 1)
143
def test_monitor_stops_flushing_after_service_stops(self):
144
"""The recurring flush event is cancelled when the service stops."""
145
self.monitor.stopService()
146
self.monitor.registry.persist.set("a", 1)
147
self.monitor.reactor.advance(self.monitor.config.flush_interval)
150
persist.load(self.monitor.registry.persist_filename)
151
self.assertEquals(persist.get("a"), None)
added
removed
landscape/monitor/tests/test_deployment.py
