import json from pathlib import Path import tempfile from threading import Event import time import unittest from unittest import mock from services.deferred_json_mapping import DeferredJsonMapping class DeferredJsonMappingTests(unittest.TestCase): def setUp(self): self.tmp = tempfile.TemporaryDirectory() self.stores = [] self.events = [] def tearDown(self): for event in self.events: event.set() for store in self.stores: store.close(timeout=1) self.tmp.cleanup() def event(self): event = Event() self.events.append(event) return event def store(self, load, save, **kwargs): cache = kwargs.pop("cache_file", Path(self.tmp.name) / (str(len(self.stores)) + ".json")) store = DeferredJsonMapping(cache, load, save, retry_interval=0.01, flush_interval=0.01, **kwargs) self.stores.append(store) return store def test_slow_restore_keeps_old_entries_and_does_not_block_updates(self): release, started = self.event(), self.event() writes = [] def load(): started.set() release.wait(3) return {"old": {"size": 7}} store = self.store(load, writes.append) store.start() self.assertTrue(started.wait(1)) start = time.monotonic() for n in range(20): store.update({str(n): {"size": n}}) self.assertLess(time.monotonic() - start, 0.5) self.assertFalse(store.flush(0.02)) self.assertEqual(writes, []) release.set() self.assertTrue(store.flush(2)) self.assertEqual(len(writes[-1]), 21) self.assertEqual(writes[-1]["old"], {"size": 7}) def test_blocked_save_has_one_worker_and_cannot_lose_concurrent_delta(self): entered, release = self.event(), self.event() writes = [] def save(items): writes.append(items) entered.set() release.wait(3) store = self.store(lambda: {"old": 1}, save) store.update({"first": 1}) self.assertTrue(entered.wait(1)) worker = store._thread store.update({"second": 2, "old": None}) self.assertIs(store._thread, worker) self.assertEqual(store.get("second"), 2) self.assertFalse(store.flush(0.02)) release.set() self.assertTrue(store.flush(2)) self.assertEqual(writes[-1], {"first": 1, "second": 2}) def test_failed_restore_never_overwrites_old_document(self): release = self.event() writes = [] def load(): if not release.is_set(): raise OSError("network mount temporarily unavailable") return {"old": 9} store = self.store(load, writes.append) store.update({"new": 2}) self.assertFalse(store.flush(0.03)) self.assertEqual(writes, []) release.set() self.assertTrue(store.flush(2)) self.assertEqual(writes[-1], {"old": 9, "new": 2}) def test_checkpoint_contains_delta_not_full_large_index(self): entered, release = self.event(), self.event() def save(items): entered.set() release.wait(3) store = self.store(lambda: {str(n): "x" * 100 for n in range(20000)}, save) store.snapshot(wait=1) store.update({"new": {"size": 3}}) self.assertTrue(entered.wait(1)) checkpoint = json.loads(store.cache_file.read_text(encoding="utf-8")) self.assertNotIn("items", checkpoint) self.assertLess(store.cache_file.stat().st_size, 500) def test_pending_delta_survives_restart_and_merges_counts_once(self): from services.error_message_service import ErrorMessageService release, entered = self.event(), self.event() def load(): entered.set() release.wait(3) return {"error": {"count": 10, "first_seen": "2026-01-01"}} store = self.store(load, lambda _: None) store.update({"error": {"count": 2, "first_seen": "2026-10-29"}}) self.assertTrue(entered.wait(1)) store.close(0) release.set() store.close(1) writes = [] restarted = self.store(load, writes.append, cache_file=store.cache_file, merge=ErrorMessageService._merge_unconfigured) self.assertTrue(restarted.flush(2)) self.assertEqual(writes[-1]["error"]["count"], 12) self.assertEqual(writes[-1]["error"]["first_seen"], "2026-01-01") def test_count_updates_during_save_are_not_dropped_or_double_counted(self): from services.error_message_service import ErrorMessageService entered, release = self.event(), self.event() writes = [] def save(items): writes.append(items) entered.set() release.wait(3) store = self.store(lambda: {"e": {"count": 10}}, save, merge=ErrorMessageService._merge_unconfigured) store.snapshot(wait=1) store.mutate(lambda data: data.update(e={"count": data["e"]["count"] + 1})) self.assertTrue(entered.wait(1)) store.mutate(lambda data: data.update(e={"count": data["e"]["count"] + 2})) release.set() self.assertTrue(store.flush(2)) self.assertEqual(writes[-1]["e"]["count"], 13) def test_clear_and_new_update_during_write_survive(self): entered, release = self.event(), self.event() writes = [] def save(items): writes.append(items) entered.set() release.wait(3) store = self.store(lambda: {"old": 1}, save) store.update({"new": 2}) self.assertTrue(entered.wait(1)) store.update({"new": 2}, clear=True) release.set() self.assertTrue(store.flush(2)) self.assertEqual(writes[-1], {"new": 2}) self.assertEqual(json.loads(store.cache_file.read_text())["pending"], {}) def test_stale_scan_cannot_overwrite_newer_entry(self): writes = [] store = self.store(lambda: {"a": {"local": True}}, writes.append) store.snapshot(wait=1) original = store.snapshot() store.update({"a": {"local": True, "webdav": True}, "b": 2}) store.update({"a": None}, expected=original) self.assertTrue(store.flush(2)) self.assertTrue(writes[-1]["a"]["webdav"]) self.assertEqual(writes[-1]["b"], 2) def test_clear_mutation_keeps_values_equal_to_previous_state(self): writes = [] store = self.store(lambda: {"a": 1, "b": 2}, writes.append) store.snapshot(wait=1) store.mutate(lambda data: data.update(a=1), clear=True) self.assertTrue(store.flush(2)) self.assertEqual(writes[-1], {"a": 1}) def test_failed_write_retries_without_losing_changes(self): writes = [] attempts = [] def save(items): attempts.append(1) if len(attempts) == 1: raise OSError("temporary I/O error") writes.append(items) store = self.store(lambda: {}, save) store.update({"a": 1}) self.assertTrue(store.flush(2)) self.assertEqual(writes[-1], {"a": 1}) def test_factory_shares_worker_and_shutdown_flush_is_bounded_and_reusable(self): import services.deferred_json_mapping as state source = Path(self.tmp.name) / "persist" / "state.json" runtime = Path(self.tmp.name) / "runtime" release, entered = self.event(), self.event() def save(items): entered.set() release.wait(3) with (mock.patch.object(state, "PERSIST_DIR", source.parent), mock.patch.object(state, "RUNTIME_DIR", runtime)): self.assertIsNone(state.deferred_mapping(runtime / "local.json", lambda: {}, save)) first = state.deferred_mapping(source, lambda: {}, save) second = state.deferred_mapping(source, lambda: {}, save) self.stores.append(first) self.addCleanup(lambda: state._STORES.pop(str(source.absolute()), None)) self.assertIs(first, second) first._flush_interval = 0.01 first.update({"a": 1}) self.assertTrue(entered.wait(1)) start = time.monotonic() state.flush_deferred_mappings(timeout=0.02) self.assertLess(time.monotonic() - start, 0.3) second.update({"b": 2}) release.set() self.assertTrue(first.flush(2)) self.assertEqual(first.snapshot(), {"a": 1, "b": 2})