1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
|
import os
from landscape.monitor.rebootrequired import RebootRequired
from landscape.tests.helpers import LandscapeIsolatedTest
from landscape.tests.helpers import MonitorHelper, LogKeeperHelper
from landscape.tests.mocker import ANY
class RebootRequiredTest(LandscapeIsolatedTest):
helpers = [MonitorHelper, LogKeeperHelper]
def setUp(self):
super(RebootRequiredTest, self).setUp()
self.reboot_required_filename = self.makeFile("")
self.plugin = RebootRequired(self.reboot_required_filename)
self.monitor.add(self.plugin)
self.mstore.set_accepted_types(["reboot-required"])
def test_wb_check_reboot_required(self):
"""
L{RebootRequired.check_reboot_required} should return C{True} if the
reboot-required flag file is present, C{False} otherwise.
"""
self.assertTrue(self.plugin._check_reboot_required())
os.remove(self.reboot_required_filename)
self.assertFalse(self.plugin._check_reboot_required())
def test_wb_create_message(self):
"""
A message should be created if and only if the reboot-required status
of the system has changed.
"""
self.assertEquals(self.plugin._create_message(), {"flag": True})
self.assertEquals(self.plugin._create_message(), {})
def test_send_message(self):
"""
A new C{"reboot-required"} message should be enqueued if and only
if the reboot-required status of the system has changed.
"""
self.plugin.send_message()
self.assertIn("Queueing message with updated reboot-required status.",
self.logfile.getvalue())
self.assertMessages(self.mstore.get_pending_messages(),
[{"type": "reboot-required", "flag": True}])
self.mstore.delete_all_messages()
self.plugin.send_message()
self.assertMessages(self.mstore.get_pending_messages(), [])
def test_run(self):
"""
If the server can accept them, the plugin should send
C{reboot-required} messages.
"""
broker_mock = self.mocker.replace(self.remote)
broker_mock.send_message(ANY, urgent=True)
self.mocker.replay()
self.plugin.run()
self.mstore.set_accepted_types([])
self.plugin.run()
|