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
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
|
# -*- coding: utf-8 -*-
"""Set of series helper functions for test."""
from __future__ import absolute_import
from __future__ import division
from __future__ import print_function
from __future__ import unicode_literals
from datetime import datetime, timedelta
import unittest
import warnings
from unittest import mock
from influxdb import SeriesHelper, InfluxDBClient
from requests.exceptions import ConnectionError
class TestSeriesHelper(unittest.TestCase):
"""Define the SeriesHelper test object."""
@classmethod
def setUpClass(cls):
"""Set up the TestSeriesHelper object."""
super(TestSeriesHelper, cls).setUpClass()
TestSeriesHelper.client = InfluxDBClient(
'host',
8086,
'username',
'password',
'database'
)
class MySeriesHelper(SeriesHelper):
"""Define a SeriesHelper object."""
class Meta:
"""Define metadata for the SeriesHelper object."""
client = TestSeriesHelper.client
series_name = 'events.stats.{server_name}'
fields = ['some_stat']
tags = ['server_name', 'other_tag']
bulk_size = 5
autocommit = True
TestSeriesHelper.MySeriesHelper = MySeriesHelper
def setUp(self):
"""Check that MySeriesHelper has empty datapoints."""
super(TestSeriesHelper, self).setUp()
self.assertEqual(
TestSeriesHelper.MySeriesHelper._json_body_(),
[],
'Resetting helper in teardown did not empty datapoints.')
def tearDown(self):
"""Deconstruct the TestSeriesHelper object."""
super(TestSeriesHelper, self).tearDown()
TestSeriesHelper.MySeriesHelper._reset_()
self.assertEqual(
TestSeriesHelper.MySeriesHelper._json_body_(),
[],
'Resetting helper did not empty datapoints.')
def test_auto_commit(self):
"""Test write_points called after valid number of events."""
class AutoCommitTest(SeriesHelper):
"""Define a SeriesHelper instance to test autocommit."""
class Meta:
"""Define metadata for AutoCommitTest."""
series_name = 'events.stats.{server_name}'
fields = ['some_stat']
tags = ['server_name', 'other_tag']
bulk_size = 5
client = InfluxDBClient()
autocommit = True
fake_write_points = mock.MagicMock()
AutoCommitTest(server_name='us.east-1', some_stat=159, other_tag='gg')
AutoCommitTest._client.write_points = fake_write_points
AutoCommitTest(server_name='us.east-1', some_stat=158, other_tag='gg')
AutoCommitTest(server_name='us.east-1', some_stat=157, other_tag='gg')
AutoCommitTest(server_name='us.east-1', some_stat=156, other_tag='gg')
self.assertFalse(fake_write_points.called)
AutoCommitTest(server_name='us.east-1', some_stat=3443, other_tag='gg')
self.assertTrue(fake_write_points.called)
@mock.patch('influxdb.helper.SeriesHelper._current_timestamp')
def testSingleSeriesName(self, current_timestamp):
"""Test JSON conversion when there is only one series name."""
current_timestamp.return_value = current_date = datetime.today()
TestSeriesHelper.MySeriesHelper(
server_name='us.east-1', other_tag='ello', some_stat=159)
TestSeriesHelper.MySeriesHelper(
server_name='us.east-1', other_tag='ello', some_stat=158)
TestSeriesHelper.MySeriesHelper(
server_name='us.east-1', other_tag='ello', some_stat=157)
TestSeriesHelper.MySeriesHelper(
server_name='us.east-1', other_tag='ello', some_stat=156)
expectation = [
{
"measurement": "events.stats.us.east-1",
"tags": {
"other_tag": "ello",
"server_name": "us.east-1"
},
"fields": {
"some_stat": 159
},
"time": current_date,
},
{
"measurement": "events.stats.us.east-1",
"tags": {
"other_tag": "ello",
"server_name": "us.east-1"
},
"fields": {
"some_stat": 158
},
"time": current_date,
},
{
"measurement": "events.stats.us.east-1",
"tags": {
"other_tag": "ello",
"server_name": "us.east-1"
},
"fields": {
"some_stat": 157
},
"time": current_date,
},
{
"measurement": "events.stats.us.east-1",
"tags": {
"other_tag": "ello",
"server_name": "us.east-1"
},
"fields": {
"some_stat": 156
},
"time": current_date,
}
]
rcvd = TestSeriesHelper.MySeriesHelper._json_body_()
self.assertTrue(all([el in expectation for el in rcvd]) and
all([el in rcvd for el in expectation]),
'Invalid JSON body of time series returned from '
'_json_body_ for one series name: {0}.'.format(rcvd))
@mock.patch('influxdb.helper.SeriesHelper._current_timestamp')
def testSeveralSeriesNames(self, current_timestamp):
"""Test JSON conversion when there are multiple series names."""
current_timestamp.return_value = current_date = datetime.today()
TestSeriesHelper.MySeriesHelper(
server_name='us.east-1', some_stat=159, other_tag='ello')
TestSeriesHelper.MySeriesHelper(
server_name='fr.paris-10', some_stat=158, other_tag='ello')
TestSeriesHelper.MySeriesHelper(
server_name='lu.lux', some_stat=157, other_tag='ello')
TestSeriesHelper.MySeriesHelper(
server_name='uk.london', some_stat=156, other_tag='ello')
expectation = [
{
'fields': {
'some_stat': 157
},
'measurement': 'events.stats.lu.lux',
'tags': {
'other_tag': 'ello',
'server_name': 'lu.lux'
},
"time": current_date,
},
{
'fields': {
'some_stat': 156
},
'measurement': 'events.stats.uk.london',
'tags': {
'other_tag': 'ello',
'server_name': 'uk.london'
},
"time": current_date,
},
{
'fields': {
'some_stat': 158
},
'measurement': 'events.stats.fr.paris-10',
'tags': {
'other_tag': 'ello',
'server_name': 'fr.paris-10'
},
"time": current_date,
},
{
'fields': {
'some_stat': 159
},
'measurement': 'events.stats.us.east-1',
'tags': {
'other_tag': 'ello',
'server_name': 'us.east-1'
},
"time": current_date,
}
]
rcvd = TestSeriesHelper.MySeriesHelper._json_body_()
self.assertTrue(all([el in expectation for el in rcvd]) and
all([el in rcvd for el in expectation]),
'Invalid JSON body of time series returned from '
'_json_body_ for several series names: {0}.'
.format(rcvd))
@mock.patch('influxdb.helper.SeriesHelper._current_timestamp')
def testSeriesWithoutTimeField(self, current_timestamp):
"""Test that time is optional on a series without a time field."""
current_date = datetime.today()
yesterday = current_date - timedelta(days=1)
current_timestamp.return_value = yesterday
TestSeriesHelper.MySeriesHelper(
server_name='us.east-1', other_tag='ello',
some_stat=159, time=current_date
)
TestSeriesHelper.MySeriesHelper(
server_name='us.east-1', other_tag='ello',
some_stat=158,
)
point1, point2 = TestSeriesHelper.MySeriesHelper._json_body_()
self.assertTrue('time' in point1 and 'time' in point2)
self.assertEqual(point1['time'], current_date)
self.assertEqual(point2['time'], yesterday)
def testSeriesWithoutAllTags(self):
"""Test that creating a data point without a tag throws an error."""
class MyTimeFieldSeriesHelper(SeriesHelper):
class Meta:
client = TestSeriesHelper.client
series_name = 'events.stats.{server_name}'
fields = ['some_stat', 'time']
tags = ['server_name', 'other_tag']
bulk_size = 5
autocommit = True
self.assertRaises(NameError, MyTimeFieldSeriesHelper,
**{"server_name": 'us.east-1',
"some_stat": 158})
@mock.patch('influxdb.helper.SeriesHelper._current_timestamp')
def testSeriesWithTimeField(self, current_timestamp):
"""Test that time is optional on a series with a time field."""
current_date = datetime.today()
yesterday = current_date - timedelta(days=1)
current_timestamp.return_value = yesterday
class MyTimeFieldSeriesHelper(SeriesHelper):
class Meta:
client = TestSeriesHelper.client
series_name = 'events.stats.{server_name}'
fields = ['some_stat', 'time']
tags = ['server_name', 'other_tag']
bulk_size = 5
autocommit = True
MyTimeFieldSeriesHelper(
server_name='us.east-1', other_tag='ello',
some_stat=159, time=current_date
)
MyTimeFieldSeriesHelper(
server_name='us.east-1', other_tag='ello',
some_stat=158,
)
point1, point2 = MyTimeFieldSeriesHelper._json_body_()
self.assertTrue('time' in point1 and 'time' in point2)
self.assertEqual(point1['time'], current_date)
self.assertEqual(point2['time'], yesterday)
def testInvalidHelpers(self):
"""Test errors in invalid helpers."""
class MissingMeta(SeriesHelper):
"""Define instance of SeriesHelper for missing meta."""
pass
class MissingClient(SeriesHelper):
"""Define SeriesHelper for missing client data."""
class Meta:
"""Define metadat for MissingClient."""
series_name = 'events.stats.{server_name}'
fields = ['time', 'server_name']
autocommit = True
class MissingSeriesName(SeriesHelper):
"""Define instance of SeriesHelper for missing series."""
class Meta:
"""Define metadata for MissingSeriesName."""
fields = ['time', 'server_name']
class MissingFields(SeriesHelper):
"""Define instance of SeriesHelper for missing fields."""
class Meta:
"""Define metadata for MissingFields."""
series_name = 'events.stats.{server_name}'
class InvalidTimePrecision(SeriesHelper):
"""Define instance of SeriesHelper for invalid time precision."""
class Meta:
"""Define metadata for InvalidTimePrecision."""
series_name = 'events.stats.{server_name}'
time_precision = "ks"
fields = ['time', 'server_name']
autocommit = True
for cls in [MissingMeta, MissingClient, MissingFields,
MissingSeriesName, InvalidTimePrecision]:
self.assertRaises(
AttributeError, cls, **{'time': 159,
'server_name': 'us.east-1'})
@unittest.skip("Fails on py32")
def testWarnBulkSizeZero(self):
"""Test warning for an invalid bulk size."""
class WarnBulkSizeZero(SeriesHelper):
class Meta:
client = TestSeriesHelper.client
series_name = 'events.stats.{server_name}'
fields = ['time', 'server_name']
tags = []
bulk_size = 0
autocommit = True
with warnings.catch_warnings(record=True) as w:
warnings.simplefilter("always")
try:
WarnBulkSizeZero(time=159, server_name='us.east-1')
except ConnectionError:
# Server defined in the client is invalid, we're testing
# the warning only.
pass
self.assertEqual(len(w), 1,
'{0} call should have generated one warning.'
.format(WarnBulkSizeZero))
self.assertIn('forced to 1', str(w[-1].message),
'Warning message did not contain "forced to 1".')
@unittest.skip("Test produces different number of warnings for Python3.11 and Python3.12 thus not possible to pass for both at the same time")
def testWarnBulkSizeNoEffect(self):
"""Test warning for a set bulk size but autocommit False."""
class WarnBulkSizeNoEffect(SeriesHelper):
"""Define SeriesHelper for warning on bulk size."""
class Meta:
"""Define metadat for WarnBulkSizeNoEffect."""
series_name = 'events.stats.{server_name}'
fields = ['time', 'server_name']
bulk_size = 5
tags = []
autocommit = False
with warnings.catch_warnings(record=True) as w:
warnings.simplefilter("always")
WarnBulkSizeNoEffect(time=159, server_name='us.east-1')
self.assertEqual(len(w), 1,
'{0} call should have generated one warning.'
.format(WarnBulkSizeNoEffect))
self.assertIn('has no affect', str(w[-1].message),
'Warning message did not contain "has not affect".')
def testSeriesWithRetentionPolicy(self):
"""Test that the data is saved with the specified retention policy."""
my_policy = 'my_policy'
class RetentionPolicySeriesHelper(SeriesHelper):
class Meta:
client = InfluxDBClient()
series_name = 'events.stats.{server_name}'
fields = ['some_stat', 'time']
tags = ['server_name', 'other_tag']
bulk_size = 2
autocommit = True
retention_policy = my_policy
fake_write_points = mock.MagicMock()
RetentionPolicySeriesHelper(
server_name='us.east-1', some_stat=159, other_tag='gg')
RetentionPolicySeriesHelper._client.write_points = fake_write_points
RetentionPolicySeriesHelper(
server_name='us.east-1', some_stat=158, other_tag='aa')
kall = fake_write_points.call_args
args, kwargs = kall
self.assertTrue('retention_policy' in kwargs)
self.assertEqual(kwargs['retention_policy'], my_policy)
def testSeriesWithoutRetentionPolicy(self):
"""Test that the data is saved without any retention policy."""
class NoRetentionPolicySeriesHelper(SeriesHelper):
class Meta:
client = InfluxDBClient()
series_name = 'events.stats.{server_name}'
fields = ['some_stat', 'time']
tags = ['server_name', 'other_tag']
bulk_size = 2
autocommit = True
fake_write_points = mock.MagicMock()
NoRetentionPolicySeriesHelper(
server_name='us.east-1', some_stat=159, other_tag='gg')
NoRetentionPolicySeriesHelper._client.write_points = fake_write_points
NoRetentionPolicySeriesHelper(
server_name='us.east-1', some_stat=158, other_tag='aa')
kall = fake_write_points.call_args
args, kwargs = kall
self.assertTrue('retention_policy' in kwargs)
self.assertEqual(kwargs['retention_policy'], None)
|
