Files
2026-07-24 21:37:03 +02:00

55 lines
2.0 KiB
Python

from __future__ import annotations
import json
import sys
from pathlib import Path
sys.path.insert(0, str(Path(__file__).parents[1]))
from analysis_store import cleanup_legacy_files, current_filename
from session_store import merge_session_patch
def test_session_patch_preserves_other_sessions_and_detects_conflicts():
existing = {
"revision": 4,
"profile": {"start_date": "2026-01-01"},
"sessions": {"w01-d01": {"status": "completed"}, "w01-d02": {"status": "planned"}},
}
merged = merge_session_patch(existing, {
"expected_revision": 4,
"session_key": "w01-d02",
"session": {"status": "stopped"},
}, "2026-01-02T00:00:00Z")
assert merged["revision"] == 5
assert merged["sessions"]["w01-d01"]["status"] == "completed"
assert merged["sessions"]["w01-d02"]["status"] == "stopped"
try:
merge_session_patch(merged, {"expected_revision": 4}, "2026-01-02T00:00:01Z")
except RuntimeError as exc:
assert str(exc) == "revision_conflict:5"
else:
raise AssertionError("Revisionskonflikt wurde nicht erkannt")
def test_analysis_cleanup_keeps_exactly_current_scope(tmp_path: Path):
records = [
"week-01_20260101T100000Z.json",
"week-01_20260102T100000Z.json",
"week-02_20260101T100000Z.json",
"overall_20260101T100000Z.json",
"index.json",
"state.json",
]
for name in records:
(tmp_path / name).write_text(json.dumps({"name": name}), encoding="utf-8")
keep = current_filename({"type": "week", "week": 1})
(tmp_path / keep).write_text("{}", encoding="utf-8")
cleanup_legacy_files(tmp_path, {"type": "week", "week": 1}, keep)
names = {path.name for path in tmp_path.glob("*.json")}
assert keep in names
assert "week-02_20260101T100000Z.json" in names
assert "overall_20260101T100000Z.json" in names
assert "week-01_20260101T100000Z.json" not in names
assert "week-01_20260102T100000Z.json" not in names