test(B05·B06): 저장 왕복 시험 — 고른 값이 저장 · 다시 읽기 뒤 그대로인지 0 차이 대조
- 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 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01AeeQRLknEpP4SFjVkjDWDs
This commit is contained in:
Claude Opus 5.5
@@ -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 });
|
||||
}
|
||||
@@ -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";
|
||||
@@ -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)
|
||||
Reference in New Issue
Block a user