1
from cStringIO import StringIO
2
from logging import getLogger, StreamHandler
4
from twisted.internet.defer import Deferred, succeed, fail
6
from landscape.sysinfo.sysinfo import SysInfoPluginRegistry, format_sysinfo
7
from landscape.plugin import PluginRegistry
8
from landscape.tests.helpers import LandscapeTest
11
class SysInfoPluginRegistryTest(LandscapeTest):
14
super(SysInfoPluginRegistryTest, self).setUp()
15
self.sysinfo = SysInfoPluginRegistry()
16
self.sysinfo_logfile = StringIO()
17
self.handler = StreamHandler(self.sysinfo_logfile)
18
self.logger = getLogger("landscape-sysinfo")
19
self.logger.addHandler(self.handler)
22
super(SysInfoPluginRegistryTest, self).tearDown()
23
self.logger.removeHandler(self.handler)
25
def test_is_plugin_registry(self):
26
self.assertTrue(isinstance(self.sysinfo, PluginRegistry))
28
def test_add_and_get_headers(self):
29
self.sysinfo.add_header("Memory usage", "65%")
30
self.sysinfo.add_header("Swap usage", "None")
32
self.sysinfo.get_headers(),
33
[("Memory usage", "65%"), ("Swap usage", "None")])
34
self.assertEquals(self.sysinfo.get_notes(), [])
35
self.assertEquals(self.sysinfo.get_footnotes(), [])
37
def test_add_same_header_twice(self):
38
self.sysinfo.add_header("Header1", "Value1")
39
self.sysinfo.add_header("Header2", "Value2")
40
self.sysinfo.add_header("Header3", "Value3")
41
self.sysinfo.add_header("Header2", "Value4")
42
self.assertEquals(self.sysinfo.get_headers(),
43
[("Header1", "Value1"),
44
("Header2", "Value4"),
45
("Header3", "Value3")])
47
def test_add_header_with_none_value(self):
48
self.sysinfo.add_header("Header1", "Value1")
49
self.sysinfo.add_header("Header2", None)
50
self.sysinfo.add_header("Header3", "Value3")
51
self.assertEquals(self.sysinfo.get_headers(),
52
[("Header1", "Value1"),
53
("Header3", "Value3")])
54
self.sysinfo.add_header("Header2", "Value2")
55
self.assertEquals(self.sysinfo.get_headers(),
56
[("Header1", "Value1"),
57
("Header2", "Value2"),
58
("Header3", "Value3")])
60
def test_add_and_get_notes(self):
61
self.sysinfo.add_note("Your laptop is burning!")
62
self.sysinfo.add_note("Oh, your house too, btw.")
64
self.sysinfo.get_notes(),
65
["Your laptop is burning!", "Oh, your house too, btw."])
66
self.assertEquals(self.sysinfo.get_headers(), [])
67
self.assertEquals(self.sysinfo.get_footnotes(), [])
69
def test_add_and_get_footnotes(self):
70
self.sysinfo.add_footnote("Graphs available at http://graph")
71
self.sysinfo.add_footnote("Go! Go!")
73
self.sysinfo.get_footnotes(),
74
["Graphs available at http://graph", "Go! Go!"])
75
self.assertEquals(self.sysinfo.get_headers(), [])
76
self.assertEquals(self.sysinfo.get_notes(), [])
80
def __init__(self, deferred):
81
self._deferred = deferred
82
def register(self, registry):
87
plugin_deferred1 = Deferred()
88
plugin_deferred2 = Deferred()
90
plugin1 = Plugin(plugin_deferred1)
91
plugin2 = Plugin(plugin_deferred2)
93
self.sysinfo.add(plugin1)
94
self.sysinfo.add(plugin2)
96
def check_result(result):
97
self.assertEquals(result, [123, 456])
99
deferred = self.sysinfo.run()
100
deferred.addBoth(check_result)
102
self.assertEquals(deferred.called, False)
103
plugin_deferred1.callback(123)
104
self.assertEquals(deferred.called, False)
105
plugin_deferred2.callback(456)
106
self.assertEquals(deferred.called, True)
108
plugin_exception_message = (
109
"There were exceptions while processing one or more plugins. "
110
"See ~/.landscape/sysinfo.log for more information.")
112
def test_plugins_run_after_synchronous_error(self):
114
Even when a plugin raises a synchronous error, other plugins will
117
self.log_helper.ignore_errors(ZeroDivisionError)
118
plugins_what_run = []
119
class BadPlugin(object):
120
def register(self, registry):
123
plugins_what_run.append(self)
125
class GoodPlugin(object):
126
def register(self, registry):
129
plugins_what_run.append(self)
131
plugin1 = BadPlugin()
132
plugin2 = GoodPlugin()
133
self.sysinfo.add(plugin1)
134
self.sysinfo.add(plugin2)
136
self.assertEquals(plugins_what_run, [plugin1, plugin2])
137
log = self.sysinfo_logfile.getvalue()
138
message = "BadPlugin plugin raised an exception."
139
self.assertIn(message, log)
140
self.assertIn("1/0", log)
141
self.assertIn("ZeroDivisionError", log)
143
self.sysinfo.get_notes(),
144
[self.plugin_exception_message])
146
def test_asynchronous_errors_logged(self):
147
self.log_helper.ignore_errors(ZeroDivisionError)
148
class BadPlugin(object):
149
def register(self, registry):
152
return fail(ZeroDivisionError("yay"))
154
self.sysinfo.add(plugin)
156
log = self.sysinfo_logfile.getvalue()
157
message = "BadPlugin plugin raised an exception."
158
self.assertIn(message, log)
159
self.assertIn("ZeroDivisionError: yay", log)
161
self.sysinfo.get_notes(),
162
[self.plugin_exception_message])
164
def test_multiple_exceptions_get_one_note(self):
165
self.log_helper.ignore_errors(ZeroDivisionError)
166
class RegularBadPlugin(object):
167
def register(self, registry):
172
class AsyncBadPlugin(object):
173
def register(self, registry):
176
return fail(ZeroDivisionError("Hi"))
178
plugin1 = RegularBadPlugin()
179
plugin2 = AsyncBadPlugin()
180
self.sysinfo.add(plugin1)
181
self.sysinfo.add(plugin2)
185
self.sysinfo.get_notes(),
186
[self.plugin_exception_message])
190
class FormatTest(LandscapeTest):
192
def test_no_headers(self):
193
output = format_sysinfo([])
194
self.assertEquals(output, "")
196
def test_one_header(self):
197
output = format_sysinfo([("Header", "Value")])
198
self.assertEquals(output, "Header: Value")
200
def test_parallel_headers_with_just_enough_space(self):
201
output = format_sysinfo([("Header1", "Value1"),
202
("Header2", "Value2")], width=34)
203
self.assertEquals(output, "Header1: Value1 Header2: Value2")
205
def test_stacked_headers_which_barely_doesnt_fit(self):
206
output = format_sysinfo([("Header1", "Value1"),
207
("Header2", "Value2")], width=33)
208
self.assertEquals(output, "Header1: Value1\nHeader2: Value2")
210
def test_stacked_headers_with_clearly_insufficient_space(self):
211
output = format_sysinfo([("Header1", "Value1"),
212
("Header2", "Value2")], width=1)
213
self.assertEquals(output, "Header1: Value1\n"
216
def test_indent_headers_in_parallel_with_just_enough_space(self):
217
output = format_sysinfo([("Header1", "Value1"),
218
("Header2", "Value2")], indent=">>", width=36)
219
self.assertEquals(output, ">>Header1: Value1 Header2: Value2")
221
def test_indent_headers_stacked_which_barely_doesnt_fit(self):
222
output = format_sysinfo([("Header1", "Value1"),
223
("Header2", "Value2")], indent=">>", width=35)
224
self.assertEquals(output, ">>Header1: Value1\n"
227
def test_parallel_and_stacked_headers(self):
228
headers = [("Header%d" % i, "Value%d" % i) for i in range(1, 6)]
229
output = format_sysinfo(headers)
230
self.assertEquals(output,
231
"Header1: Value1 Header3: Value3 Header5: Value5\n"
232
"Header2: Value2 Header4: Value4")
234
def test_value_alignment(self):
235
output = format_sysinfo([("Header one", "Value one"),
236
("Header2", "Value2"),
237
("Header3", "Value3"),
238
("Header4", "Value4"),
239
("Header5", "Value five")], width=45)
240
# These headers and values were crafted to cover several cases:
242
# - Header padding (Header2 and Header3)
243
# - Value padding (Value2)
244
# - Lack of value padding due to a missing last column (Value3)
245
# - Lack of value padding due to being a last column (Value4)
247
self.assertEquals(output,
248
"Header one: Value one Header4: Value4\n"
249
"Header2: Value2 Header5: Value five\n"
252
def test_one_note(self):
253
self.assertEquals(format_sysinfo(notes=["Something's wrong!"]),
254
"=> Something's wrong!")
256
def test_more_notes(self):
257
self.assertEquals(format_sysinfo(notes=["Something's wrong",
258
"You should look at it",
260
"=> Something's wrong\n"
261
"=> You should look at it\n"
264
def test_indented_notes(self):
265
self.assertEquals(format_sysinfo(notes=["Something's wrong",
266
"You should look at it",
267
"Really"], indent=">>"),
268
">>=> Something's wrong\n"
269
">>=> You should look at it\n"
272
def test_header_and_note(self):
273
self.assertEquals(format_sysinfo(headers=[("Header", "Value")],
279
def test_one_footnote(self):
281
self.assertEquals(format_sysinfo(footnotes=["Graphs at http://..."]),
282
"Graphs at http://...")
284
def test_more_footnotes(self):
286
self.assertEquals(format_sysinfo(footnotes=["Graphs at http://...",
288
"Graphs at http://...\n"
291
def test_indented_footnotes(self):
292
# Barely more interesting.
293
self.assertEquals(format_sysinfo(footnotes=["Graphs at http://...",
296
">>Graphs at http://...\n"
299
def test_header_and_footnote(self):
301
self.assertEquals(format_sysinfo(headers=[("Header", "Value")],
302
footnotes=["Footnote"]),
308
def test_header_note_and_footnote(self):
310
self.assertEquals(format_sysinfo(headers=[("Header", "Value")],
312
footnotes=["Footnote"]),
320
def test_indented_headers_notes_and_footnotes(self):
322
self.assertEquals(format_sysinfo(headers=[("Header1", "Value1"),
323
("Header2", "Value2"),
324
("Header3", "Value3")],
325
notes=["Note1", "Note2"],
326
footnotes=["Footnote1", "Footnote2"],
329
" Header1: Value1 Header3: Value3\n"
339
def test_wrap_long_notes(self):
341
format_sysinfo(notes=[
342
"I do believe that a very long note, such as one that is "
343
"longer than about 50 characters, should wrap at the specified "
344
"width."], width=50, indent="Z"),
346
Z=> I do believe that a very long note, such as
347
one that is longer than about 50 characters,
348
should wrap at the specified width.""")
b'\\ No newline at end of file'
added
removed
landscape/sysinfo/tests/test_sysinfo.py
