3
from twisted.internet.defer import Deferred
5
from landscape.package.changer import find_changer_command
6
from landscape.package.store import PackageStore
8
from landscape.manager.packagemanager import PackageManager
9
from landscape.manager.manager import ManagerPluginRegistry
10
from landscape.tests.helpers import (
11
LandscapeIsolatedTest, RemoteBrokerHelper, EnvironSaverHelper)
14
class PackageManagerTest(LandscapeIsolatedTest):
15
"""Tests for the temperature plugin."""
17
helpers = [RemoteBrokerHelper, EnvironSaverHelper]
20
"""Initialize test helpers and create a sample thermal zone."""
21
LandscapeIsolatedTest.setUp(self)
23
self.manager = ManagerPluginRegistry(self.broker_service.reactor,
25
self.broker_service.config)
27
self.package_store_filename = self.makeFile()
28
self.package_store = PackageStore(self.package_store_filename)
29
self.package_manager = PackageManager(self.package_store_filename)
31
def test_create_default_store_on_registration(self):
32
filename = os.path.join(self.broker_service.config.data_path,
34
package_manager = PackageManager()
36
self.assertFalse(os.path.isfile(filename))
37
self.manager.add(package_manager)
38
self.assertTrue(os.path.isfile(filename))
40
def test_dont_spawn_changer_if_message_not_accepted(self):
41
self.manager.add(self.package_manager)
43
package_manager_mock = self.mocker.patch(self.package_manager)
44
package_manager_mock.spawn_changer()
49
return self.package_manager.run()
51
def test_spawn_changer_on_registration_when_already_accepted(self):
52
package_manager_mock = self.mocker.patch(self.package_manager)
53
package_manager_mock.spawn_changer()
55
# Slightly tricky as we have to wait for the result of run(),
56
# but we don't have its deferred yet. To handle it, we create
57
# our own deferred, and register a callback for when run()
58
# returns, chaining both deferreds at that point.
60
def run_has_run(run_result_deferred):
61
return run_result_deferred.chainDeferred(deferred)
63
package_manager_mock.run()
64
self.mocker.passthrough(run_has_run)
68
service = self.broker_service
69
service.message_store.set_accepted_types(["change-packages-result"])
70
self.manager.add(self.package_manager)
74
def test_spawn_changer_on_run_if_message_accepted(self):
75
self.manager.add(self.package_manager)
77
service = self.broker_service
78
service.message_store.set_accepted_types(["change-packages-result"])
80
package_manager_mock = self.mocker.patch(self.package_manager)
81
package_manager_mock.spawn_changer()
82
self.mocker.count(2) # Once for registration, then again explicitly.
86
return self.package_manager.run()
88
def test_change_packages_handling(self):
89
self.manager.add(self.package_manager)
91
package_manager_mock = self.mocker.patch(self.package_manager)
92
package_manager_mock.spawn_changer()
95
message = {"type": "change-packages"}
96
service = self.broker_service
97
self.manager.dispatch_message(message)
98
task = self.package_store.get_next_task("changer")
100
self.assertEquals(task.data, message)
102
def test_spawn_changer(self):
103
command = self.makeFile("#!/bin/sh\necho 'I am the changer!' >&2\n")
104
os.chmod(command, 0755)
105
find_command_mock = self.mocker.replace(find_changer_command)
107
self.mocker.result(command)
110
self.package_store.add_task("changer", "Do something!")
112
package_manager = PackageManager(self.package_store_filename)
113
self.manager.add(package_manager)
114
result = package_manager.spawn_changer()
116
def got_result(result):
117
log = self.logfile.getvalue()
118
self.assertIn("I am the changer!", log)
119
self.assertNotIn(command, log)
121
return result.addCallback(got_result)
123
def test_spawn_changer_without_output(self):
124
find_command_mock = self.mocker.replace(find_changer_command)
126
self.mocker.result("/bin/true")
129
self.package_store.add_task("changer", "Do something!")
131
package_manager = PackageManager(self.package_store_filename)
132
self.manager.add(package_manager)
133
result = package_manager.spawn_changer()
135
def got_result(result):
136
log = self.logfile.getvalue()
137
self.assertNotIn("changer output", log)
139
return result.addCallback(got_result)
141
def test_spawn_changer_copies_environment(self):
142
command = self.makeFile("#!/bin/sh\necho VAR: $VAR\n")
143
os.chmod(command, 0755)
144
find_command_mock = self.mocker.replace(find_changer_command)
146
self.mocker.result(command)
149
package_manager = PackageManager(self.package_store_filename)
150
self.manager.add(package_manager)
152
self.package_store.add_task("changer", "Do something!")
154
os.environ["VAR"] = "HI!"
156
result = package_manager.spawn_changer()
158
def got_result(result):
159
log = self.logfile.getvalue()
160
self.assertIn("VAR: HI!", log)
161
self.assertNotIn(command, log)
163
return result.addCallback(got_result)
165
def test_spawn_changer_passes_quiet_option(self):
166
command = self.makeFile("#!/bin/sh\necho OPTIONS: $@\n")
167
os.chmod(command, 0755)
168
find_command_mock = self.mocker.replace(find_changer_command)
170
self.mocker.result(command)
173
package_manager = PackageManager(self.package_store_filename)
174
self.manager.add(package_manager)
176
self.package_store.add_task("changer", "Do something!")
178
result = package_manager.spawn_changer()
180
def got_result(result):
181
log = self.logfile.getvalue()
182
self.assertIn("OPTIONS: --quiet", log)
183
self.assertNotIn(command, log)
185
return result.addCallback(got_result)
187
def test_spawn_changer_wont_run_without_tasks(self):
188
command = self.makeFile("#!/bin/sh\necho RUN!\n")
189
os.chmod(command, 0755)
190
find_command_mock = self.mocker.replace(find_changer_command)
192
self.mocker.result(command)
195
package_manager = PackageManager(self.package_store_filename)
196
self.manager.add(package_manager)
198
result = package_manager.spawn_changer()
200
def got_result(result):
201
log = self.logfile.getvalue()
202
self.assertNotIn("RUN!", log)
204
return result.addCallback(got_result)
206
def test_spawn_changer_doesnt_chdir(self):
207
command = self.makeFile("#!/bin/sh\necho RUN\n")
208
os.chmod(command, 0755)
209
dir = self.make_dir()
213
find_command_mock = self.mocker.replace(find_changer_command)
215
self.mocker.result(command)
218
package_manager = PackageManager(self.package_store_filename)
219
self.manager.add(package_manager)
221
self.package_store.add_task("changer", "Do something!")
223
result = package_manager.spawn_changer()
225
def got_result(result):
226
log = self.logfile.getvalue()
227
self.assertIn("RUN", log)
228
# restore permissions to the dir so tearDown can clean it up
231
return result.addCallback(got_result)
added
removed
landscape/manager/tests/test_packagemanager.py
