eomsangdonandClaude Opus 5.5
7dbae140c1
- sub5 고침(끌기 onMove → 시설 보관소 move)을 받은 뒤 본문 만들기도 같은 차례로 - sub5 · sub6 받은 뒤 남은 차이: 관 측점 정수 m 격자 붙이기 4 곳(440.24 · 619.71 · 720.44 · 899.91) · B06 저장 차례에서 옮긴 관 측점 횡단 patch 가 받을 행이 없어 버려짐 Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01AeeQRLknEpP4SFjVkjDWDs
501 lines
23 KiB
Python
501 lines
23 KiB
Python
"""B05 · B06 저장 왕복 — 사용자가 바꾼 값이 저장 · 다시 읽기 뒤 그대로인가(PLAN 19장).
|
|
|
|
환경 배정 storage 프로젝트(랩탑_보조)의 **사본**에서 화면 저장 흐름과 같은 차례로 서버 API 를
|
|
부르고, 다시 읽은 값(pipe_points.json · structures.json · 종단 파일 · 응답)을 사용자가 고른 값과
|
|
0 차이로 대조. 저장 본문은 화면 코드 그대로(`helper_b05_save_payload.mjs` — 시설 되붙이기 ·
|
|
구조물 측점 칸 · 구간값 예약).
|
|
|
|
- 파일: 프로젝트 폴더를 임시 폴더로 복사(큰 원본 · 점군 제외) · `STORAGE_BASE_DIR` 를 그리로 돌림.
|
|
- DB: 읽기는 진짜 DB(같은 프로젝트 행) · 세션을 읽기 전용으로 열어 쓰기가 새지 않음 ·
|
|
쓰는 함수는 받은 값을 적어 두는 가짜로 갈음(DB 에 남기는 값도 보낸 값과 대조).
|
|
- 원본 프로젝트 · 사용자 프로젝트 · DB 는 바뀌지 않음.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import shutil
|
|
import subprocess
|
|
from contextlib import asynccontextmanager
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
import pytest
|
|
|
|
ROOT = Path(__file__).resolve().parents[2]
|
|
PROJECT_ID = "936be972-11bc-46c2-8bf3-b15d8de7df0d" # OWNERS.md — laptop-sub `랩탑_보조`
|
|
SOURCE = ROOT / "storage" / "1" / "3" / PROJECT_ID
|
|
# 저장 흐름이 읽고 쓰는 것만 — 원본 입력(B03) · 점군(structured.npz 등)은 안 씀
|
|
_COPY_DIRS = ("B04_PreProcess/drainage", "B05_Profile", "B06_Section")
|
|
_COPY_MAX_BYTES = 20 * 1024 * 1024
|
|
|
|
pytestmark = pytest.mark.skipif(not SOURCE.is_dir(), reason="배정 storage 프로젝트가 없음")
|
|
|
|
|
|
def _copy_project(dest: Path) -> None:
|
|
for rel in _COPY_DIRS:
|
|
shutil.copytree(SOURCE / rel, dest / rel)
|
|
for rel in ("B04_PreProcess/models", "B04_PreProcess/processed"):
|
|
(dest / rel).mkdir(parents=True, exist_ok=True)
|
|
for item in (SOURCE / rel).iterdir():
|
|
if item.is_file() and item.stat().st_size <= _COPY_MAX_BYTES:
|
|
shutil.copy2(item, dest / rel / item.name)
|
|
for item in SOURCE.iterdir():
|
|
if item.is_file():
|
|
shutil.copy2(item, dest / item.name)
|
|
|
|
|
|
class Recorder:
|
|
"""DB 에 쓰는 함수 대신 — 받은 값을 적어 둠."""
|
|
|
|
def __init__(self) -> None:
|
|
self.calls: list[tuple[str, dict[str, Any]]] = []
|
|
|
|
def fake(self, name: str, result: Any = None):
|
|
async def _fake(*args: Any, **kwargs: Any) -> Any:
|
|
self.calls.append((name, {"args": args[1:], **kwargs}))
|
|
return result
|
|
|
|
return _fake
|
|
|
|
def of(self, name: str) -> list[dict[str, Any]]:
|
|
return [call for key, call in self.calls if key == name]
|
|
|
|
|
|
@pytest.fixture()
|
|
def api(tmp_path, monkeypatch):
|
|
aiomysql = pytest.importorskip("aiomysql")
|
|
from fastapi import FastAPI
|
|
from fastapi.testclient import TestClient
|
|
|
|
import B04_PreProcess.B04_PreProcess_Router_Basins as basins
|
|
import B05_Profile.B05_Profile_Router as route
|
|
import B05_Profile.B05_Profile_Router_Lifecycle as lifecycle
|
|
import B05_Profile.B05_Profile_Structures_Router as structures
|
|
import B06_Section.B06_Section_Repository_Bulk as bulk
|
|
import B06_Section.B06_Section_Router_Confirm as sections
|
|
import common_util.common_util_storage as storage
|
|
import config.config_db as config_db
|
|
from config import config_system as cfg
|
|
|
|
storage_root = tmp_path / "storage"
|
|
_copy_project(storage_root / "1" / "3" / PROJECT_ID)
|
|
monkeypatch.setattr(storage, "STORAGE_BASE_DIR", str(storage_root))
|
|
|
|
rec = Recorder()
|
|
monkeypatch.setattr(route, "update_longitudinal_grade_summary", rec.fake("grade_summary"))
|
|
monkeypatch.setattr(structures, "_invalidate_downstream", rec.fake("invalidate", True))
|
|
monkeypatch.setattr(lifecycle, "confirm_route", rec.fake("confirm_route"))
|
|
monkeypatch.setattr(lifecycle, "complete_stage", rec.fake("complete_stage"))
|
|
monkeypatch.setattr(bulk, "merge_cross_section_designs", rec.fake("designs", 0))
|
|
monkeypatch.setattr(sections, "merge_cross_section_designs", rec.fake("designs", 0))
|
|
monkeypatch.setattr(sections, "merge_longitudinal_section_options", rec.fake("options"))
|
|
monkeypatch.setattr(sections, "merge_longitudinal_section_data", rec.fake("data"))
|
|
monkeypatch.setattr(sections, "confirm_sections_for_route", rec.fake("confirm_sections"))
|
|
monkeypatch.setattr(sections, "confirm_route_status", rec.fake("confirm_route"))
|
|
monkeypatch.setattr(sections, "complete_stage", rec.fake("complete_stage"))
|
|
monkeypatch.setattr(sections, "start_stage", rec.fake("start_stage"))
|
|
monkeypatch.setattr(sections, "_recompute_stored_designs", rec.fake("recompute"))
|
|
real_known = sections.get_cross_section_chainages
|
|
|
|
async def spy_known(connection, route_id):
|
|
rows = await real_known(connection, route_id)
|
|
rec.calls.append(("known", {"chainages": list(rows)}))
|
|
return rows
|
|
|
|
monkeypatch.setattr(sections, "get_cross_section_chainages", spy_known)
|
|
|
|
@asynccontextmanager
|
|
async def lifespan(_app):
|
|
try:
|
|
pool = await aiomysql.create_pool(
|
|
host=cfg.DB_HOST,
|
|
port=cfg.DB_PORT,
|
|
db=cfg.DB_NAME,
|
|
user=cfg.DB_USER,
|
|
password=cfg.DB_PASSWORD,
|
|
minsize=1,
|
|
maxsize=6,
|
|
autocommit=True,
|
|
charset="utf8mb4",
|
|
# 읽기 전용 세션 — 가짜로 못 갈음한 쓰기가 있으면 DB 를 바꾸지 않고 터짐
|
|
init_command="SET time_zone='+00:00', SESSION tx_read_only=1",
|
|
)
|
|
except Exception as exc: # DB 가 안 닿는 곳
|
|
pytest.skip(f"DB 에 안 닿음: {exc}")
|
|
config_db.db_pool = pool
|
|
try:
|
|
yield
|
|
finally:
|
|
config_db.db_pool = None
|
|
pool.close()
|
|
await pool.wait_closed()
|
|
|
|
app = FastAPI(lifespan=lifespan)
|
|
for module in (basins, route, lifecycle, structures, sections):
|
|
app.include_router(module.router)
|
|
with TestClient(app) as client:
|
|
client.rec = rec # type: ignore[attr-defined]
|
|
client.root = storage_root / "1" / "3" / PROJECT_ID # type: ignore[attr-defined]
|
|
yield client
|
|
|
|
|
|
# ── 화면 코드 그대로 본문 만들기(node) ────────────────────────────────────────
|
|
HELPER = Path(__file__).with_name("helper_b05_save_payload.mjs")
|
|
STATION_INTERVAL_M = 20.0
|
|
# 화면이 관 옵션 수정 끝에 측점을 다시 읽어 이만큼 넘게 다르면 관을 옮김
|
|
# (`B05_Profile_UI_Drainage_Panel.ts` `updatePipeFacility`)
|
|
PIPE_MOVE_THRESHOLD_M = 0.005
|
|
|
|
|
|
def _screen(jobs: list[dict[str, Any]]) -> list[Any]:
|
|
if shutil.which("node") is None:
|
|
pytest.skip("node 가 없음")
|
|
out = subprocess.run(
|
|
["node", str(HELPER), str(ROOT)],
|
|
input=json.dumps(jobs),
|
|
cwd=ROOT,
|
|
capture_output=True,
|
|
text=True,
|
|
encoding="utf-8",
|
|
timeout=300,
|
|
check=True,
|
|
)
|
|
return json.loads(out.stdout.strip().splitlines()[-1])
|
|
|
|
|
|
# ── 비교 모양 ─────────────────────────────────────────────────────────────────
|
|
LONGITUDINAL = "B06_Section/longitudinal/longitudinal.json"
|
|
PIPES_FILE = "B04_PreProcess/drainage/edits/pipe_points.json"
|
|
STRUCTURES_FILE = "B05_Profile/route/structures.json"
|
|
|
|
|
|
def _key(chainage: float) -> str:
|
|
return f"{chainage:.2f}"
|
|
|
|
|
|
def _pipes_view(points: list[dict[str, Any]]) -> dict[str, dict[str, Any]]:
|
|
"""관 — 측점(0.01) → 시설 · 구간 · 옵션(좌표 · 생성 사유는 뺌)."""
|
|
return {
|
|
_key(p["chainage_m"]): {
|
|
"facility": p.get("facility", "pipe"),
|
|
"span": (p.get("start_m"), p.get("end_m")),
|
|
"options": p.get("options") or {},
|
|
}
|
|
for p in points
|
|
}
|
|
|
|
|
|
def _read(root: Path, rel: str) -> Any:
|
|
return json.loads((root / rel).read_text(encoding="utf-8"))
|
|
|
|
|
|
def _strip(point: dict[str, Any]) -> dict[str, Any]:
|
|
"""화면이 서버에 되보내는 관 — 좌표는 서버가 다시 냄(`DetailPipeInput`)."""
|
|
return {k: v for k, v in point.items() if k != "lonlat"}
|
|
|
|
|
|
class Flow:
|
|
"""화면 저장 흐름 하나(B05 [임시저장] · B06 [저장]).
|
|
|
|
사용자가 고른 값(`want_*`)을 따로 적어 두고, 화면 코드가 만든 본문을 서버에 보낸 뒤
|
|
다시 읽은 값을 고른 값과 대조. 바꿀 대상은 사본에 있는 것 중 자리 순번으로 고름.
|
|
"""
|
|
|
|
def __init__(self, api) -> None:
|
|
self.api = api
|
|
self.base = f"/api/projects/{PROJECT_ID}"
|
|
latest = api.get(f"{self.base}/route/latest").json()
|
|
self.route_id = latest["route"]["id"]
|
|
self.surface = latest["surface_params"]
|
|
self.model_id = latest["route"]["surface_model_id"]
|
|
self.diffs: list[str] = []
|
|
self.sent_irregular: list[float] = []
|
|
self._plan()
|
|
|
|
def server_pipes(self) -> list[dict[str, Any]]:
|
|
got = self.api.get(f"{self.base}/drainage/pipe-points").json()["pipe_points"]
|
|
return [_strip(p) for p in got]
|
|
|
|
# ── 사용자가 고른 값 ──
|
|
def _plan(self) -> None:
|
|
pipes = self.server_pipes()
|
|
if len(pipes) < 9:
|
|
pytest.skip(f"관이 {len(pipes)} 개 — 시험에 9 개 넘게 필요")
|
|
self.pipes0 = pipes
|
|
ch = [p["chainage_m"] for p in pipes]
|
|
self.pipe_move = (ch[2], round(ch[2] + 5.35, 2)) # 관 옮김 — 시설 · 옵션이 따라가야 함
|
|
self.pipe_option = (ch[5], {"pipe_diameter_mm": 1350}) # 옵션만 — 자리 그대로
|
|
self.culvert = (ch[8], {"pipe_diameter_mm": 1200}) # B06 조정창 구간값 예약
|
|
self.stale_move = (ch[4], round(ch[4] + 4.76, 2)) # 옛 세션에 남은 관 이동 예약
|
|
self.pipe_new = round((ch[6] + ch[7]) / 2, 2) # 새 관 하나(맨 배관)
|
|
want = []
|
|
for p in pipes:
|
|
item = dict(p)
|
|
if item["chainage_m"] == self.pipe_move[0]:
|
|
item.update(chainage_m=self.pipe_move[1], source="user")
|
|
elif item["chainage_m"] == self.pipe_option[0]:
|
|
item["options"] = {**(item.get("options") or {}), **self.pipe_option[1]}
|
|
elif item["chainage_m"] == self.culvert[0]:
|
|
item["options"] = {**(item.get("options") or {}), **self.culvert[1]}
|
|
want.append(item)
|
|
want.append({"chainage_m": self.pipe_new, "source": "user"})
|
|
self.want_pipes = sorted(want, key=lambda x: x["chainage_m"])
|
|
|
|
got = self.api.get(f"{self.base}/route/structures").json()
|
|
self.revision0, items = got["revision"], got["structures"]
|
|
spans = [s for s in items if s.get("start_m") is not None and s.get("end_m") is not None]
|
|
if len(spans) < 3:
|
|
pytest.skip("구간 구조물이 3 개 넘게 필요")
|
|
moved, optioned, memoed = spans[0], spans[1], spans[2]
|
|
width = moved["end_m"] - moved["start_m"]
|
|
to = round(moved["start_m"] + 1.37, 2) # 측점 칸에 소수 2자리로 쳐 넣음
|
|
self.struct_move = (moved["structure_id"], to, round(to + width, 2))
|
|
self.want_structures = json.loads(json.dumps(items))
|
|
for s in self.want_structures:
|
|
if s["structure_id"] == moved["structure_id"]:
|
|
s.update(chainage_m=to, start_m=to, end_m=round(to + width, 2))
|
|
s["memo"] = "옮긴 뒤 메모" # 옮긴 칸에서 메모만 한 번 더 고침
|
|
elif s["structure_id"] == optioned["structure_id"]:
|
|
s["options"] = {**s["options"], "height_m": 3}
|
|
s["memo"] = "왕복 시험 메모"
|
|
elif s["structure_id"] == memoed["structure_id"]:
|
|
s["memo"] = "메모만 바꿈"
|
|
|
|
lon = _read(self.api.root, LONGITUDINAL)
|
|
grid = [s for s in lon["stations"] if s.get("kind") == "regular"]
|
|
g0, g1 = grid[4], grid[len(grid) // 2]
|
|
self.offsets = {f"{g0['chainage_m']:.3f}": 0.35, f"{g1['chainage_m']:.3f}": -0.2}
|
|
side = grid[3].get("uphill_side")
|
|
self.uphill = (grid[3]["chainage_m"], "left" if side != "left" else "right")
|
|
self.cross_patches = [
|
|
{
|
|
"chainage_m": g0["chainage_m"],
|
|
"rock_boundary_offset_m": -0.8,
|
|
"cut_slope_ratio_user": 0.5,
|
|
},
|
|
{"chainage_m": self.pipe_move[1], "pipe_length_m": 9.0},
|
|
]
|
|
|
|
# ── 화면 코드가 만드는 본문 ──
|
|
def b05_pipe_payload(self) -> list[dict[str, Any]]:
|
|
"""B05 `savePipes` — 화면 관 목록(측점 · 사유) + 시설 보관소 `attach`."""
|
|
(station,) = _screen(
|
|
[{"op": "station", "chainage": self.pipe_option[0], "interval": STATION_INTERVAL_M}]
|
|
)
|
|
entries = [dict(p) for p in self.pipes0]
|
|
screen_ch = {}
|
|
for p in self.want_pipes:
|
|
c = p["chainage_m"]
|
|
if c == self.pipe_option[0] and abs(station - c) > PIPE_MOVE_THRESHOLD_M:
|
|
c = station # 옵션 폼 확정이 측점을 표시값(0.1)에서 다시 읽어 관을 옮김
|
|
screen_ch[p["chainage_m"]] = c
|
|
for e in entries: # 폼에서 고친 시설 정보는 그 관의 (새) 자리로 보관소에 들어감
|
|
if e["chainage_m"] == self.pipe_option[0]:
|
|
e["chainage_m"] = screen_ch[self.pipe_option[0]]
|
|
e["options"] = {**(e.get("options") or {}), **self.pipe_option[1]}
|
|
points = [
|
|
{
|
|
"chainage_m": screen_ch[p["chainage_m"]],
|
|
"source": "confirmed" if p.get("source") == "user" else p.get("source", "user"),
|
|
}
|
|
for p in self.want_pipes
|
|
]
|
|
job = {"op": "attach", "entries": entries, "moves": [self.pipe_move], "points": points}
|
|
(attached,) = _screen([job])
|
|
return attached
|
|
|
|
def b06_pipe_payload(self) -> list[dict[str, Any]]:
|
|
"""B06 `persistPipes` — 화면 관 상태 그대로(구간값은 예약으로 따로 나감)."""
|
|
before = {p["chainage_m"]: p.get("options") for p in self.pipes0}
|
|
out = []
|
|
for p in self.want_pipes:
|
|
item = dict(p)
|
|
if p["chainage_m"] == self.culvert[0]:
|
|
item.pop("options", None)
|
|
if before[self.culvert[0]]:
|
|
item["options"] = before[self.culvert[0]]
|
|
out.append(item)
|
|
return out
|
|
|
|
def screen_structures(self, *, reread: bool) -> list[dict[str, Any]]:
|
|
"""구조물 — 옮긴 칸에서 메모를 고치면 `commit` 이 측점 칸(표시 0.1)을 다시 읽음."""
|
|
items = json.loads(json.dumps(self.want_structures))
|
|
if not reread:
|
|
return items
|
|
sid, start, end = self.struct_move
|
|
ch = next(s for s in items if s["structure_id"] == sid)
|
|
a, b, c = _screen(
|
|
[
|
|
{"op": "station", "chainage": v, "interval": STATION_INTERVAL_M}
|
|
for v in (start, ch["chainage_m"], end)
|
|
]
|
|
)
|
|
ch.update(start_m=a, chainage_m=b, end_m=c)
|
|
return items
|
|
|
|
def culvert_payload(self, current: list[dict[str, Any]]) -> list[dict[str, Any]] | None:
|
|
"""`flushCulvertOptions` — 구간값 예약 + 옛 세션 이동 예약을 서버 목록에 얹음."""
|
|
pending = {_key(self.culvert[0]): self.culvert[1]}
|
|
moves = {_key(self.stale_move[0]): self.stale_move[1]}
|
|
(body,) = _screen(
|
|
[{"op": "culvert", "server": current, "pending": pending, "moves": moves}]
|
|
)
|
|
return body
|
|
|
|
# ── API(화면 Api_Fetch 와 같은 경로 · 본문) ──
|
|
def put_pipes(self, pipes: list[dict[str, Any]]) -> None:
|
|
res = self.api.put(f"{self.base}/drainage/pipe-points", json={"points": pipes})
|
|
assert res.status_code == 200, res.text
|
|
|
|
def flush_culvert(self) -> None:
|
|
body = self.culvert_payload(
|
|
self.api.get(f"{self.base}/drainage/pipe-points").json()["pipe_points"]
|
|
)
|
|
if body is not None:
|
|
self.put_pipes(body)
|
|
|
|
def put_alignment(self) -> None:
|
|
body = {"route_id": self.route_id, "station_offsets": self.offsets, "curve_radii": {}}
|
|
res = self.api.put(f"{self.base}/route/profile-alignment", json=body)
|
|
assert res.status_code == 200, res.text
|
|
|
|
def put_structures(self, items: list[dict[str, Any]]) -> None:
|
|
revision = self.api.get(f"{self.base}/route/structures").json()["revision"]
|
|
body = {"base_revision": revision, "structures": items}
|
|
res = self.api.put(f"{self.base}/route/structures", json=body)
|
|
assert res.status_code == 200, res.text
|
|
|
|
def confirm(self, *, with_irregular: bool) -> None:
|
|
body: dict[str, Any] = {
|
|
"uphill_overrides": [{"chainage_m": self.uphill[0], "side": self.uphill[1]}]
|
|
}
|
|
if with_irregular: # B05 — 관 투영(물넘이포장 뺌) · A군 구조물
|
|
self.sent_irregular = [
|
|
p["chainage_m"] for p in self.server_pipes() if p.get("facility") != "ford_pavement"
|
|
]
|
|
body.update(
|
|
filter_key=self.surface["source_filter"],
|
|
method=self.surface["method"],
|
|
smooth=self.surface["smooth"],
|
|
surface_model_id=self.model_id,
|
|
irregular_stations=[
|
|
{"chainage_m": c, "structure": "배관"} for c in self.sent_irregular
|
|
],
|
|
)
|
|
res = self.api.post(f"{self.base}/route/confirm?mark_stage_complete=false", json=body)
|
|
assert res.status_code == 200, res.text
|
|
|
|
def save_sections(self) -> None:
|
|
body = {"cross_patches": self.cross_patches}
|
|
res = self.api.post(f"{self.base}/sections/{self.route_id}/save", json=body)
|
|
assert res.status_code == 200, res.text
|
|
|
|
# ── 다시 읽기 · 대조(사용자가 고른 값 기준) ──
|
|
def check(self) -> None:
|
|
root, d = self.api.root, self.diffs
|
|
want = _pipes_view(self.want_pipes)
|
|
for where, got in (
|
|
("응답", _pipes_view(self.server_pipes())),
|
|
("pipe_points.json", _pipes_view(_read(root, PIPES_FILE)["points"])),
|
|
):
|
|
for k in sorted(set(want) | set(got), key=float):
|
|
if want.get(k) != got.get(k):
|
|
d.append(f"관 {k} ({where}) — 고름 {want.get(k)} · 읽음 {got.get(k)}")
|
|
saved = {s["structure_id"]: s for s in _read(root, STRUCTURES_FILE)["structures"]}
|
|
for s in self.want_structures:
|
|
got_s = saved.get(s["structure_id"], {})
|
|
for field in ("chainage_m", "start_m", "end_m", "options", "memo", "placement_source"):
|
|
if s.get(field) != got_s.get(field):
|
|
d.append(
|
|
f"구조물 {s['type_id']} {field} — "
|
|
f"고름 {s.get(field)!r} · 읽음 {got_s.get(field)!r}"
|
|
)
|
|
lon = _read(root, LONGITUDINAL)
|
|
edits = (lon["profile_alignment"].get("edits") or {}).get("station_offsets") or {}
|
|
if edits != self.offsets:
|
|
d.append(f"계획고 수정 — 고름 {self.offsets} · 읽음 {edits}")
|
|
up = next((s for s in lon["stations"] if abs(s["chainage_m"] - self.uphill[0]) < 0.005), {})
|
|
if (up.get("uphill_side"), up.get("uphill_side_source")) != (self.uphill[1], "user"):
|
|
d.append(
|
|
f"상단측 {self.uphill[0]} — 고름 {self.uphill[1]} · "
|
|
f"읽음 {up.get('uphill_side')}/{up.get('uphill_side_source')}"
|
|
)
|
|
for c in self.sent_irregular:
|
|
near = [s["chainage_m"] for s in lon["stations"] if abs(s["chainage_m"] - c) < 1.0]
|
|
if not any(abs(n - c) < 0.005 for n in near):
|
|
d.append(f"관 측점 {c} — 종단 측점에 없음(가까운 측점 {near})")
|
|
self._check_designs()
|
|
|
|
def _check_designs(self) -> None:
|
|
"""횡단 patch — DB 로 가는 값이 고른 값과 같고, 받을 행이 있어야 함(없으면 DB 가 버림)."""
|
|
rec = self.api.rec
|
|
known = [c for call in rec.of("known") for c in call["chainages"]]
|
|
designs = rec.of("designs")
|
|
made = [e[0] for call in designs if call.get("replace") for e in call["entries"]]
|
|
patched = {
|
|
round(e[0], 2): e[1]
|
|
for call in designs
|
|
if call.get("replace") is False
|
|
for e in call["entries"]
|
|
}
|
|
for patch in self.cross_patches:
|
|
ch = patch["chainage_m"]
|
|
body = {k: v for k, v in patch.items() if k != "chainage_m"}
|
|
got = patched.get(round(ch, 2))
|
|
if got is None or any(got.get(k) != v for k, v in body.items()):
|
|
self.diffs.append(f"횡단 patch {ch} — 고름 {body} · DB 로 {got}")
|
|
elif not any(abs(float(k) - ch) < 0.01 for k in [*known, *made]):
|
|
self.diffs.append(f"횡단 patch {ch} — 받을 행이 없어 DB 가 버림")
|
|
|
|
|
|
def _snapshot(root: Path) -> dict[str, Any]:
|
|
lon = _read(root, LONGITUDINAL)
|
|
return {
|
|
"pipe_points.json": _read(root, PIPES_FILE)["points"],
|
|
"structures.json": _read(root, STRUCTURES_FILE)["structures"],
|
|
"계획고 수정": lon["profile_alignment"].get("edits"),
|
|
"종단 측점": [
|
|
(s["chainage_m"], s.get("kind"), s.get("uphill_side")) for s in lon["stations"]
|
|
],
|
|
}
|
|
|
|
|
|
def _resave_unchanged(flow: Flow) -> None:
|
|
"""다시 열어 아무것도 안 바꾸고 [저장] — 저장본이 한 글자도 안 바뀌어야 함."""
|
|
flow.put_pipes(flow.server_pipes())
|
|
flow.put_alignment()
|
|
flow.put_structures(flow.api.get(f"{flow.base}/route/structures").json()["structures"])
|
|
|
|
|
|
def _b05_save(flow: Flow) -> None:
|
|
"""B05 [임시저장] 차례 — 관 → 계획선 → 구조물 → 확정(관 측점 · 상단측) → 구간값 → 횡단."""
|
|
flow.put_pipes(flow.b05_pipe_payload())
|
|
flow.put_alignment()
|
|
flow.put_structures(flow.screen_structures(reread=True))
|
|
flow.confirm(with_irregular=True)
|
|
flow.flush_culvert()
|
|
flow.save_sections()
|
|
|
|
|
|
def _b06_save(flow: Flow) -> None:
|
|
"""B06 [저장] 차례 — 관 → 구간값 → 상단측만 확정 → 계획선 → 구조물 → 횡단."""
|
|
flow.put_pipes(flow.b06_pipe_payload())
|
|
flow.flush_culvert()
|
|
flow.confirm(with_irregular=False)
|
|
flow.put_alignment()
|
|
flow.put_structures(flow.screen_structures(reread=False))
|
|
flow.save_sections()
|
|
|
|
|
|
@pytest.mark.parametrize("save", [_b05_save, _b06_save], ids=["B05", "B06"])
|
|
def test_저장_왕복_0_차이(api, save):
|
|
flow = Flow(api)
|
|
save(flow)
|
|
flow.check()
|
|
before = _snapshot(api.root)
|
|
_resave_unchanged(flow)
|
|
after = _snapshot(api.root)
|
|
flow.diffs += [f"다시 열어 그대로 저장 — {k} 가 바뀜" for k in before if before[k] != after[k]]
|
|
assert not flow.diffs, "\n".join(flow.diffs)
|