From af1b419d5df5cef8bb53b0682d0c2e2aafe25ecf Mon Sep 17 00:00:00 2001 From: umsangdon Date: Tue, 29 Sep 2026 10:00:03 +0900 Subject: [PATCH] =?UTF-8?q?test(B05=C2=B7B06):=20=EC=A0=80=EC=9E=A5=20?= =?UTF-8?q?=EC=99=95=EB=B3=B5=20=EC=8B=9C=ED=97=98=20=E2=80=94=20=EA=B3=A0?= =?UTF-8?q?=EB=A5=B8=20=EA=B0=92=EC=9D=B4=20=EC=A0=80=EC=9E=A5=20=C2=B7=20?= =?UTF-8?q?=EB=8B=A4=EC=8B=9C=20=EC=9D=BD=EA=B8=B0=20=EB=92=A4=20=EA=B7=B8?= =?UTF-8?q?=EB=8C=80=EB=A1=9C=EC=9D=B8=EC=A7=80=200=20=EC=B0=A8=EC=9D=B4?= =?UTF-8?q?=20=EB=8C=80=EC=A1=B0?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - test_b05_b06_save_roundtrip.py: 배정 storage 프로젝트(랩탑_보조) 사본에서 B05 [임시저장] · B06 [저장] 차례대로 서버 API 호출(관 → 계획선 → 구조물 → 확정 → 구간값 → 횡단) · pipe_points.json · structures.json · 종단 파일 · 응답 · DB 로 가는 횡단 patch 를 고른 값과 대조 · 다시 열어 그대로 저장해도 안 바뀌는지 - DB 는 읽기 전용 세션으로 읽기만 · 쓰는 함수는 받은 값을 적는 가짜 · 원본 · DB 안 바뀜 - helper_b05_save_payload.mjs: 저장 본문을 화면 코드 그대로(시설 되붙이기 · 구조물 측점 칸 · 구간값 · 이동 예약 흘려보내기) node 로 만듦 - 지금 코드에서 잡힌 것: 옛 관 이동 예약 적용 · 측점 0.1m 재읽기(관 · 구조물) · 새 관이 이웃 관 속성을 가져감 · 관 측점 격자 붙이기 · B06 저장에서 옮긴 관 측점 횡단 patch 버려짐 Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01AeeQRLknEpP4SFjVkjDWDs --- resources/tester/helper_b05_save_payload.mjs | 109 ++++ .../tester/helper_b05_save_payload_entry.ts | 9 + .../tester/test_b05_b06_save_roundtrip.py | 499 ++++++++++++++++++ 3 files changed, 617 insertions(+) create mode 100644 resources/tester/helper_b05_save_payload.mjs create mode 100644 resources/tester/helper_b05_save_payload_entry.ts create mode 100644 resources/tester/test_b05_b06_save_roundtrip.py diff --git a/resources/tester/helper_b05_save_payload.mjs b/resources/tester/helper_b05_save_payload.mjs new file mode 100644 index 00000000..363a51f3 --- /dev/null +++ b/resources/tester/helper_b05_save_payload.mjs @@ -0,0 +1,109 @@ +// 저장 왕복 시험용 — 화면 TS 를 node 로 묶어 저장 본문을 만든다. +// node helper_b05_save_payload.mjs <저장소 뿌리> · 표준입력 = 작업 목록(JSON) · 표준출력 끝 줄 = 결과 목록 +// {op:"attach", entries:[관 응답], points:[{chainage_m, source}]} → attach 결과(DetailPipeInput[]) +// {op:"station", chainage, interval} → 구조물 측점 칸에 쓰고 다시 읽은 값 +// {op:"culvert", server:[관 응답], pending:{측점:조각}, moves:{측점:새 측점}} → PUT 본문 points 또는 null +import fs from "node:fs"; +import os from "node:os"; +import path from "node:path"; +import { pathToFileURL } from "node:url"; + +const root = path.resolve(process.argv[2] ?? "."); +const { build } = await import( + pathToFileURL(path.join(root, "config/node_modules/vite/dist/node/index.js")).href +); + +// 화면 칸 · 세션만 흉내 — 측점 칸은 값 · 클래스 · 이벤트만 쓴다 +class Node { + constructor() { + this.value = ""; + this.children = []; + this.classList = { toggle() {}, add() {}, remove() {} }; + this.style = {}; + this.dataset = {}; + } + append(...kids) { + this.children.push(...kids); + } + appendChild(kid) { + this.children.push(kid); + return kid; + } + addEventListener() {} + setAttribute() {} +} +const session = new Map(); +globalThis.document = { createElement: () => new Node(), createTextNode: () => new Node() }; +globalThis.window = { + sessionStorage: { + getItem: (k) => (session.has(k) ? session.get(k) : null), + setItem: (k, v) => session.set(k, String(v)), + removeItem: (k) => session.delete(k), + }, + location: { origin: "http://localhost" }, + setTimeout: (...a) => setTimeout(...a), + clearTimeout: (id) => clearTimeout(id), +}; +globalThis.sessionStorage = globalThis.window.sessionStorage; +globalThis.localStorage = globalThis.window.sessionStorage; + +const outDir = fs.mkdtempSync(path.join(os.tmpdir(), "aislo-save-payload-")); +try { + await build({ + configFile: false, + logLevel: "silent", + root, + resolve: { + alias: { + "@ui": path.join(root, "ui_template"), + "@config": path.join(root, "config"), + "@util": path.join(root, "common_util"), + }, + }, + build: { + outDir, + minify: false, + target: "es2022", + lib: { + entry: path.join(root, "resources/tester/helper_b05_save_payload_entry.ts"), + formats: ["es"], + fileName: "entry", + }, + }, + }); + const mod = await import(pathToFileURL(path.join(outDir, "entry.js")).href); + const jobs = JSON.parse(fs.readFileSync(0, "utf-8")); + const results = []; + for (const job of jobs) { + if (job.op === "attach") { + const store = mod.createFacilityStore(); + store.replaceFromResponse(job.entries); + results.push(store.attach(job.points)); + } else if (job.op === "station") { + const fields = mod.stationFields("측점", () => job.interval); + fields.write(job.chainage, job.interval); + results.push(fields.read(job.interval, true)); + } else if (job.op === "culvert") { + session.clear(); + session.set("opt", JSON.stringify(job.pending ?? {})); + session.set("move", JSON.stringify(job.moves ?? {})); + let sent = null; + globalThis.fetch = async (_url, init = {}) => { + const body = init.method === "PUT" ? JSON.parse(init.body) : null; + if (body) sent = body.points; + const payload = body ? { pipe_points: body.points } : { pipe_points: job.server }; + return new Response(JSON.stringify(payload), { + status: 200, + headers: { "Content-Type": "application/json" }, + }); + }; + await mod.flushCulvertOptions("p", "opt", "move"); + results.push(sent); + } else { + throw new Error(`모르는 작업 ${job.op}`); + } + } + console.log(JSON.stringify(results)); +} finally { + fs.rmSync(outDir, { recursive: true, force: true }); +} diff --git a/resources/tester/helper_b05_save_payload_entry.ts b/resources/tester/helper_b05_save_payload_entry.ts new file mode 100644 index 00000000..e2d102bd --- /dev/null +++ b/resources/tester/helper_b05_save_payload_entry.ts @@ -0,0 +1,9 @@ +/* ============================================================================= + * helper_b05_save_payload_entry.ts + * 저장 왕복 시험(`test_b05_b06_save_roundtrip.py`)이 **화면 코드 그대로** 저장 본문을 만들 때 + * 묶는 진입점 — 관 시설 되붙이기 · 구조물 측점 칸(표시 → 다시 읽기) · 구간값 예약 흘려보내기. + * ========================================================================== */ + +export { createFacilityStore } from "../../B05_Profile/B05_Profile_UI_Drainage_Facility"; +export { stationFields } from "../../B05_Profile/B05_Profile_UI_Structures_Fields"; +export { flushCulvertOptions } from "../../B06_Section/B06_Section_Api_Culvert_Options"; diff --git a/resources/tester/test_b05_b06_save_roundtrip.py b/resources/tester/test_b05_b06_save_roundtrip.py new file mode 100644 index 00000000..4132a008 --- /dev/null +++ b/resources/tester/test_b05_b06_save_roundtrip.py @@ -0,0 +1,499 @@ +"""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 + ] + (attached,) = _screen([{"op": "attach", "entries": entries, "points": points}]) + 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)