eomsangdonandClaude Opus 5.5
c87fa54f5b
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01Se7ytqbohmBQnfYpwYFnYf
144 lines
4.8 KiB
Python
144 lines
4.8 KiB
Python
"""431 [초기화] 빈틈(분석 424 3장).
|
|
|
|
- 스냅샷 없는 폴백 = 덧 측점(`plus_stations.json`)도 지움
|
|
- 배수 폴더 없는 스냅샷 = 관 편집 비움
|
|
- 복원 길 = stage 2 · 3 검토 대기 · B07 · B08 STALE · M02 구조물 사본 맞춤(진짜 `safe_sync`)
|
|
"""
|
|
|
|
import asyncio
|
|
import json
|
|
from contextlib import asynccontextmanager
|
|
from pathlib import Path
|
|
from uuid import uuid4
|
|
|
|
from common_util.common_util_initial_snapshot import (
|
|
restore_snapshot_files,
|
|
snapshot_dir,
|
|
wipe_edited_masters,
|
|
)
|
|
|
|
EDITS = "B04_PreProcess/drainage/edits"
|
|
|
|
|
|
def _touch(path: Path, text: str = "{}") -> None:
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
path.write_text(text, encoding="utf-8")
|
|
|
|
|
|
def test_폴백은_덧_측점도_지움(tmp_path: Path):
|
|
_touch(tmp_path / "B05_Profile/route/plus_stations.json")
|
|
_touch(tmp_path / "B05_Profile/route/structures.json")
|
|
_touch(tmp_path / EDITS / "pipe_points.json")
|
|
removed = wipe_edited_masters(tmp_path)
|
|
assert "B05_Profile/route/plus_stations.json" in removed
|
|
assert not (tmp_path / "B05_Profile/route/plus_stations.json").exists()
|
|
assert not (tmp_path / EDITS).exists()
|
|
|
|
|
|
def _snapshot_without_drainage(root: Path) -> None:
|
|
_touch(snapshot_dir(root) / "B05_Profile__route" / "planned_route.csv", "x,y\n")
|
|
_touch(snapshot_dir(root) / "db.json", json.dumps({"routes": [{"id": 1}]}))
|
|
|
|
|
|
def test_배수_폴더_없는_스냅샷은_관_편집_비움(tmp_path: Path):
|
|
_snapshot_without_drainage(tmp_path)
|
|
_touch(tmp_path / EDITS / "pipe_points.json", '[{"chainage_m": 3}]')
|
|
_touch(tmp_path / "B04_PreProcess/drainage/basins.json") # 분석 산출물은 그대로
|
|
restore_snapshot_files(tmp_path)
|
|
assert not (tmp_path / EDITS).exists()
|
|
assert (tmp_path / "B04_PreProcess/drainage/basins.json").is_file()
|
|
|
|
|
|
def test_옛_스냅샷은_edits_만_되돌림(tmp_path: Path):
|
|
_snapshot_without_drainage(tmp_path)
|
|
_touch(snapshot_dir(tmp_path) / "B04_PreProcess__drainage__edits/pipe_points.json", "[]")
|
|
_touch(tmp_path / EDITS / "pipe_points.json", '[{"chainage_m": 3}]')
|
|
restore_snapshot_files(tmp_path)
|
|
assert (tmp_path / EDITS / "pipe_points.json").read_text(encoding="utf-8") == "[]"
|
|
|
|
|
|
class _Cursor:
|
|
def __init__(self, log: list[tuple[str, tuple]]):
|
|
self._log = log
|
|
self.rowcount = 1
|
|
|
|
async def __aenter__(self):
|
|
return self
|
|
|
|
async def __aexit__(self, *_):
|
|
return False
|
|
|
|
async def execute(self, sql, params=None):
|
|
self._log.append((" ".join(sql.split()), params))
|
|
|
|
|
|
class _Connection:
|
|
def __init__(self, log):
|
|
self._log = log
|
|
|
|
def cursor(self, *_a, **_k):
|
|
return _Cursor(self._log)
|
|
|
|
async def begin(self):
|
|
pass
|
|
|
|
async def commit(self):
|
|
self._log.append(("COMMIT", ()))
|
|
|
|
async def rollback(self):
|
|
self._log.append(("ROLLBACK", ()))
|
|
|
|
|
|
class _Pool:
|
|
def __init__(self, log):
|
|
self._log = log
|
|
|
|
@asynccontextmanager
|
|
async def acquire(self):
|
|
yield _Connection(self._log)
|
|
|
|
|
|
def _async(value):
|
|
async def run(*_a, **_k):
|
|
return value
|
|
|
|
return run
|
|
|
|
|
|
def test_복원_길은_단계_되돌리고_사본_맞춤(tmp_path: Path, monkeypatch):
|
|
from B05_Profile import B05_Profile_Router_Corridor as corridor
|
|
from B05_Profile import B05_Profile_Router_Lifecycle as lifecycle
|
|
from common_util import common_util_initial_snapshot as snapshot
|
|
|
|
_snapshot_without_drainage(tmp_path)
|
|
_touch(tmp_path / EDITS / "pipe_points.json", "[]")
|
|
# 사용자가 넣은 구조물의 사본 — 초기값엔 그 구조물 없음
|
|
_touch(tmp_path / "templates/structures/s9/manifest.json", '{"key": "SS9"}')
|
|
|
|
log: list[tuple[str, tuple]] = []
|
|
monkeypatch.setattr(lifecycle, "get_db_pool", lambda: _Pool(log))
|
|
monkeypatch.setattr(lifecycle, "get_project_storage_relative_path", _async("p"))
|
|
monkeypatch.setattr(lifecycle, "resolve_stored_project_path", lambda _p: str(tmp_path))
|
|
monkeypatch.setattr(lifecycle, "get_latest_route", _async({"id": 9}))
|
|
monkeypatch.setattr(snapshot, "restore_initial_snapshot", _async(9))
|
|
monkeypatch.setattr(corridor, "prune_corridor_files", lambda *_a: 0)
|
|
|
|
pid = uuid4()
|
|
response = asyncio.run(lifecycle.reset_route_design(pid, None))
|
|
body = json.loads(response.body)
|
|
assert body["restored"] is True and body["route_id"] == 9
|
|
|
|
started = [
|
|
p[3]
|
|
for sql, p in log
|
|
if sql.startswith("UPDATE project_workflow_stages SET state = 'IN_PROGRESS'")
|
|
]
|
|
assert started == [2, 3]
|
|
stale = [p[1] for sql, p in log if "SET state = 'STALE'" in sql]
|
|
assert stale == [2, 3] # stage > 3 (B07 · B08 · B09) 완료분 STALE
|
|
assert log.index(("COMMIT", ())) > max(
|
|
i for i, (sql, _) in enumerate(log) if "project_workflow_stages" in sql
|
|
)
|
|
assert not (tmp_path / EDITS).exists()
|
|
assert not (tmp_path / "templates/structures/s9").exists()
|