← Back to branch summary

~ahasenack/landscape-client/landscape-client-1.5.5-0ubuntu0.9.04.0

« back to all changes in this revision

Viewing changes to landscape/monitor/tests/test_computerinfo.py

  • Committer: Bazaar Package Importer
  • Author(s): Rick Clark
  • Date: 2008-09-08 16:35:57 UTC
  • mfrom: (1.1.1 upstream)
  • Revision ID: james.westby@ubuntu.com-20080908163557-l3ixzj5dxz37wnw2
Tags: 1.0.18-0ubuntu1
New upstream release 

Show diffs side-by-side

added added

removed removed

Lines of Context:
landscape/monitor/tests/test_computerinfo.py
 
1
import re
 
2
 
 
3
from landscape.monitor.computerinfo import ComputerInfo
 
4
from landscape.tests.helpers import LandscapeTest, MakePathHelper, MonitorHelper
 
5
from landscape.tests.mocker import ANY
 
6
 
 
7
 
 
8
def get_hostname():
 
9
    return "ooga"
 
10
 
 
11
 
 
12
class ComputerInfoTest(LandscapeTest):
 
13
 
 
14
    helpers = [MonitorHelper, MakePathHelper]
 
15
 
 
16
    sample_memory_info = """
 
17
MemTotal:      1547072 kB
 
18
MemFree:        106616 kB
 
19
Buffers:        267088 kB
 
20
Cached:         798388 kB
 
21
SwapCached:          0 kB
 
22
Active:         728952 kB
 
23
Inactive:       536512 kB
 
24
HighTotal:      646016 kB
 
25
HighFree:        42204 kB
 
26
LowTotal:       901056 kB
 
27
LowFree:         64412 kB
 
28
SwapTotal:     1622524 kB
 
29
SwapFree:      1622524 kB
 
30
Dirty:              24 kB
 
31
Writeback:           0 kB
 
32
Mapped:         268756 kB
 
33
Slab:           105492 kB
 
34
CommitLimit:   2396060 kB
 
35
Committed_AS:  1166936 kB
 
36
PageTables:       2748 kB
 
37
VmallocTotal:   114680 kB
 
38
VmallocUsed:      6912 kB
 
39
VmallocChunk:   107432 kB
 
40
"""
 
41
 
 
42
    def setUp(self):
 
43
        LandscapeTest.setUp(self)
 
44
        self.lsb_release_filename = self.make_path("""\
 
45
DISTRIB_ID=Ubuntu
 
46
DISTRIB_RELEASE=6.06
 
47
DISTRIB_CODENAME=dapper
 
48
DISTRIB_DESCRIPTION="Ubuntu 6.06.1 LTS"
 
49
""")
 
50
 
 
51
    def test_get_hostname(self):
 
52
        self.mstore.set_accepted_types(["computer-info"])
 
53
        plugin = ComputerInfo(get_hostname=get_hostname,
 
54
                              lsb_release_filename=self.lsb_release_filename)
 
55
        self.monitor.add(plugin)
 
56
        plugin.exchange()
 
57
        messages = self.mstore.get_pending_messages()
 
58
        self.assertEquals(len(messages), 1)
 
59
        self.assertEquals(messages[0]["type"], "computer-info")
 
60
        self.assertEquals(messages[0]["hostname"], "ooga")
 
61
 
 
62
    def test_get_real_hostname(self):
 
63
        self.mstore.set_accepted_types(["computer-info"])
 
64
        plugin = ComputerInfo()
 
65
        self.monitor.add(plugin)
 
66
        plugin.exchange()
 
67
        messages = self.mstore.get_pending_messages()
 
68
        self.assertEquals(len(messages), 1)
 
69
        self.assertEquals(messages[0]["type"], "computer-info")
 
70
        self.assertNotEquals(len(messages[0]["hostname"]), 0)
 
71
        self.assertTrue(re.search("\w", messages[0]["hostname"]))
 
72
 
 
73
    def test_only_report_changed_hostnames(self):
 
74
        self.mstore.set_accepted_types(["computer-info"])
 
75
        plugin = ComputerInfo(get_hostname=get_hostname)
 
76
        self.monitor.add(plugin)
 
