test(B05): 초기노선 갈래 저장 왕복 — [확인]에 실은 초기값이 다시 열기 뒤 그대로인지 사본에서 대조
- 데스크탑_메인 storage 사본 · DB 읽기 전용 · 재계산 체인은 성공만 돌려주는 가짜 - 추종 · 토공 최소 · 기울기 순한 · 비중 50/25/25 마다 /route/initial → 화면 코드로 [확인] 몸통 → /route/replan → /route/plan - 현재 초기값 = 보낸 것 · 다시 연 노드 = 보낸 노드 · [초기화] = 마지막 초기값(캐시 없으면 서버가 같은 노드) · 비교값 다 있음 - 화면 셈은 helper_b05_routeedit_initial_apply.cjs 가 화면 코드 그대로 Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_015sCeZz8z8YnbovsKgn7HK7
This commit is contained in:
Claude Opus 5.5
@@ -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));
|
||||
@@ -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)
|
||||
Reference in New Issue
Block a user