-
Notifications
You must be signed in to change notification settings - Fork 9
/
Copy pathmetrics_test.py
202 lines (174 loc) · 7.54 KB
/
metrics_test.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
"""Unit test for the metrics library.
This test is primarily intended to ensure the metrics codepaths for V1
and V2 devices are executed in case folks working on the codebase have one
type of unit and not the other.
This test is a little gross; it forcibly injects a dict that looks a lot
like unmarshalled JSON from the device into libdyson's handlers. We then
check for the values on the far-side of the Prometheus metrics, which
ensures things are hooked up right, but limits our ability to test. For
example, we cannot currently test enum values with this test.
"""
import enum
import unittest
from prometheus_client import registry
import libdyson
import metrics
NAME = 'test device'
SERIAL = 'XX1-ZZ-1234ABCD'
CREDENTIALS = 'credz'
class TestMetrics(unittest.TestCase):
def setUp(self):
self.registry = registry.CollectorRegistry(auto_describe=True)
self.metrics = metrics.Metrics(registry=self.registry)
def test_enum_values(self):
test = enum.Enum('testEnum', 'RED GREEN BLUE')
self.assertEqual(metrics.enum_values(test), [1, 2, 3])
def test_update_v1_environmental(self):
device = libdyson.DysonPureCoolLink(
SERIAL, CREDENTIALS, libdyson.DEVICE_TYPE_PURE_COOL_LINK)
payload = {
'msg': 'ENVIRONMENTAL-CURRENT-SENSOR-DATA',
'time': '2021-03-17T15:09:23.000Z',
'data': {'tact': '2956', 'hact': '0047', 'pact': '0005', 'vact': 'INIT', 'sltm': 'OFF'}
}
device._handle_message(payload)
labels = {'name': NAME, 'serial': SERIAL}
self.metrics.update(NAME, device, is_state=False,
is_environmental=True)
cases = {
'dyson_temperature_celsius': 22.6,
'dyson_volatile_organic_compounds_units': 0,
'dyson_dust_units': 5
}
for metric, want in cases.items():
got = self.registry.get_sample_value(metric, labels)
self.assertEqual(got, want, f'metric {metric}')
def test_update_v2_environmental(self):
device = libdyson.DysonPureCool(
SERIAL, CREDENTIALS, libdyson.DEVICE_TYPE_PURE_COOL)
payload = {
'msg': 'ENVIRONMENTAL-CURRENT-SENSOR-DATA',
'time': '2021-03-17T15:09:23.000Z',
'data': {'tact': '2956', 'hact': '0047', 'pm10': '3', 'pm25': 'INIT',
'noxl': 30, 'va10': 'INIT', 'sltm': 'OFF'}
}
device._handle_message(payload)
labels = {'name': NAME, 'serial': SERIAL}
self.metrics.update(NAME, device, is_state=False,
is_environmental=True)
cases = {
'dyson_temperature_celsius': 22.6,
'dyson_pm25_units': 0,
'dyson_pm10_units': 3,
'dyson_volatile_organic_compounds_units': 0,
'dyson_nitrogen_oxide_units': 3
}
for metric, want in cases.items():
got = self.registry.get_sample_value(metric, labels)
self.assertEqual(got, want, f'metric {metric}')
def test_update_formaldehyde_environmental(self):
device = libdyson.DysonPureCoolFormaldehyde(
SERIAL, CREDENTIALS, libdyson.DEVICE_TYPE_PURE_COOL_FORMALDEHYDE)
payload = {
'msg': 'ENVIRONMENTAL-CURRENT-SENSOR-DATA',
'time': '2021-03-17T15:09:23.000Z',
'data': {'tact': '2956', 'hact': '0047', 'pm10': '3', 'pm25': 'INIT',
'noxl': 30, 'va10': 'INIT', 'hcho': '0002', 'hchr': '0003',
'sltm': 'OFF'}
}
device._handle_message(payload)
labels = {'name': NAME, 'serial': SERIAL}
self.metrics.update(NAME, device, is_state=False,
is_environmental=True)
cases = {
'dyson_formaldehyde_units': 2,
}
for metric, want in cases.items():
got = self.registry.get_sample_value(metric, labels)
self.assertEqual(got, want, f'metric {metric}')
def test_update_v1_state(self):
device = libdyson.DysonPureHotCoolLink(
SERIAL, CREDENTIALS, libdyson.DEVICE_TYPE_PURE_HOT_COOL_LINK)
payload = {
'msg': 'STATE-CHANGE',
'time': '2021-03-17T15:27:30.000Z',
'mode-reason': 'PRC',
'state-reason': 'MODE',
'product-state': {
'fmod': ['AUTO', 'FAN'],
'fnst': ['FAN', 'FAN'],
'fnsp': ['AUTO', 'AUTO'],
'qtar': ['0003', '0003'],
'oson': ['ON', 'ON'],
'rhtm': ['ON', 'ON'],
'filf': ['2209', '2209'],
'ercd': ['NONE', 'NONE'],
'nmod': ['OFF', 'OFF'],
'wacd': ['NONE', 'NONE'],
'hmod': ['OFF', 'OFF'],
'hmax': ['2960', '2960'],
'hsta': ['OFF', 'OFF'],
'ffoc': ['ON', 'ON'],
'tilt': ['OK', 'OK']},
'scheduler': {'srsc': 'a58d', 'dstv': '0001', 'tzid': '0001'}}
device._handle_message(payload)
labels = {'name': NAME, 'serial': SERIAL}
self.metrics.update(NAME, device, is_state=True,
is_environmental=False)
cases = {
'dyson_fan_speed_units': -1,
'dyson_filter_life_seconds': 2209 * 60 * 60,
'dyson_quality_target_units': 3,
'dyson_heat_target_celsius': 23,
}
# We can't currently test Enums, so we skip those for now and only evaluate gauges.
for metric, want in cases.items():
got = self.registry.get_sample_value(metric, labels)
self.assertEqual(got, want, f'metric {metric}')
def test_update_v2_state(self):
device = libdyson.DysonPureHotCool(
SERIAL, CREDENTIALS, libdyson.DEVICE_TYPE_PURE_HOT_COOL)
payload = {
'msg': 'STATE-CHANGE',
'time': '2021-03-17T15:27:30.000Z',
'mode-reason': 'PRC',
'state-reason': 'MODE',
'product-state': {
'auto': ['ON', 'ON'],
'fpwr': ['ON', 'ON'],
'fmod': ['AUTO', 'FAN'],
'fnst': ['FAN', 'FAN'],
'fnsp': ['AUTO', 'AUTO'],
'nmdv': ['0002', '0002'],
'oson': ['ON', 'ON'],
'oscs': ['ON', 'ON'],
'osal': ['0136', '0136'],
'osau': ['0226', '0226'],
'rhtm': ['ON', 'ON'],
'cflr': ['0055', '0055'],
'hflr': ['0097', '0097'],
'cflt': ['SCOF', 'SCOF'],
'hflt': ['GCOM', 'GCOM'],
'ercd': ['NONE', 'NONE'],
'nmod': ['OFF', 'OFF'],
'wacd': ['NONE', 'NONE'],
'hmod': ['OFF', 'OFF'],
'hmax': ['2960', '2960'],
'hsta': ['OFF', 'OFF'],
'fdir': ['ON', 'ON']}}
device._handle_message(payload)
labels = {'name': NAME, 'serial': SERIAL}
self.metrics.update(NAME, device, is_state=True,
is_environmental=False)
# We can't currently test Enums, so we skip those for now and only evaluate gauges.
cases = {
'dyson_fan_speed_units': -1,
'dyson_carbon_filter_life_percent': 55,
'dyson_hepa_filter_life_percent': 97,
'dyson_heat_target_celsius': 23,
}
for metric, want in cases.items():
got = self.registry.get_sample_value(metric, labels)
self.assertEqual(got, want, f'metric {metric}')
if __name__ == '__main__':
unittest.main()