4
from landscape.monitor.temperature import Temperature
5
from landscape.lib.tests.test_sysstats import ThermalZoneTest
6
from landscape.tests.helpers import MonitorHelper
7
from landscape.tests.mocker import ANY
10
class TemperatureTestWithSampleData(ThermalZoneTest):
11
"""Tests for the temperature plugin."""
13
helpers = [MonitorHelper]
16
"""Initialize test helpers and create a sample thermal zone."""
17
super(TemperatureTestWithSampleData, self).setUp()
18
self.mstore.set_accepted_types(["temperature"])
19
self.write_thermal_zone("ZONE1", "50 C")
21
def test_wb_disabled_with_no_thermal_zones(self):
23
When no thermal zones are available /proc/acpi/thermal_zone
24
will be empty. In this case, the plugin won't register itself
25
to respond to client events such as exchange.
27
thermal_zone_path = tempfile.mkdtemp()
28
os.rmdir(thermal_zone_path)
29
plugin = Temperature(thermal_zone_path=thermal_zone_path)
30
self.assertEquals(plugin._thermal_zones, [])
32
def test_no_messages_without_thermal_zones(self):
34
Messages should never be generated by the plugin when no
35
thermal zones are available.
37
thermal_zone_path = self.make_dir()
38
plugin = Temperature(interval=1, thermal_zone_path=thermal_zone_path)
39
self.monitor.add(plugin)
40
self.reactor.advance(self.monitor.step_size)
41
self.assertEquals(len(self.mstore.get_pending_messages()), 0)
43
def test_disjointed_thermal_zone_temperature_changes(self):
45
Changing data needs to be tracked according to the thermal
46
zone the data is for. This test ensures that the plugin
47
creates messages with changes reported correctly.
49
self.write_thermal_zone("ZONE2", "50 C")
50
plugin = Temperature(thermal_zone_path=self.thermal_zone_path,
51
create_time=self.reactor.time)
52
step_size = self.monitor.step_size
53
self.monitor.add(plugin)
55
self.reactor.advance(step_size)
57
self.write_thermal_zone("ZONE2", "56 C")
58
self.reactor.advance(step_size)
60
messages = list(plugin.create_messages())
61
self.assertEquals(len(messages), 2)
63
self.assertEquals(messages[0]["thermal-zone"], "ZONE1")
64
self.assertEquals(len(messages[0]["temperatures"]), 2)
65
self.assertEquals(messages[0]["temperatures"][0],
67
self.assertEquals(messages[0]["temperatures"][1],
68
(step_size * 2, 50.0))
70
self.assertEquals(messages[1]["thermal-zone"], "ZONE2")
71
self.assertEquals(len(messages[1]["temperatures"]), 2)
72
self.assertEquals(messages[1]["temperatures"][0],
74
self.assertEquals(messages[1]["temperatures"][1],
75
(step_size * 2, 56.0))
77
def test_messaging_flushes(self):
79
Duplicate message should never be created. If no data is
80
available, a message with an empty C{temperatures} list is
83
plugin = Temperature(thermal_zone_path=self.thermal_zone_path,
84
create_time=self.reactor.time)
85
self.monitor.add(plugin)
87
self.reactor.advance(self.monitor.step_size)
89
messages = plugin.create_messages()
90
self.assertEquals(len(messages), 1)
92
messages = plugin.create_messages()
93
self.assertEquals(len(messages), 0)
95
def test_never_exchange_empty_messages(self):
97
The plugin will only create messages when data is available.
98
If no data is available when an exchange occurs no messages
101
self.write_thermal_zone("ZONE2", "50 C")
102
plugin = Temperature(thermal_zone_path=self.thermal_zone_path,
103
create_time=self.reactor.time)
104
step_size = self.monitor.step_size
105
self.monitor.add(plugin)
106
self.assertEquals(len(self.mstore.get_pending_messages()), 0)
108
def test_exchange_messages(self):
110
The temperature plugin queues message when an exchange
111
happens. Each message should be aligned to a step boundary;
112
messages collected bewteen exchange periods should be
113
delivered in a single message.
115
self.write_thermal_zone("ZONE2", "50 C")
116
plugin = Temperature(thermal_zone_path=self.thermal_zone_path,
117
create_time=self.reactor.time)
118
step_size = self.monitor.step_size
119
self.monitor.add(plugin)
120
self.reactor.advance(step_size)
121
self.monitor.exchange()
123
self.assertMessages(self.mstore.get_pending_messages(),
124
[{"type": "temperature",
125
"thermal-zone": "ZONE1",
126
"temperatures": [(step_size, 50.0)]},
127
{"type": "temperature",
128
"thermal-zone": "ZONE2",
129
"temperatures": [(step_size, 50.0)]}])
131
def test_no_messages_on_bad_values(self):
133
If the temperature is in an unknown format, the plugin won't
134
break and no messages are sent.
136
self.write_thermal_zone("ZONE1", "UNKNOWN C")
137
plugin = Temperature(thermal_zone_path=self.thermal_zone_path,
138
create_time=self.reactor.time)
139
step_size = self.monitor.step_size
140
self.monitor.add(plugin)
141
self.reactor.advance(step_size)
142
self.monitor.exchange()
144
self.assertMessages(self.mstore.get_pending_messages(), [])
146
def test_call_on_accepted(self):
147
plugin = Temperature(thermal_zone_path=self.thermal_zone_path,
148
create_time=self.reactor.time)
149
self.monitor.add(plugin)
151
self.reactor.advance(plugin.registry.step_size)
153
remote_broker_mock = self.mocker.replace(self.remote)
154
remote_broker_mock.send_message(ANY, urgent=True)
157
self.reactor.fire(("message-type-acceptance-changed", "temperature"),
160
def test_no_message_if_not_accepted(self):
162
Don't add any messages at all if the broker isn't currently
163
accepting their type.
165
self.mstore.set_accepted_types([])
166
plugin = Temperature(thermal_zone_path=self.thermal_zone_path,
167
create_time=self.reactor.time)
168
self.monitor.add(plugin)
169
self.reactor.advance(self.monitor.step_size * 2)
170
self.monitor.exchange()
172
self.mstore.set_accepted_types(["temperature"])
173
self.assertMessages(list(self.mstore.get_pending_messages()), [])
added
removed
landscape/monitor/tests/test_temperature.py
