"""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 from B05_Profile.B05_Profile_Engine_Sections import cross_filename 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})") # 소수 측점 횡단(PLAN 20장) — 그 자리 횡단 파일이 소수 2자리 이름으로 따로 있음 name = cross_filename(c) if not (root / "B06_Section" / "cross_sections" / name).is_file(): d.append(f"관 측점 {c} — 횡단 파일 {name} 없음") 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"]) # 두 페이지 모두 저장 한 줄기(`A00_Common/design/design_save.ts` `DESIGN_SAVE_ORDER`)를 탄다 — # 관 → 구조물 → 계획고 → 암거 옵션(구간값) → 오르막(상단측 확정) → 횡단(계획서 24-2a). def _b05_save(flow: Flow) -> None: """B05 [임시저장] — 오르막 차례에 관 측점(비정규 측점)도 함께 확정.""" flow.put_pipes(flow.b05_pipe_payload()) flow.put_structures(flow.screen_structures(reread=True)) flow.put_alignment() flow.flush_culvert() flow.confirm(with_irregular=True) flow.save_sections() def _b06_save(flow: Flow) -> None: """B06 [저장] — 오르막 차례는 상단측만 확정.""" flow.put_pipes(flow.b06_pipe_payload()) flow.put_structures(flow.screen_structures(reread=False)) flow.put_alignment() flow.flush_culvert() flow.confirm(with_irregular=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)