diff --git a/resources/tester/helper_b05_routeedit_initial_apply.cjs b/resources/tester/helper_b05_routeedit_initial_apply.cjs new file mode 100644 index 00000000..b267032d --- /dev/null +++ b/resources/tester/helper_b05_routeedit_initial_apply.cjs @@ -0,0 +1,86 @@ +/* 초기노선 갈래 **저장 왕복**(계획서 20) 헬퍼 — 서버 답을 화면 코드 그대로 편다. + * + * 표준입력 JSON `{ jobs: [...] }` 를 받아 일마다 결과 한 줄을 돌려준다. + * - `apply` : `/route/initial` 답 → 편집값(`normalizeRouteInitial`) → [확인] 몸통 + * (`_RouteEdit_Apply` 의 vertices 만들기 · `_Initial.current()` 의 initial) + * - `reopen` : `/route/plan` 답 → 다시 연 노드(`flattenServerPlan`) · 현재 초기값(`serverBase`) + * · 캐시 초기값과 같은가(`sameBase` — [초기화]가 캐시 노드를 쓸지 서버에 물을지) + * + * ⚠ 확장자가 `.cjs` 인 까닭은 `helper_b05_cut_merged_loop.cjs` 머리에 적힌 것과 같다. + */ +const fs = require("fs"); +const path = require("path"); +const ts = require(path.join(__dirname, "..", "..", "config", "node_modules", "typescript")); + +const ROOT = path.join(__dirname, "..", ".."); +const cache = new Map(); + +function memoryStorage() { + const map = new Map(); + return { + getItem: (key) => (map.has(key) ? map.get(key) : null), + setItem: (key, value) => map.set(key, String(value)), + removeItem: (key) => map.delete(key), + key: (index) => [...map.keys()][index] ?? null, + get length() { + return map.size; + }, + }; +} +globalThis.window = { localStorage: memoryStorage(), sessionStorage: memoryStorage() }; +globalThis.fetch = () => Promise.resolve({ ok: false }); + +const STUBS = { "@ui/ui_template_palette": { themeColor: (_name, fallback) => fallback } }; + +function loadTs(file) { + const full = path.resolve(file); + if (cache.has(full)) return cache.get(full); + const js = ts.transpileModule(fs.readFileSync(full, "utf8"), { + compilerOptions: { module: ts.ModuleKind.CommonJS, target: ts.ScriptTarget.ES2020 }, + }).outputText; + const box = { exports: {} }; + cache.set(full, box.exports); + const localRequire = (request) => { + if (STUBS[request]) return STUBS[request]; + if (request.startsWith(".")) return loadTs(path.join(path.dirname(full), `${request}.ts`)); + return require(request); + }; + new Function("exports", "module", "require", js)(box.exports, box, localRequire); + cache.set(full, box.exports); + return box.exports; +} + +const model = loadTs(path.join(ROOT, "B05_Profile", "B05_Profile_UI_RouteEdit_Initial_Model.ts")); +const edits = loadTs(path.join(ROOT, "B05_Profile", "B05_Profile_UI_RouteEdit_Edits.ts")); + +const { jobs } = JSON.parse(fs.readFileSync(0, "utf8")); +const out = jobs.map((job) => { + if (job.kind === "apply") { + const choice = model.normalizeRouteInitial(job.response, job.logic, job.weights ?? null); + const { planned, curveOn, curveRadius } = choice.snapshot; + return { + choice, + // `_RouteEdit_Apply` [확인] 몸통 그대로. + body: { + vertices: planned.map(([x, y], index) => ({ + x, + y, + curve: curveOn[index] !== false, + radius_m: curveRadius[index] ?? null, + })), + initial: { mode: choice.logic, weights: choice.weights }, + }, + label: model.choiceLabel(choice.logic, choice.weights), + metrics: model.metricsLine(choice.metrics), + }; + } + const flat = edits.flattenServerPlan(job.plan.nodes ?? [], job.plan.curves ?? []); + const base = model.serverBase(job.plan.current_initial); + return { + planned: flat.planned, + base, + label: model.choiceLabel(base.logic, base.weights), + sameAsCache: job.cached ? model.sameBase(job.cached, base) : null, + }; +}); +process.stdout.write(JSON.stringify(out)); diff --git a/resources/tester/test_b05_routeedit_initial_roundtrip.py b/resources/tester/test_b05_routeedit_initial_roundtrip.py new file mode 100644 index 00000000..195423b6 --- /dev/null +++ b/resources/tester/test_b05_routeedit_initial_roundtrip.py @@ -0,0 +1,235 @@ +"""초기노선 갈래 저장 왕복 — [확인]에 실은 초기값이 다시 열기 뒤 그대로인가(계획서 20). + +환경 배정 storage 프로젝트(데스크탑_메인)의 **사본**에서 화면과 같은 차례로 서버 길을 부른다. +갈래(추종 · 토공 최소 · 기울기 순한 · 비중 50/25/25)마다 + ① `POST /route/initial` → 비교값이 다 있음 + ② 화면 코드로 [확인] 몸통을 만들어 `POST /route/replan` → 서버가 initial 을 적음 + ③ `GET /route/plan` → 현재 초기값 = 보낸 것 · 다시 연 노드 = 보낸 노드 + ④ [초기화] — 캐시가 있으면 그 노드, 없으면(다른 PC) 서버가 같은 갈래로 같은 노드를 냄 +화면 쪽 셈은 `helper_b05_routeedit_initial_apply.cjs` 가 화면 코드 그대로 돌린다. + +- 파일: 프로젝트 폴더를 임시 폴더로 복사(원본 입력 · 점군 · 큰 파일 제외) · + `STORAGE_BASE_DIR` 를 그리로. +- DB: 읽기 전용 세션으로 진짜 DB 를 읽기만 함. +- 재계산 체인(`_recompute`, 배수유역부터 약 90초 · 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 = "79da822d-921a-40f8-bb93-d8ffd90eeedf" # OWNERS.md — desktop-main `데스크탑_메인` +SOURCE = ROOT / "storage" / "1" / "3" / PROJECT_ID +_COPY_DIRS = ("B04_PreProcess/drainage", "B05_Profile", "B06_Section", "initial_snapshot") +_COPY_MAX_BYTES = 20 * 1024 * 1024 +HELPER = Path(__file__).with_name("helper_b05_routeedit_initial_apply.cjs") + +pytestmark = pytest.mark.skipif(not SOURCE.is_dir(), reason="배정 storage 프로젝트가 없음") + +LOGICS: list[tuple[str, dict[str, float] | None]] = [ + ("follow", None), + ("earthwork", None), + ("gentle", None), + ("weighted", {"expected": 50, "earthwork": 25, "grade": 25}), +] + + +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) + + +@pytest.fixture(scope="module") +def api(tmp_path_factory): + aiomysql = pytest.importorskip("aiomysql") + from fastapi import FastAPI + from fastapi.testclient import TestClient + + import B05_Profile.B05_Profile_Router_Replan as replan + 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_factory.mktemp("initial") / "storage" + project_root = storage_root / "1" / "3" / PROJECT_ID + _copy_project(project_root) + patch = pytest.MonkeyPatch() + patch.setattr(storage, "STORAGE_BASE_DIR", str(storage_root)) + recomputed: list[str] = [] + + async def fake_recompute(project_id, root, stored_path): + recomputed.append(str(root)) + return {"route_id": None, "total_length_m": None, "vertex_count": 0} + + patch.setattr(replan, "_recompute", fake_recompute) + + @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=4, + 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) + app.include_router(replan.router) + try: + with TestClient(app) as client: + client.root = project_root # type: ignore[attr-defined] + client.recomputed = recomputed # type: ignore[attr-defined] + yield client + finally: + patch.undo() + + +def _screen(jobs: list[dict[str, Any]]) -> list[Any]: + if shutil.which("node") is None: + pytest.skip("node 가 없음") + out = subprocess.run( + ["node", str(HELPER)], + input=json.dumps({"jobs": jobs}), + cwd=ROOT, + capture_output=True, + text=True, + encoding="utf-8", + timeout=120, + check=True, + ) + return json.loads(out.stdout) + + +def _initial(api, logic: str, weights: dict[str, float] | None) -> dict[str, Any]: + body: dict[str, Any] = {"logic": logic} + if weights: + body["weights"] = weights + response = api.post(f"/api/projects/{PROJECT_ID}/route/initial", json=body) + assert response.status_code == 200, response.text + return response.json() + + +def _close(a: list[list[float]], b: list[list[float]], tol: float = 1e-3) -> bool: + return len(a) == len(b) and all( + abs(p[0] - q[0]) <= tol and abs(p[1] - q[1]) <= tol for p, q in zip(a, b) + ) + + +@pytest.fixture(scope="module") +def rounds(api): + """갈래마다 ①~③ — 결과를 모아 두고 시험마다 나눠 본다.""" + out = [] + before = api.get(f"/api/projects/{PROJECT_ID}/route/plan").json() + for logic, weights in LOGICS: + answer = _initial(api, logic, weights) + [applied] = _screen( + [{"kind": "apply", "response": answer, "logic": logic, "weights": weights}] + ) + confirmed = api.post(f"/api/projects/{PROJECT_ID}/route/replan", json=applied["body"]) + assert confirmed.status_code == 200, confirmed.text + plan = api.get(f"/api/projects/{PROJECT_ID}/route/plan").json() + [reopened] = _screen([{"kind": "reopen", "plan": plan, "cached": applied["choice"]}]) + out.append( + { + "logic": logic, + "weights": weights, + "answer": answer, + "applied": applied, + "confirmed": confirmed.json(), + "plan": plan, + "reopened": reopened, + } + ) + return {"before": before, "rounds": out} + + +def test_원본은_처음_추종(rounds): + assert rounds["before"]["current_initial"]["logic"] == "follow" + + +def test_비교값이_응답에_다_있음(rounds): + for item in rounds["rounds"]: + metrics = item["answer"]["metrics"] + for key in ("length_m", "max_grade_pct", "avg_grade_pct", "cut_m3", "fill_m3"): + assert isinstance(metrics[key], (int, float)), (item["logic"], key) + assert isinstance(metrics["over_limit"]["total"], int) + assert isinstance(metrics["offset_avg_m"], (int, float)) + # 화면 한 줄에 빈 칸(—)이 없다. + assert "—" not in item["applied"]["metrics"], item["applied"]["metrics"] + + +def test_확인하면_서버가_초기값을_적고_다시_열면_그대로(rounds, api): + assert len(api.recomputed) == len(LOGICS) + for item in rounds["rounds"]: + sent = item["applied"]["body"]["initial"] + assert sent["mode"] == item["logic"] + assert item["confirmed"]["current_initial"]["logic"] == item["logic"] + current = item["plan"]["current_initial"] + assert current["logic"] == item["logic"] + if item["weights"]: + assert {k: round(v) for k, v in current["weights"].items()} == item["weights"] + reopened = item["reopened"] + assert reopened["base"]["logic"] == item["logic"] + assert reopened["label"] == item["applied"]["label"] + assert reopened["sameAsCache"] is True # [초기화]가 캐시 노드를 씀 + saved = json.loads( + (api.root / "B05_Profile" / "route" / "initial_choice.json").read_text("utf-8") + ) + assert saved["logic"] == "weighted" + + +def test_다시_연_노선은_확인한_노선(rounds): + for item in rounds["rounds"]: + sent = [[v["x"], v["y"]] for v in item["applied"]["body"]["vertices"]] + assert _close(item["reopened"]["planned"], sent), item["logic"] + + +def test_초기화는_마지막_초기값으로(rounds, api): + last = rounds["rounds"][-1] + cached = last["applied"]["choice"]["snapshot"]["planned"] + # 캐시가 있으면 그 노드 — 마지막 적용(비중 50/25/25). + assert _close(cached, [[v["x"], v["y"]] for v in last["applied"]["body"]["vertices"]]) + # 캐시가 없으면(다른 PC) 서버가 적어 둔 갈래 · 비중으로 다시 물음 → 같은 노드. + base = last["reopened"]["base"] + again = _initial(api, base["logic"], base["weights"]) + [redo] = _screen( + [{"kind": "apply", "response": again, "logic": base["logic"], "weights": base["weights"]}] + ) + assert _close(redo["choice"]["snapshot"]["planned"], cached) + # 다른 갈래들과는 다른 노선이다(되돌림이 실제로 갈래를 가림). + others = [r["applied"]["choice"]["snapshot"]["planned"] for r in rounds["rounds"][:-1]] + assert all(not _close(o, cached) for o in others)