322 lines
13 KiB
Python
322 lines
13 KiB
Python
"""PLAN 511-B B-5 (요구 320-4) — 노선 · 따라감 서버 몫 = 화면과 같은 TS(늘 떠 있는 Node).
|
|
|
|
못박는 것
|
|
- 노선 저장 폴리라인 파일(선 csv · 노드 csv · 곡선표 + 편집표) — Node `route.polyline` 길 =
|
|
파이썬 길 바이트 같음(거울 CASES · 위반 글 딱 반 R 12.25 · 12.75 · 겹친 점 · 짧은 칸 · 고정 ·
|
|
고정 길이 · 알려진 차 `KNOWN_ARC` = 직각 원호 점 ±1) ·
|
|
검증_sub3 저장 노선(노드 · 편집표) 다시 씀 = 저장된 곡선표 · 노드 그대로.
|
|
- 저장 좌표 맞춤(F1) — Node `facility.follow` + fill = 파이썬 짝
|
|
`_fill_structure_coordinates_py`(좌표 없음 · 측점 낳음 · 낡은 좌표 · 옛 좌표 지킴 · 자리 바뀜 ·
|
|
site · 읽을 때 · 노선 밖).
|
|
- 남은 측점 계획(F3) — Node `stations.plan` = 파이썬 `plan_stations`(322-3 CASES · 묶음마다).
|
|
- 공유 DB 안 씀 · storage 읽기만(검증_sub3 · 없으면 건너뜀) · 쓰기는 tmp.
|
|
|
|
./venv/Scripts/python.exe -m pytest resources/tester/test_511_b5_route_node.py -q
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import sys
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
|
|
import pytest
|
|
|
|
HERE = Path(__file__).resolve().parent
|
|
ROOT = HERE.parents[1]
|
|
for folder in (ROOT, HERE):
|
|
if str(folder) not in sys.path:
|
|
sys.path.insert(0, str(folder))
|
|
|
|
from B05_Profile import B05_Profile_Route_Polyline_Write as writer # noqa: E402
|
|
from common_util import common_util_route_follow as follow # noqa: E402
|
|
|
|
pytestmark = pytest.mark.skipif(
|
|
not (ROOT / "config" / "node_modules" / "vite").is_dir(), reason="프론트엔드 의존성 미설치"
|
|
)
|
|
|
|
SUB3 = (
|
|
ROOT / "storage" / "1" / "3" / "12953f5a-5ba1-42ad-a427-8ac857a58bc2" / "B05_Profile" / "route"
|
|
)
|
|
FIXTURE = HERE / "fixtures" / "golden_472"
|
|
NAMES = ("planned_route.csv", "planned_route_nodes.csv", "planned_route_curves.json")
|
|
# 알려진 차(브레인 승인 · 511-B B-5) — 교각 딱 5° 배수(합성 직각)면 원호 칸 수가 libm 끝자리로 갈림
|
|
# (파이썬 89.99999999999992° → 17 칸 · Node 90.0° → 18 칸) · 원호 점 ±1 · 실데이터 저장 노선 0 차
|
|
KNOWN_ARC = {
|
|
"직각 한 번 · R 자동",
|
|
"R 못박음 30m",
|
|
"R 이 접선 자리보다 커서 눌림",
|
|
"겹친 점(0.3m) 섞임 — 서버와 같이 버려야 함",
|
|
"딱 반 R(위반 글 짝수 쪽)",
|
|
}
|
|
|
|
|
|
def _files(folder: Path, monkeypatch, node: bool, args: dict) -> tuple[dict, dict]:
|
|
folder.mkdir(parents=True, exist_ok=True)
|
|
with monkeypatch.context() as patch:
|
|
if not node:
|
|
patch.setattr(writer, "node_polyline", lambda *a: None)
|
|
else: # Node 못 돌리면 파이썬으로 새는 것을 막음
|
|
made = writer.node_polyline
|
|
patch.setattr(writer, "node_polyline", lambda *a: made(*a) or pytest.fail("Node 없음"))
|
|
summary = writer._write_planned_polyline(folder / NAMES[0], simplify=False, **args)
|
|
return summary, {name: (folder / name).read_bytes() for name in NAMES}
|
|
|
|
|
|
def _cases() -> list[tuple[str, dict]]:
|
|
import test_route_polyline_browser_mirror as mirror
|
|
|
|
out = [
|
|
(
|
|
name,
|
|
{
|
|
"points": points,
|
|
"radius_m": mirror.MIN_RADIUS_M,
|
|
"curve_flags": flags,
|
|
"radii": radii,
|
|
},
|
|
)
|
|
for name, points, flags, radii in mirror.CASES
|
|
]
|
|
corner = [(0.0, 0.0), (100.0, 0.0), (100.0, 100.0), (200.0, 100.0)]
|
|
out += [
|
|
(
|
|
"딱 반 R(위반 글 짝수 쪽)",
|
|
{
|
|
"points": corner,
|
|
"radius_m": 15.0,
|
|
"curve_flags": [True] * 4,
|
|
"radii": [None, 12.25, 12.75, None],
|
|
},
|
|
),
|
|
(
|
|
"겹친 점 · 짧은 칸 · 고정 · 고정 길이",
|
|
{
|
|
"points": [(0.0, 0.0), (80.0, 0.0), (80.3, 0.1), (120.0, 60.0), (200.0, 60.0)],
|
|
"radius_m": 12.0,
|
|
"curve_flags": [True, True, True, False],
|
|
"radii": [None, 30.0],
|
|
"locks": [None, "radius", "arc", "bad", None],
|
|
"arcs": [None, None, 20.5, None],
|
|
},
|
|
),
|
|
(
|
|
"점 둘",
|
|
{
|
|
"points": [(0.0, 0.0), (10.0, 0.0)],
|
|
"radius_m": 12.0,
|
|
"curve_flags": [True, True],
|
|
"radii": [None, None],
|
|
},
|
|
),
|
|
]
|
|
return out
|
|
|
|
|
|
@pytest.mark.parametrize("name,args", _cases(), ids=[name for name, _ in _cases()])
|
|
def test_노선_폴리라인_Node_파이썬_바이트_같음(tmp_path, monkeypatch, name, args):
|
|
(py_sum, py), (node_sum, node) = (
|
|
_files(tmp_path / mode, monkeypatch, mode == "node", args) for mode in ("py", "node")
|
|
)
|
|
if name in KNOWN_ARC and node[NAMES[0]] != py[NAMES[0]]:
|
|
# 알려진 차 — 선 csv 원호 점 수만 ±1 · 노드 · 곡선표 · 편집표는 같음
|
|
assert abs(node_sum["vertices"] - py_sum["vertices"]) == 1, name
|
|
node, py = ({k: v for k, v in f.items() if k != NAMES[0]} for f in (node, py))
|
|
node_sum, py_sum = ({**s, "vertices": 0} for s in (node_sum, py_sum))
|
|
assert (node_sum, node) == (py_sum, py), name
|
|
|
|
|
|
def test_딱_반_R_위반_글은_파이썬_꼴(tmp_path, monkeypatch):
|
|
args = dict(_cases())["딱 반 R(위반 글 짝수 쪽)"]
|
|
_, files = _files(tmp_path, monkeypatch, True, args)
|
|
text = files[NAMES[2]].decode("utf-8")
|
|
assert "최소곡선반지름 미달(12.2 < 15.0m)" in text and "(12.8 < 15.0m)" in text
|
|
|
|
|
|
@pytest.mark.skipif(not (SUB3 / "planned_route_nodes.csv").is_file(), reason="검증_sub3 노선 없음")
|
|
def test_검증_sub3_저장_노선_다시_씀_같음(tmp_path, monkeypatch):
|
|
from common_util.common_util_route_geometry import read_planned_route_csv
|
|
|
|
table = json.loads((SUB3 / NAMES[2]).read_text(encoding="utf-8"))
|
|
nodes = [(v.x, v.y) for v in read_planned_route_csv(SUB3 / NAMES[1]).vertices]
|
|
edits = table["edits"]
|
|
args = {
|
|
"points": nodes,
|
|
"radius_m": table["min_radius_m"],
|
|
"curve_flags": [e["curve"] for e in edits],
|
|
"radii": [e["radius_m"] for e in edits],
|
|
"locks": [e["lock"] for e in edits],
|
|
"arcs": [e["arc_m"] for e in edits],
|
|
"station_snap": table.get("station_snap"),
|
|
"station_snap_step": table.get("station_snap_step"),
|
|
}
|
|
py = _files(tmp_path / "py", monkeypatch, False, args)
|
|
node = _files(tmp_path / "node", monkeypatch, True, args)
|
|
assert node == py
|
|
again = json.loads(node[1][NAMES[2]])
|
|
assert again["curves"] == table["curves"] and again["edits"] == edits
|
|
assert node[1][NAMES[1]] == (SUB3 / NAMES[1]).read_bytes()
|
|
|
|
|
|
def _line() -> list[list[float]]:
|
|
plan = json.loads((FIXTURE / "plan.json").read_text(encoding="utf-8"))
|
|
return plan["planned"]
|
|
|
|
|
|
def _items(line: list[list[float]]) -> tuple[list[SimpleNamespace], dict]:
|
|
on = follow.LineString(line)
|
|
|
|
def at(m: float) -> tuple[float, float]:
|
|
p = on.interpolate(m)
|
|
return round(p.x, 4), round(p.y, 4)
|
|
|
|
def s(sid, chainage, x=None, y=None, placement="point", **span):
|
|
return SimpleNamespace(
|
|
structure_id=sid,
|
|
placement=placement,
|
|
chainage_m=chainage,
|
|
start_m=span.get("start_m"),
|
|
end_m=span.get("end_m"),
|
|
x=x,
|
|
y=y,
|
|
)
|
|
|
|
hold = at(150.0)
|
|
items = [
|
|
s("bare", 120.0),
|
|
s("holds", 150.0, *hold),
|
|
s("stale", 210.0, *at(180.0)),
|
|
s("kept", 260.0, 1.0, 2.0),
|
|
s("moved", 300.0, 3.0, 4.0),
|
|
s("site", 340.0, *at(341.3), placement="site"),
|
|
s("span", 380.0, start_m=370.0, end_m=395.0),
|
|
s("none", None, 5.0, 6.0),
|
|
s("beyond", 99999.0),
|
|
s("prev-bare", 420.0),
|
|
]
|
|
previous = {
|
|
"kept": s("kept", 260.0, 208000.5, 367800.25),
|
|
"moved": s("moved", 290.0, 9.0, 9.0),
|
|
"prev-bare": s("prev-bare", 420.0, 7.0, 8.0),
|
|
}
|
|
return items, previous
|
|
|
|
|
|
@pytest.mark.parametrize("only_missing", [False, True])
|
|
def test_좌표_맞춤_Node_파이썬_같음(only_missing):
|
|
line = _line()
|
|
results = []
|
|
for fill in (follow.fill_structure_coordinates, follow._fill_structure_coordinates_py):
|
|
items, previous = _items(line)
|
|
count = fill(items, line, previous=previous, only_missing=only_missing)
|
|
results.append((count, [(item.x, item.y) for item in items]))
|
|
assert results[0] == results[1]
|
|
assert results[0][0] > 0
|
|
|
|
|
|
def test_좌표_맞춤_Node_길이_실제로_탐(monkeypatch):
|
|
called = []
|
|
real = follow.run_calc
|
|
monkeypatch.setattr(follow, "run_calc", lambda *a: called.append(a[0]) or real(*a))
|
|
items, previous = _items(_line())
|
|
follow.fill_structure_coordinates(items, _line(), previous=previous)
|
|
assert called == [["facility.follow"]]
|
|
|
|
|
|
def test_측점_계획_Node_파이썬_같음():
|
|
import test_322_3_plan_mirror as plan_mirror
|
|
|
|
from B05_Profile.B05_Profile_Engine_Sections_Node import node_plan_stations
|
|
from B05_Profile.B05_Profile_Structures_Schema import StructureInstance
|
|
from common_util.common_util_drainage_pipes import parse_pipe_points
|
|
from common_util.common_util_structure_station import plan_stations
|
|
|
|
groups = [
|
|
(
|
|
parse_pipe_points(case["pipes"]),
|
|
[StructureInstance.model_validate(s) for s in case["structures"]],
|
|
list(case.get("plus", ())),
|
|
)
|
|
for case in plan_mirror.CASES.values()
|
|
]
|
|
node = node_plan_stations(groups)
|
|
assert node is not None
|
|
assert node == [plan_stations(*group) for group in groups]
|
|
|
|
|
|
# ── 화면 — 계산 자리 route 「서버」 노선 = 화면 셈 노선(검증_sub3 · Playwright · 저장 안 함) ──
|
|
|
|
VERIFY = "12953f5a-5ba1-42ad-a427-8ac857a58bc2" # 검증_sub3(OWNERS.md)
|
|
|
|
JS_ROUTE = r"""async () => {
|
|
const urls = performance.getEntriesByType('resource').map((e) => e.name);
|
|
const page = await import(urls.find((n) => /design_pipeline_page\.ts/.test(n)));
|
|
const pid = localStorage.getItem('frd_current_project_id');
|
|
const pipe = page.pagePipeline(pid);
|
|
await page.reconcilePagePipeline(pid);
|
|
await pipe.settled();
|
|
const calls = (window.__aisloCalcServer?.calls ?? []).filter((c) => c.id === 'route.polyline');
|
|
const plain = (v) => JSON.parse(JSON.stringify(v ?? null));
|
|
return { route: plain(pipe.output('route')), stations: plain(pipe.output('stations')),
|
|
calls: calls.map((c) => ({ round: c.roundMs, node: c.nodeMs, same: c.same })),
|
|
runs: (window.__aisloPipeline?.runs ?? []).slice(-1).map((r) => r.outcome) };
|
|
}"""
|
|
|
|
|
|
@pytest.fixture(scope="module")
|
|
def screens():
|
|
from test_511_b1_calc_node_stages import _UP, BASE, JS_DRAFT, _started
|
|
|
|
if not BASE or _UP is None or _UP != _started(BASE.replace("5178", "8005")):
|
|
pytest.skip("이 폴더 서버 없음")
|
|
sync = pytest.importorskip("playwright.sync_api")
|
|
cred = json.loads((ROOT / "tmp" / "aislo_cred.json").read_text(encoding="utf-8"))
|
|
shots = {}
|
|
with sync.sync_playwright() as play:
|
|
chrome = play.chromium.launch(channel="msedge", headless=True)
|
|
try:
|
|
for mode in ("both", "server"):
|
|
context = chrome.new_context(viewport={"width": 1600, "height": 1000})
|
|
login = {k: cred[k] for k in ("email", "password")}
|
|
r = context.request.post(
|
|
f"{BASE}/api/auth/login/request", data=login, timeout=60000
|
|
)
|
|
assert r.ok, f"로그인 실패 {r.status}" # 비밀번호는 안 찍음
|
|
if mode == "server": # 판은 안 고침 — 이 창이 받는 판만 route = server
|
|
|
|
def board(route):
|
|
body = route.fetch().json()
|
|
body["sites"]["route"] = "server"
|
|
route.fulfill(json=body)
|
|
|
|
context.route("**/api/dashboard/calc-sites", board)
|
|
page = context.new_page()
|
|
page.add_init_script(
|
|
f"localStorage.setItem('frd_current_project_id', '{VERIFY}');"
|
|
" performance.setResourceTimingBufferSize(20000);"
|
|
)
|
|
page.goto(f"{BASE}/#/b05-profile", wait_until="domcontentloaded")
|
|
page.wait_for_function("() => !!window.__aisloPipeline", timeout=90000)
|
|
page.wait_for_timeout(9000)
|
|
shots[mode] = {"saved": page.evaluate(JS_ROUTE)}
|
|
page.evaluate(JS_DRAFT) # 교각점 하나 옮김(세션 초안 · 창과 함께 버림)
|
|
page.wait_for_timeout(3000)
|
|
shots[mode]["draft"] = page.evaluate(JS_ROUTE)
|
|
context.close()
|
|
finally:
|
|
chrome.close()
|
|
(ROOT / "tmp" / "511b5_screens.json").write_text(json.dumps(shots), encoding="utf-8")
|
|
return shots
|
|
|
|
|
|
@pytest.mark.parametrize("case", ["saved", "draft"])
|
|
def test_화면_서버_노선_화면_셈과_같음(screens, case):
|
|
both, server = screens["both"][case], screens["server"][case]
|
|
assert both["runs"] == server["runs"] == ["done"]
|
|
assert len(both["route"]["line"]) > 10
|
|
assert both["route"] == server["route"] and both["stations"] == server["stations"]
|
|
assert both["calls"] == []
|
|
# 저장 노선(초안 없음) = 서버 안 부름 · 초안 = server 판만 서버 노선
|
|
assert bool(server["calls"]) is (case == "draft")
|