77
        plugin.exchange()
 
78
        messages = self.mstore.get_pending_messages()
 
79
        self.assertEquals(len(messages), 1)
 
80
        plugin.exchange()
 
81
        messages = self.mstore.get_pending_messages()
 
82
        self.assertEquals(len(messages), 1)
 
83
 
 
84
    def test_report_changed_hostnames(self):
 
85
        def hostname_factory(hostnames=["ooga", "wubble", "wubble"]):
 
86
            i = 0
 
87
            while i < len(hostnames):
 
88
                yield hostnames[i]
 
89
                i = i + 1
 
90
 
 
91
        self.mstore.set_accepted_types(["computer-info"])
 
92
        plugin = ComputerInfo(get_hostname=hostname_factory().next)
 
93
        self.monitor.add(plugin)
 
94
 
 
95
        plugin.exchange()
 
96
        messages = self.mstore.get_pending_messages()
 
97
        self.assertEquals(len(messages), 1)
 
98
        self.assertEquals(messages[0]["hostname"], "ooga")
 
99
 
 
100
        plugin.exchange()
 
101
        messages = self.mstore.get_pending_messages()
 
102
        self.assertEquals(len(messages), 2)
 
103
        self.assertEquals(messages[1]["hostname"], "wubble")
 
104
 
 
105
    def test_get_total_memory(self):
 
106
        self.mstore.set_accepted_types(["computer-info"])
 
107
        meminfo_filename = self.make_path(self.sample_memory_info)
 
108
        plugin = ComputerInfo(meminfo_file=meminfo_filename)
 
109
        self.monitor.add(plugin)
 
110
        plugin.exchange()
 
111
        messages = self.mstore.get_pending_messages()
 
112
        message = messages[0]
 
113
        self.assertEquals(message["type"], "computer-info")
 
114
        self.assertEquals(message["total-memory"], 1510)
 
115
        self.assertEquals(message["total-swap"], 1584)
 
116
 
 
117
    def test_get_real_total_memory(self):
 
118
        self.mstore.set_accepted_types(["computer-info"])
 
119
        meminfo_filename = self.make_path(self.sample_memory_info)
 
120
        plugin = ComputerInfo()
 
121
        self.monitor.add(plugin)
 
122
        plugin.exchange()
 
123
        message = self.mstore.get_pending_messages()[0]
 
124
        self.assertEquals(message["type"], "computer-info")
 
125
        self.assertTrue(isinstance(message["total-memory"], int))
 
126
        self.assertTrue(isinstance(message["total-swap"], int))
 
127
 
 
128
    def test_wb_report_changed_total_memory(self):
 
129
        self.mstore.set_accepted_types(["computer-info"])
 
130
        plugin = ComputerInfo()
 
131
        self.monitor.add(plugin)
 
132
 
 
133
        plugin._get_memory_info = lambda: (1510, 1584)
 
134
        plugin.exchange()
 
135
        message = self.mstore.get_pending_messages()[0]
 
136
        self.assertEquals(message["total-memory"], 1510)
 
137
        self.assertTrue("total-swap" in message)
 
138
 
 
139
        plugin._get_memory_info = lambda: (2048, 1584)
 
140
        plugin.exchange()
 
141
        message = self.mstore.get_pending_messages()[1]
 
142
        self.assertEquals(message["total-memory"], 2048)
 
143
        self.assertTrue("total-swap" not in message)
 
144
 
 
145
    def test_wb_report_changed_total_swap(self):
 
146
        self.mstore.set_accepted_types(["computer-info"])
 
147
        plugin = ComputerInfo()
 
148
        self.monitor.add(plugin)
 
149
 
 
150
        plugin._get_memory_info = lambda: (1510, 1584)
 
151
        plugin.exchange()
 
152
        message = self.mstore.get_pending_messages()[0]
 
153
        self.assertEquals(message["total-swap"], 1584)
 
154
        self.assertTrue("total-memory" in message)
 
155
 
 
156
        plugin._get_memory_info = lambda: (1510, 2048)
 
157
        plugin.exchange()
 
158
        message = self.mstore.get_pending_messages()[1]
 
159
        self.assertEquals(message["total-swap"], 2048)
 
