3
from landscape.monitor.rebootrequired import RebootRequired
4
from landscape.tests.helpers import (
5
LandscapeTest, MonitorHelper, LogKeeperHelper)
8
class RebootRequiredTest(LandscapeTest):
10
helpers = [MonitorHelper, LogKeeperHelper]
13
super(RebootRequiredTest, self).setUp()
14
self.reboot_required_filename = self.makeFile()
15
self.plugin = RebootRequired(self.reboot_required_filename)
16
self.monitor.add(self.plugin)
17
self.mstore.set_accepted_types(["reboot-required-info"])
19
def test_wb_get_flag(self):
21
L{RebootRequired._get_flag} returns C{True} if the reboot-required
22
flag file is present, C{False} otherwise.
24
self.assertFalse(self.plugin._get_flag())
25
self.makeFile(path=self.reboot_required_filename, content="")
26
self.assertTrue(self.plugin._get_flag())
28
def test_wb_get_packages(self):
30
L{RebootRequired._get_packages} returns the packages listed in the
31
reboot-required packages file if present, or an empty list otherwise.
33
self.assertEqual([], self.plugin._get_packages())
34
self.makeFile(path=self.reboot_required_filename + ".pkgs",
36
self.assertEqual(["bar", "foo"], self.plugin._get_packages())
38
def test_wb_get_packages_with_duplicates(self):
40
The list of packages returned by L{RebootRequired._get_packages} does
41
not contain duplicate values.
43
self.assertEqual([], self.plugin._get_packages())
44
self.makeFile(path=self.reboot_required_filename + ".pkgs",
46
self.assertEqual(["foo"], self.plugin._get_packages())
48
def test_wb_get_packages_with_blank_lines(self):
50
Blank lines are ignored by L{RebootRequired._get_packages}.
52
self.assertEqual([], self.plugin._get_packages())
53
self.makeFile(path=self.reboot_required_filename + ".pkgs",
54
content="bar\n\nfoo\n")
55
self.assertEqual(["bar", "foo"], self.plugin._get_packages())
57
def test_wb_create_message(self):
59
A message should be created if and only if the reboot-required status
60
of the system has changed.
62
self.assertEqual({"flag": False, "packages": []},
63
self.plugin._create_message())
64
self.makeFile(path=self.reboot_required_filename, content="")
65
self.assertEqual({"flag": True},
66
self.plugin._create_message())
67
self.makeFile(path=self.reboot_required_filename + ".pkgs",
69
self.assertEqual({"packages": [u"foo"]},
70
self.plugin._create_message())
72
def test_send_message(self):
74
A new C{"reboot-required-info"} message should be enqueued if and only
75
if the reboot-required status of the system has changed.
77
self.makeFile(path=self.reboot_required_filename + ".pkgs",
79
self.makeFile(path=self.reboot_required_filename, content="")
80
self.plugin.send_message()
81
self.assertIn("Queueing message with updated reboot-required status.",
82
self.logfile.getvalue())
83
self.assertMessages(self.mstore.get_pending_messages(),
84
[{"type": "reboot-required-info",
86
"packages": [u"foo"]}])
87
self.mstore.delete_all_messages()
88
self.plugin.send_message()
89
self.assertMessages(self.mstore.get_pending_messages(), [])
91
def test_run_interval(self):
93
The L{RebootRequired} plugin will be scheduled to run every 15 minutes.
95
self.assertEqual(900, self.plugin.run_interval)
97
def test_run_immediately(self):
99
The L{RebootRequired} plugin will be run immediately at startup.
101
self.assertTrue(True, self.plugin.run_immediately)
105
If the server can accept them, the plugin should send
106
C{reboot-required} messages.
108
with mock.patch.object(self.remote, "send_message"):
110
self.remote.send_message.assert_called_once_with(
111
mock.ANY, mock.ANY, urgent=True)
112
self.mstore.set_accepted_types([])
115
def test_resynchronize(self):
117
The "resynchronize" reactor message cause the plugin to send fresh
121
self.reactor.fire("resynchronize", scopes=["package"])
123
messages = self.mstore.get_pending_messages()
124
self.assertEqual(len(messages), 2)
added
removed
landscape/monitor/tests/test_rebootrequired.py
