test(472-6 저장 시험): 한 본문 저장 시험이 박은 노선 380 대신 검증_sub8 최신 노선 · 추가 측점 자리를 그때 읽음 — 노선 바뀌어도 됨

- 최신 노선 id = 저장 창구가 견주는 workflow 노선 · 추가 측점 = 가운데 두 측점 사이
- 옛 차례 대 한 본문 비교에서 행 자동 번호(id) 뺌 — 새 추가 측점 행은 두 길이 다른 번호를 받음

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
2026-10-19 15:42:09 +09:00
co-authored by Claude Opus 5.5
parent 8d577ef330
commit 83e85ed13b
+41 -5
View File
@@ -4,6 +4,7 @@
② 실제 창구(검증_sub8 · DB · storage) — 중간 차례 실패 → DB 7 표 행 · storage 해시 시작 그대로 ·
차례 이름 · 까닭 · restored. 같은 본문 정상 = 옛 차례(안쪽 창구 차례로 부름)와 행 · 파일 같음.
검증 프로젝트가 없는 자리(다른 PC · DB 없음)는 ② 건너뜀.
② 는 살아 있는 검증_sub8 사용 · 노선 바뀌어도 됨(최신 노선 id · 추가 측점 자리를 그때 읽음).
./venv/Scripts/python.exe -m pytest resources/tester/test_472_6_design_save.py -q
"""
@@ -106,13 +107,46 @@ def _state(root: Path) -> tuple[dict, dict]:
return clean, files
def _latest_route() -> tuple[int, float]:
"""(최신 노선 id — 저장 창구가 견주는 그 값, 추가 측점 자리 — 가운데 두 측점 사이)."""
from B06_Section.B06_Section_Repository import get_workflow_route_context
from config.config_db import close_db_pool, init_db_pool
async def run():
pool = await init_db_pool()
try:
async with pool.acquire() as connection:
return await get_workflow_route_context(connection, UUID(PROJECT))
finally:
await close_db_pool()
route_id = int(asyncio.run(run())["route_id"])
longitudinal = ROOT_DIR / "B06_Section" / "longitudinal" / "longitudinal.json"
stations = json.loads(longitudinal.read_text(encoding="utf-8"))["stations"]
middle = len(stations) // 2
a, b = (float(stations[i]["chainage_m"]) for i in (middle - 1, middle))
return route_id, round((a + b) / 2, 2)
def _without_ids(rows: dict) -> dict:
"""행 자동 번호(`id`) 뺀 비교 꼴 — 새 추가 측점 행은 두 길이 다른 번호를 받음."""
return {
table: sorted(
json.dumps({k: v for k, v in row.items() if k != "id"}, sort_keys=True)
for row in values
)
for table, values in rows.items()
}
def _request(**extra):
from B06_Section.B06_Section_Router_DesignSave import DesignSaveRequest
route_id, plus_at = _latest_route()
body = {
"route_id": 380,
"route_id": route_id,
"settings": {"design_speed_kph": 20, "terrain_type": "special", "max_grade_pct": 13.0},
"plus_stations": {"chainages": [500.0]},
"plus_stations": {"chainages": [plus_at]},
"cross": {},
**extra,
}
@@ -171,7 +205,7 @@ def test_정상_한_본문은_옛_차례와_같음() -> None:
files = guard.snapshot_storage(ROOT_DIR, guard.STORAGE_SCOPE)
await save_page_settings(UUID(PROJECT), request.settings)
await put_plus_stations(UUID(PROJECT), request.plus_stations)
await save_sections(UUID(PROJECT), 380, request.cross)
await save_sections(UUID(PROJECT), request.route_id, request.cross)
return rows, files
finally:
await close_db_pool()
@@ -195,7 +229,7 @@ def test_정상_한_본문은_옛_차례와_같음() -> None:
assert status == 200 and body["status"] == "success", body
assert set(body["steps"]) == {"settings", "plusStations", "cross"}
new = _state(ROOT_DIR)
assert new[0] == old[0]
assert _without_ids(new[0]) == _without_ids(old[0])
assert new[1] == old[1]
finally:
asyncio.run(undo())
@@ -225,13 +259,15 @@ def test_같은_프로젝트_저장이_겹치면_뒤_요청이_기다림() -> No
from B06_Section.B06_Section_Router_DesignSave import DesignSaveRequest, save_design
from config.config_db import close_db_pool, init_db_pool
route_id = _latest_route()[0]
async def run():
await init_db_pool()
try:
lock = guard.project_lock(PROJECT)
await lock.acquire()
task = asyncio.create_task(
save_design(UUID(PROJECT), DesignSaveRequest(route_id=380), None)
save_design(UUID(PROJECT), DesignSaveRequest(route_id=route_id), None)
)
await asyncio.sleep(0.3)
assert not task.done() # 앞 저장(잠금)이 끝날 때까지 기다림