160
        self.assertTrue("total-memory" not in message)
 
161
 
 
162
    def test_get_distribution(self):
 
163
        """
 
164
        Various details about the distribution should be reported by
 
165
        the plugin.  This test ensures that the right kinds of details
 
166
        end up in messages produced by the plugin.
 
167
        """
 
168
        self.mstore.set_accepted_types(["distribution-info"])
 
169
        plugin = ComputerInfo()
 
170
        self.monitor.add(plugin)
 
171
 
 
172
        plugin.exchange()
 
173
        message = self.mstore.get_pending_messages()[0]
 
174
        self.assertEquals(message["type"], "distribution-info")
 
175
        self.assertTrue("distributor-id" in message)
 
176
        self.assertTrue("description" in message)
 
177
        self.assertTrue("release" in message)
 
178
        self.assertTrue("code-name" in message)
 
179
 
 
180
    def test_get_sample_distribution(self):
 
181
        """
 
182
        Sample data is used to ensure that expected values end up in
 
183
        the distribution data reported by the plugin.
 
184
        """
 
185
        self.mstore.set_accepted_types(["distribution-info"])
 
186
        plugin = ComputerInfo(lsb_release_filename=self.lsb_release_filename)
 
187
        self.monitor.add(plugin)
 
188
 
 
189
        plugin.exchange()
 
190
        message = self.mstore.get_pending_messages()[0]
 
191
        self.assertEquals(message["type"], "distribution-info")
 
192
        self.assertEquals(message["distributor-id"], "Ubuntu")
 
193
        self.assertEquals(message["description"], "Ubuntu 6.06.1 LTS")
 
194
        self.assertEquals(message["release"], "6.06")
 
195
        self.assertEquals(message["code-name"], "dapper")
 
196
 
 
197
    def test_report_once(self):
 
198
        """
 
199
        Distribution data shouldn't be reported unless it's changed
 
200
        since the last time it was reported.
 
201
        """
 
202
        self.mstore.set_accepted_types(["distribution-info"])
 
203
        plugin = ComputerInfo()
 
204
        self.monitor.add(plugin)
 
205
 
 
206
        plugin.exchange()
 
207
        messages = self.mstore.get_pending_messages()
 
208
        self.assertEquals(len(messages), 1)
 
209
        self.assertTrue(messages[0]["type"], "distribution-info")
 
210
 
 
211
        plugin.exchange()
 
212
        messages = self.mstore.get_pending_messages()
 
213
        self.assertEquals(len(messages), 1)
 
214
 
 
215
    def test_wb_report_changed_distribution(self):
 
216
        """
 
217
        When distribution data changes, the new data should be sent to
 
218
        the server.
 
219
        """
 
220
        self.mstore.set_accepted_types(["distribution-info"])
 
221
        plugin = ComputerInfo(lsb_release_filename=self.lsb_release_filename)
 
222
        self.monitor.add(plugin)
 
223
 
 
224
        plugin.exchange()
 
225
        message = self.mstore.get_pending_messages()[0]
 
226
        self.assertEquals(message["type"], "distribution-info")
 
227
        self.assertEquals(message["distributor-id"], "Ubuntu")
 
228
        self.assertEquals(message["description"], "Ubuntu 6.06.1 LTS")
 
229
        self.assertEquals(message["release"], "6.06")
 
230
        self.assertEquals(message["code-name"], "dapper")
 
231
 
 
232
        plugin._lsb_release_filename = self.make_path("""\
 
233
DISTRIB_ID=Ubuntu
 
234
DISTRIB_RELEASE=6.10
 
235
DISTRIB_CODENAME=edgy
 
236
DISTRIB_DESCRIPTION="Ubuntu 6.10"
 
237
""")
 
238
        plugin.exchange()
 
239
        message = self.mstore.get_pending_messages()[1]
 
240
        self.assertEquals(message["type"], "distribution-info")
 
241
        self.assertEquals(message["distributor-id"], "Ubuntu")
 
242
        self.assertEquals(message["description"], "Ubuntu 6.10")
 
243
        self.assertEquals(message["release"], "6.10")
 
244
        self.assertEquals(message["code-name"], "edgy")
 
