File size: 1,966 Bytes
e2ce547 | 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 | from __future__ import annotations
import gzip
import json
import tempfile
import unittest
from pathlib import Path
from polymarket_collector.storage import RotatingJsonlWriter
class StorageTests(unittest.IsolatedAsyncioTestCase):
async def test_rotates_and_compresses_jsonl(self) -> None:
with tempfile.TemporaryDirectory() as temporary:
root = Path(temporary)
writer = RotatingJsonlWriter(
root,
"test",
rotation_seconds=3600,
max_file_bytes=1,
flush_interval_seconds=1,
)
await writer.start()
await writer.write({"hello": "world"})
await writer.close()
files = list(root.rglob("*.jsonl.gz"))
self.assertEqual(len(files), 1)
with gzip.open(files[0], "rt", encoding="utf-8") as handle:
self.assertEqual(json.loads(handle.readline()), {"hello": "world"})
self.assertEqual(list(root.rglob("*.part")), [])
async def test_recovers_part_file_on_start(self) -> None:
with tempfile.TemporaryDirectory() as temporary:
root = Path(temporary)
part = root / "2026/07/22/crashed.jsonl.part"
part.parent.mkdir(parents=True)
part.write_text('{"recovered":true}\n', encoding="utf-8")
writer = RotatingJsonlWriter(
root,
"test",
rotation_seconds=3600,
max_file_bytes=1024,
flush_interval_seconds=1,
)
await writer.start()
await writer.close()
recovered = list(root.rglob("*.jsonl.gz"))
self.assertEqual(len(recovered), 1)
with gzip.open(recovered[0], "rt", encoding="utf-8") as handle:
self.assertEqual(json.loads(handle.readline()), {"recovered": True})
if __name__ == "__main__":
unittest.main()
|