Mercurial > personal > weather-server

comparison weather_server/logfile_test.py @ 0:efe7a1eff167

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
Create initial logger for weather server.
author Paul Fisher <paul@pfish.zone>
date Sat, 28 Sep 2019 23:17:21 -0400
parents
children beb42c835c52
comparison
equal deleted inserted replaced
-1:000000000000 0:efe7a1eff167
1 import contextlib
2 import datetime
3 import pathlib
4 import tempfile
5 import unittest
6
7 import bson
8 import pytz
9
10 from . import common
11 from . import logfile
12 from . import types
13
14
15 def ts(n):
16 return datetime.datetime.utcfromtimestamp(n).replace(tzinfo=pytz.UTC)
17
18
19 class LoggerTest(unittest.TestCase):
20
21 maxDiff = None
22
23 def setUp(self):
24 super().setUp()
25 self.temp_dir = tempfile.TemporaryDirectory()
26 self.log_path = pathlib.Path(self.temp_dir.name) / 'test.bson'
27
28 def tearDown(self):
29 self.temp_dir.cleanup()
30 super().tearDown()
31
32 def test_empty(self):
33 with contextlib.closing(logfile.Logger(str(self.log_path))) as logger:
34 self.assertEqual(logger.data, ())
35
36 def test_loading(self):
37 with self.log_path.open('wb') as outfile:
38 outfile.write(common.bson_encode(dict(
39 sample_time=ts(123),
40 temp_c=420,
41 rh_pct=69,
42 ingest_time=ts(125),
43 )))
44 outfile.write(b'garbage to ignore')
45 with contextlib.closing(logfile.Logger(str(self.log_path))) as logger:
46 self.assertEqual(
47 logger.data,
48 (types.Reading(ts(123), 420, 69, ts(125)),))
49
50 def test_writing(self):
51 with self.log_path.open('wb') as outfile:
52 outfile.write(common.bson_encode(dict(
53 sample_time=ts(123),
54 temp_c=420,
55 rh_pct=69,
56 ingest_time=ts(125),
57 )))
58 with contextlib.closing(logfile.Logger(str(self.log_path))) as logger:
59 logger.write_rows([
60 types.Reading(ts(100), 999, 666, ts(101)),
61 types.Reading(ts(125), 333, 777, ts(200)),
62 ])
63 self.assertEqual(
64 logger.data,
65 (
66 types.Reading(ts(123), 420, 69, ts(125)),
67 types.Reading(ts(125), 333, 777, ts(200)),
68 )
69 )
70
71 self.assertEqual(self.read_bsons(), [
72 dict(
73 sample_time=ts(123),
74 temp_c=420,
75 rh_pct=69,
76 ingest_time=ts(125),
77 ),
78 dict(
79 sample_time=ts(125),
80 temp_c=333,
81 rh_pct=777,
82 ingest_time=ts(200),
83 ),
84 ])
85
86 def test_outside_writes(self):
87 with contextlib.closing(logfile.Logger(str(self.log_path))) as logger:
88 logger.write_rows([
89 types.Reading(ts(100), 999, 666, ts(101)),
90 types.Reading(ts(125), 333, 777, ts(200)),
91 ])
92 with self.log_path.open('ab') as outfile:
93 outfile.write(common.bson_encode(dict(
94 sample_time=ts(1024),
95 temp_c=256,
96 rh_pct=128,
97 ingest_time=ts(4096),
98 )))
99 outfile.flush()
100 self.assertEqual(logger.data, (
101 types.Reading(ts(100), 999, 666, ts(101)),
102 types.Reading(ts(125), 333, 777, ts(200)),
103 types.Reading(ts(1024), 256, 128, ts(4096)),
104 ))
105
106 def read_bsons(self):
107 with self.log_path.open('rb') as infile:
108 return bson.decode_all(
109 infile.read(), common.BSON_OPTIONS)
110
111
112 if __name__ == '__main__':
113 unittest.main()