test(export): add unit tests for load_export_state_from_disk validation

Author: leostar0412

tests/test_export_state_store.py

@@ -0,0 +1,83 @@
+"""Tests for utils/export_state_store.load_export_state_from_disk validation."""
+
+from __future__ import annotations
+
+import json
+from pathlib import Path
+
+from utils.export_state_store import load_export_state_from_disk
+
+
+def test_load_rejects_non_object_json(tmp_path: Path):
+    p = tmp_path / "export_state.json"
+    p.write_text(json.dumps([1, 2, 3]), encoding="utf-8")
+    assert load_export_state_from_disk(str(p)) == {}
+
+
+def test_load_rejects_null_json(tmp_path: Path):
+    p = tmp_path / "export_state.json"
+    p.write_text("null", encoding="utf-8")
+    assert load_export_state_from_disk(str(p)) == {}
+
+
+def test_load_sanitizes_non_dict_sessions(tmp_path: Path):
+    p = tmp_path / "export_state.json"
+    p.write_text(
+        json.dumps(
+            {
+                "lastExportTime": "2026-01-01T00:00:00",
+                "exportedCount": 1,
+                "sessions": [],
+            }
+        ),
+        encoding="utf-8",
+    )
+    out = load_export_state_from_disk(str(p))
+    assert out["sessions"] == {}
+    assert out["lastExportTime"] == "2026-01-01T00:00:00"
+    assert out["exportedCount"] == 1
+
+
+def test_load_adds_sessions_when_missing_but_has_last_export(tmp_path: Path):
+    p = tmp_path / "export_state.json"
+    p.write_text(
+        json.dumps({"lastExportTime": "2026-01-01T00:00:00", "exportedCount": 0}),
+        encoding="utf-8",
+    )
+    out = load_export_state_from_disk(str(p))
+    assert out["sessions"] == {}
+    assert out["lastExportTime"] == "2026-01-01T00:00:00"
+
+
+def test_load_legacy_flat_dict_unchanged_shape(tmp_path: Path):
+    p = tmp_path / "export_state.json"
+    legacy = {"uuid-one": 1740000000.0}
+    p.write_text(json.dumps(legacy), encoding="utf-8")
+    out = load_export_state_from_disk(str(p))
+    assert out == {"sessions": legacy}
+
+
+def test_export_state_lock_windows_branch_uses_msvcrt_when_no_fcntl(
+    monkeypatch, tmp_path: Path
+):
+    """When ``fcntl`` is absent, use ``msvcrt.locking`` (cross-process on Windows)."""
+    import utils.export_state_store as mod
+
+    monkeypatch.setattr(mod, "fcntl", None)
+    calls: list[tuple[int, int]] = []
+
+    class FakeMsvcrt:
+        LK_LOCK = 1
+        LK_UNLCK = 2
+
+        @staticmethod
+        def locking(fd, mode, nbytes):
+            calls.append((mode, nbytes))
+
+    monkeypatch.setattr(mod, "msvcrt", FakeMsvcrt)
+
+    state_file = tmp_path / "export_state.json"
+    state_file.write_text("{}", encoding="utf-8")
+    with mod.export_state_lock(str(state_file)):
+        assert (FakeMsvcrt.LK_LOCK, 1) in calls
+    assert calls[-1] == (FakeMsvcrt.LK_UNLCK, 1)

utils/export_state_store.py

@@ -11,8 +11,14 @@
 try:
     import fcntl
 except ImportError:
-    fcntl = None  # Windows: fall back to threading lock (same process only)
+    fcntl = None
 
+try:
+    import msvcrt
+except ImportError:
+    msvcrt = None
+
+# Only when neither fcntl nor msvcrt exists (very rare): same-process only.
 _fallback_locks: dict[str, threading.Lock] = {}
 _fallback_locks_guard = threading.Lock()
 
@@ -30,7 +36,12 @@ def _fallback_lock_for(path: str) -> threading.Lock:
 
 @contextmanager
 def export_state_lock(state_path: str | None = None):
-    """Serialize export_state.json reads/writes (POSIX: flock; else threading)."""
+    """Serialize export_state.json reads/writes across processes.
+
+    POSIX: ``flock`` on a sidecar ``*.lock`` file. Windows: ``msvcrt.locking`` on
+    the same sidecar (byte-range lock). If neither is available, falls back to
+    a per-path ``threading.Lock`` (same process only).
+    """
     path = EXPORT_STATE_FILE if state_path is None else state_path
     if fcntl is not None:
         lock_path = path + ".lock"
@@ -44,6 +55,28 @@ def export_state_lock(state_path: str | None = None):
         finally:
             fcntl.flock(lock_fp.fileno(), fcntl.LOCK_UN)
             lock_fp.close()
+    elif msvcrt is not None:
+        lock_path = path + ".lock"
+        dir_name = os.path.dirname(lock_path)
+        if dir_name:
+            os.makedirs(dir_name, exist_ok=True)
+        if not os.path.exists(lock_path):
+            with open(lock_path, "wb") as f:
+                f.write(b"\x00")
+        lock_fp = open(lock_path, "r+b")
+        try:
+            if os.path.getsize(lock_path) == 0:
+                lock_fp.write(b"\x00")
+                lock_fp.flush()
+            lock_fp.seek(0)
+            msvcrt.locking(lock_fp.fileno(), msvcrt.LK_LOCK, 1)
+            try:
+                yield
+            finally:
+                lock_fp.seek(0)
+                msvcrt.locking(lock_fp.fileno(), msvcrt.LK_UNLCK, 1)
+        finally:
+            lock_fp.close()
     else:
         with _fallback_lock_for(path):
             yield
@@ -53,6 +86,8 @@ def load_export_state_from_disk(state_path: str | None = None) -> dict:
     """Load state from disk (call under :func:`export_state_lock` for consistency).
 
     Migrates legacy flat ``{session_id: mtime, ...}`` to ``{"sessions": ...}``.
+    Returns a dict with a mapping ``sessions``; malformed top-level values or
+    a non-dict ``sessions`` entry are sanitized so callers always see a dict.
     """
     path = EXPORT_STATE_FILE if state_path is None else state_path
     if not os.path.isfile(path):
@@ -62,8 +97,13 @@ def load_export_state_from_disk(state_path: str | None = None) -> dict:
             data = json.load(f)
     except Exception:
         return {}
+    if not isinstance(data, dict):
+        return {}
     if "sessions" not in data and "lastExportTime" not in data:
         return {"sessions": data}
+    if not isinstance(data.get("sessions"), dict):
+        data = dict(data)
+        data["sessions"] = {}
     return data