Mercurial > personal > weather-server
view weather_server/logfile_test.py @ 16:9a609bcf0809
Port main script to TypeScript and prepare for serving it.
author | Paul Fisher <paul@pfish.zone> |
---|---|
date | Sat, 12 Oct 2019 14:03:52 -0400 |
parents | efe7a1eff167 |
children | beb42c835c52 |
line wrap: on
line source
import contextlib import datetime import pathlib import tempfile import unittest import bson import pytz from . import common from . import logfile from . import types def ts(n): return datetime.datetime.utcfromtimestamp(n).replace(tzinfo=pytz.UTC) class LoggerTest(unittest.TestCase): maxDiff = None def setUp(self): super().setUp() self.temp_dir = tempfile.TemporaryDirectory() self.log_path = pathlib.Path(self.temp_dir.name) / 'test.bson' def tearDown(self): self.temp_dir.cleanup() super().tearDown() def test_empty(self): with contextlib.closing(logfile.Logger(str(self.log_path))) as logger: self.assertEqual(logger.data, ()) def test_loading(self): with self.log_path.open('wb') as outfile: outfile.write(common.bson_encode(dict( sample_time=ts(123), temp_c=420, rh_pct=69, ingest_time=ts(125), ))) outfile.write(b'garbage to ignore') with contextlib.closing(logfile.Logger(str(self.log_path))) as logger: self.assertEqual( logger.data, (types.Reading(ts(123), 420, 69, ts(125)),)) def test_writing(self): with self.log_path.open('wb') as outfile: outfile.write(common.bson_encode(dict( sample_time=ts(123), temp_c=420, rh_pct=69, ingest_time=ts(125), ))) with contextlib.closing(logfile.Logger(str(self.log_path))) as logger: logger.write_rows([ types.Reading(ts(100), 999, 666, ts(101)), types.Reading(ts(125), 333, 777, ts(200)), ]) self.assertEqual( logger.data, ( types.Reading(ts(123), 420, 69, ts(125)), types.Reading(ts(125), 333, 777, ts(200)), ) ) self.assertEqual(self.read_bsons(), [ dict( sample_time=ts(123), temp_c=420, rh_pct=69, ingest_time=ts(125), ), dict( sample_time=ts(125), temp_c=333, rh_pct=777, ingest_time=ts(200), ), ]) def test_outside_writes(self): with contextlib.closing(logfile.Logger(str(self.log_path))) as logger: logger.write_rows([ types.Reading(ts(100), 999, 666, ts(101)), types.Reading(ts(125), 333, 777, ts(200)), ]) with self.log_path.open('ab') as outfile: outfile.write(common.bson_encode(dict( sample_time=ts(1024), temp_c=256, rh_pct=128, ingest_time=ts(4096), ))) outfile.flush() self.assertEqual(logger.data, ( types.Reading(ts(100), 999, 666, ts(101)), types.Reading(ts(125), 333, 777, ts(200)), types.Reading(ts(1024), 256, 128, ts(4096)), )) def read_bsons(self): with self.log_path.open('rb') as infile: return bson.decode_all( infile.read(), common.BSON_OPTIONS) if __name__ == '__main__': unittest.main()