108 lines
4.8 KiB
Python
108 lines
4.8 KiB
Python
from concurrent.futures import ThreadPoolExecutor
|
|
from datetime import datetime, timedelta, timezone
|
|
import json
|
|
from pathlib import Path
|
|
import tempfile
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
import pyarrow.parquet as pq
|
|
|
|
from solar_relay.cli import next_poll_at
|
|
from solar_relay.storage import Store, StoreError
|
|
from tests.test_status import observed
|
|
|
|
|
|
class ParquetTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.directory = tempfile.TemporaryDirectory()
|
|
self.addCleanup(self.directory.cleanup)
|
|
self.path = Path(self.directory.name) / "archive"
|
|
self.store = Store(self.path)
|
|
self.now = datetime(2026, 11, 1, 12, tzinfo=timezone.utc)
|
|
|
|
def test_real_parquet_has_typed_readings_and_original_payload(self):
|
|
value = observed(self.now)
|
|
value["readings"]["boiler_enabled"] = False
|
|
self.store.add(value)
|
|
path = next(self.path.glob("date=*/samples.parquet"))
|
|
self.assertEqual(path.read_bytes()[:4], b"PAR1")
|
|
row = pq.ParquetFile(path).read().to_pylist()[0]
|
|
self.assertEqual(row["production_ac_w"], 2000)
|
|
self.assertIs(row["boiler_enabled"], False)
|
|
self.assertIsNone(row["uploaded_at"])
|
|
self.assertEqual(json.loads(row["payload"]), value)
|
|
self.assertEqual(self.store.status()["format"], "parquet")
|
|
|
|
def test_retention_uses_ack_time_and_never_deletes_pending(self):
|
|
items = [observed(self.now - timedelta(days=90, minutes=index)) for index in range(4)]
|
|
for item in items:
|
|
self.store.add(item)
|
|
self.store.acknowledge([items[0]["id"]], now=self.now - timedelta(days=31))
|
|
self.store.acknowledge([items[1]["id"]], now=self.now - timedelta(days=30))
|
|
self.store.acknowledge([items[2]["id"]], now=self.now)
|
|
self.assertEqual(self.store.prune(now=self.now), 1)
|
|
self.assertEqual({s["id"] for s in self.store.records()}, {s["id"] for s in items[1:]})
|
|
self.assertEqual(self.store.pending(100), [items[3]])
|
|
self.assertEqual(self.store.prune(now=self.now + timedelta(days=31)), 2)
|
|
self.assertEqual(list(self.store.records()), [items[3]])
|
|
|
|
def test_replay_cancels_expiry_until_the_next_acknowledgement(self):
|
|
value = observed(self.now)
|
|
self.store.add(value)
|
|
self.store.acknowledge([value["id"]], now=self.now - timedelta(days=40))
|
|
self.store.replay()
|
|
self.assertEqual(self.store.prune(now=self.now), 0)
|
|
self.store.acknowledge([value["id"]], now=self.now)
|
|
self.assertEqual(self.store.prune(now=self.now), 0)
|
|
|
|
def test_interrupted_atomic_write_preserves_the_previous_partition(self):
|
|
first, second = observed(self.now), observed(self.now + timedelta(minutes=5))
|
|
self.store.add(first)
|
|
with patch("solar_relay.storage.os.replace", side_effect=OSError("simulated write failure")):
|
|
with self.assertRaises(OSError):
|
|
self.store.add(second)
|
|
self.assertEqual(list(Store(self.path).records()), [first])
|
|
self.assertEqual(list(self.path.glob("date=*/.samples-*.tmp")), [])
|
|
|
|
def test_two_store_instances_cannot_lose_concurrent_records(self):
|
|
other = Store(self.path)
|
|
values = [observed(self.now + timedelta(minutes=index * 5)) for index in range(12)]
|
|
def add(pair):
|
|
index, value = pair
|
|
(other if index % 2 else self.store).add(value)
|
|
with ThreadPoolExecutor(max_workers=2) as workers:
|
|
list(workers.map(add, enumerate(values)))
|
|
self.assertEqual({value["id"] for value in self.store.records()}, {value["id"] for value in values})
|
|
|
|
def test_pruning_an_entire_day_removes_its_parquet_file(self):
|
|
value = observed(self.now - timedelta(days=60))
|
|
self.store.add(value)
|
|
self.store.acknowledge([value["id"]], now=self.now - timedelta(days=31))
|
|
self.assertEqual(self.store.prune(now=self.now), 1)
|
|
self.assertEqual(list(self.path.glob("date=*/samples.parquet")), [])
|
|
self.assertEqual(self.store.status()["samples"], 0)
|
|
|
|
def test_corrupt_archive_fails_without_silently_dropping_records(self):
|
|
value = observed(self.now)
|
|
self.store.add(value)
|
|
next(self.path.glob("date=*/samples.parquet")).write_bytes(b"broken")
|
|
with self.assertRaises(StoreError):
|
|
self.store.pending(100)
|
|
|
|
def test_archive_path_requires_a_directory(self):
|
|
occupied = self.path.parent / "existing-file"
|
|
occupied.write_bytes(b"existing data")
|
|
with self.assertRaises(StoreError):
|
|
Store(occupied)
|
|
self.assertEqual(occupied.read_bytes(), b"existing data")
|
|
|
|
def test_next_poll_is_aligned_to_the_five_minute_clock(self):
|
|
self.assertEqual(next_poll_at(301, 300), 600)
|
|
self.assertEqual(next_poll_at(599.5, 300), 600)
|
|
self.assertEqual(next_poll_at(600, 300), 900)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|