245
 
 
246
    def test_unknown_distribution_key(self):
 
247
        self.mstore.set_accepted_types(["distribution-info"])
 
248
        lsb_release_filename = self.make_path("""\
 
249
DISTRIB_ID=Ubuntu
 
250
DISTRIB_RELEASE=6.10
 
251
DISTRIB_CODENAME=edgy
 
252
DISTRIB_DESCRIPTION="Ubuntu 6.10"
 
253
DISTRIB_NEW_UNEXPECTED_KEY=ooga
 
254
""")
 
255
        plugin = ComputerInfo(lsb_release_filename=lsb_release_filename)
 
256
        self.monitor.add(plugin)
 
257
 
 
258
        plugin.exchange()
 
259
        message = self.mstore.get_pending_messages()[0]
 
260
        self.assertEquals(message["type"], "distribution-info")
 
261
        self.assertEquals(message["distributor-id"], "Ubuntu")
 
262
        self.assertEquals(message["description"], "Ubuntu 6.10")
 
263
        self.assertEquals(message["release"], "6.10")
 
264
        self.assertEquals(message["code-name"], "edgy")
 
265
 
 
266
    def test_resynchronize(self):
 
267
        """
 
268
        If a reactor event "resynchronize" is received, messages for
 
269
        all computer info should be generated.
 
270
        """
 
271
        self.mstore.set_accepted_types(["distribution-info", "computer-info"])
 
272
        meminfo_filename = self.make_path(self.sample_memory_info)
 
273
        plugin = ComputerInfo(get_hostname=get_hostname,
 
274
                              meminfo_file=meminfo_filename,
 
275
                              lsb_release_filename=self.lsb_release_filename)
 
276
        self.monitor.add(plugin)
 
277
        plugin.exchange()
 
278
        self.reactor.fire("resynchronize")
 
279
        plugin.exchange()
 
280
        computer_info = {"type": "computer-info", "hostname": "ooga",
 
281
                         "timestamp": 0, "total-memory": 1510,
 
282
                         "total-swap": 1584}
 
283
        dist_info = {"type": "distribution-info",
 
284
                     "code-name": "dapper", "description": "Ubuntu 6.06.1 LTS",
 
285
                     "distributor-id": "Ubuntu", "release": "6.06"}
 
286
        self.assertMessages(self.mstore.get_pending_messages(),
 
287
                            [computer_info, dist_info,
 
288
                             computer_info, dist_info])
 
289
 
 
290
    def test_computer_info_call_on_accepted(self):
 
291
        plugin = ComputerInfo()
 
292
        self.monitor.add(plugin)
 
293
 
 
294
        remote_broker_mock = self.mocker.replace(self.remote)
 
295
        remote_broker_mock.send_message(ANY, urgent=True)
 
296
        self.mocker.replay()
 
297
 
 
298
        self.mstore.set_accepted_types(["computer-info"])
 
299
        self.reactor.fire(("message-type-acceptance-changed", "computer-info"),
 
300
                          True)
 
301
 
 
302
    def test_distribution_info_call_on_accepted(self):
 
303
        plugin = ComputerInfo()
 
304
        self.monitor.add(plugin)
 
305
 
 
306
        remote_broker_mock = self.mocker.replace(self.remote)
 
307
        remote_broker_mock.send_message(ANY, urgent=True)
 
308
        self.mocker.replay()
 
309
 
 
310
        self.mstore.set_accepted_types(["distribution-info"])
 
311
        self.reactor.fire(("message-type-acceptance-changed",
 
312
                           "distribution-info"),
 
313
                          True)
 
314
 
 
315
    def test_message_if_not_accepted(self):
 
316
        """
 
317
        Don't add any messages at all if the broker isn't currently
 
318
        accepting their type.
 
319
        """
 
320
        plugin = ComputerInfo()
 
321
        self.monitor.add(plugin)
 
322
        self.reactor.advance(self.monitor.step_size * 2)
 
323
        self.monitor.exchange()
 
324
 
 
325
        self.mstore.set_accepted_types(["distribution-info", "computer-info"])
 
326
        self.assertMessages(list(self.mstore.get_pending_messages()), [])