| 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}) |
|
|
