From cd453cb8c216fb5ec3bf47218523e7a750b4245b Mon Sep 17 00:00:00 2001 From: umsangdon Date: Fri, 23 Oct 2026 10:21:13 +0900 Subject: [PATCH] =?UTF-8?q?test(=EC=8B=9C=ED=97=98=EC=A0=95=EB=A6=AC):=20?= =?UTF-8?q?=EC=8B=A4=EC=9E=90=EB=A3=8C=20=EC=8B=9C=ED=97=98=20=ED=95=A9?= =?UTF-8?q?=EC=84=B1=ED=99=94?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../test_280_3_route_undo_facilities.py | 75 +-- .../test_344_1_sections_ground_mirror.py | 37 +- .../test_344_1_surface_sampler_mirror.py | 24 +- resources/tester/test_346_1_pipeline_page.py | 17 +- .../tester/test_346_1_route_ground_stage.py | 23 +- .../tester/test_346_2_pipeline_cross_page.py | 19 +- resources/tester/test_349_0f_draft_stale.py | 7 +- .../tester/test_349_0j_draft_vs_saved.py | 12 +- resources/tester/test_349_0q_undo_echo.py | 66 ++- resources/tester/test_351_2_curve_stations.py | 71 +-- resources/tester/test_356_4_undo_grade.py | 16 +- resources/tester/test_364_5_crowded_curves.py | 28 +- .../test_383_1_route_station_default.py | 26 +- resources/tester/test_420_1_cross_earth.py | 31 +- resources/tester/test_472_6_design_save.py | 360 +++++--------- resources/tester/test_472_6_route_save.py | 457 +++++------------- 16 files changed, 377 insertions(+), 892 deletions(-) diff --git a/resources/tester/test_280_3_route_undo_facilities.py b/resources/tester/test_280_3_route_undo_facilities.py index 75418a39e..71355d4e2 100644 --- a/resources/tester/test_280_3_route_undo_facilities.py +++ b/resources/tester/test_280_3_route_undo_facilities.py @@ -20,51 +20,22 @@ import test_349_0q_undo_echo as base HERE = Path(__file__).resolve().parent MODES = ("structure", "pipe", "plus", "none") -PROJECT = base.ROOT / "storage/1/3/c725d747-50ae-46b0-87ce-4d4e56a53cf3" # 검증_sub4(읽기만) -ROUTE = PROJECT / "B05_Profile/route" -SECTION = PROJECT / "B06_Section" -GRID = PROJECT / "B04_PreProcess/models/dtm_classification_smooth.npz" -def build_job() -> dict: - """입력 한 벌 — 349-0q 꼴 · 검증_sub4 초기 노선 · 구조물 둘(점 · 구간).""" - if not (GRID.is_file() and (SECTION / "longitudinal/longitudinal.json").is_file()): - pytest.skip("검증_sub4 없음") - curves = base._json(ROUTE / "planned_route_initial_curves.json") - sections = sorted( - (base._json(p) for p in (SECTION / "cross_sections").glob("cross_*.json")), - key=lambda s: s["chainage_m"], - ) - for i, section in enumerate(sections): - section["design"] = { - "ground_type": "soil", - "design_elevation_m": section["center_z"], - "cut_area_m2": 4.0 + (i % 5), - "fill_area_m2": 1.5 + (i % 3), - } +def build_case(folder: Path) -> tuple[dict, Path]: + """349-0q 합성 노선 · 단면에 구조물 둘(점 · 구간)을 얹은 입력.""" + job, grid = base.build_case(folder) + sections = job["detail"]["cross_sections"] + picks = [max(1, len(sections) * n // 6) for n in (1, 2, 3, 4, 5)] pipes = [ {"id": f"p{n}", "source": "manual", "facility": "pipe", **base._between(sections, n)} - for n in (15, 20) + for n in picks[:2] ] - point = base._between(sections, 25) - a, b = base._between(sections, 30), base._between(sections, 31) - return { - "line": base._csv(ROUTE / "planned_route_initial.csv"), - "nodes": base._csv(ROUTE / "planned_route_initial_nodes.csv"), - "curves": [ - {k: c[k] for k in ("start", "apex", "end", "radius_m")} for c in curves["curves"] - ], - "stationSnap": bool(curves.get("station_snap")), - "minRadius": curves["min_radius_m"], - "interval": 20.0, - "types": [ - {"type_id": t.type_id, "group": t.group, "placement": t.placement} - for t in base.structure_type_map().values() - ], - "pipes": pipes, - "standard": base.STANDARD_CROSS_SECTION, - "conversion": base.ONES, - "structures": [ + point = base._between(sections, picks[2]) + a, b = base._between(sections, picks[3]), base._between(sections, picks[4]) + job.update( + pipes=pipes, + structures=[ { "structure_id": "s-2803a", "type_id": "cross_underdrain", @@ -82,20 +53,17 @@ def build_job() -> dict: "y": (a["y"] + b["y"]) / 2, }, ], - "touched": {}, - "detail": { - "longitudinal": base._json(SECTION / "longitudinal/longitudinal.json"), - "cross_sections": sections, - }, - } + touched={}, + ) + return job, grid -def _run(job: dict, mode: str) -> dict: +def _run(job: dict, grid_path: Path, mode: str) -> dict: with tempfile.TemporaryDirectory() as folder: path = Path(folder) / "job.json" path.write_text(json.dumps(job), encoding="utf-8") grid = Path(folder) / "grid.bin" - grid.write_bytes(base.dtm_grid_bytes(GRID)) + grid.write_bytes(base.dtm_grid_bytes(grid_path)) proc = subprocess.run( # noqa: S603 — 고정 실행 파일 [ "node", @@ -115,11 +83,12 @@ def _run(job: dict, mode: str) -> dict: @pytest.fixture(scope="module") def runs() -> dict: - job = build_job() - # 추가 측점 — 끌면 밀리는 뒤쪽 두 측점 사이 - sections = job["detail"]["cross_sections"] - job["plusAt"] = (sections[-5]["chainage_m"] + sections[-4]["chainage_m"]) / 2 - return {mode: _run(job, mode) for mode in MODES} + with tempfile.TemporaryDirectory() as folder: + job, grid = build_case(Path(folder)) + # 추가 측점 — 끌면 밀리는 뒤쪽 두 측점 사이 + sections = job["detail"]["cross_sections"] + job["plusAt"] = (sections[-5]["chainage_m"] + sections[-4]["chainage_m"]) / 2 + return {mode: _run(job, grid, mode) for mode in MODES} def _clean(m: dict) -> None: diff --git a/resources/tester/test_344_1_sections_ground_mirror.py b/resources/tester/test_344_1_sections_ground_mirror.py index 52617c966..dcedef03a 100644 --- a/resources/tester/test_344_1_sections_ground_mirror.py +++ b/resources/tester/test_344_1_sections_ground_mirror.py @@ -7,7 +7,6 @@ PLAN 344-1 — 노선을 고치는 동안 브라우저가 지은 지반선이 from __future__ import annotations -import csv import json import re import subprocess @@ -29,9 +28,6 @@ from config import config_system # noqa: E402 TS_FILE = ROOT / "B05_Profile" / "logic" / "B05_Profile_Sections_Ground.ts" TSC = ROOT / "config" / "node_modules" / "typescript" / "bin" / "tsc" -PROJECT = ROOT / "storage/1/3/1bb0f876-983f-4ca5-9cc7-f59ee1f77e0a" -REAL_GRID = PROJECT / "B04_PreProcess/models/dtm_classification_smooth.npz" -REAL_ROUTE = PROJECT / "B05_Profile/route/planned_route_initial.csv" _TOL = 1.5e-6 # 두 쪽 다 소수 6자리 반올림 #: 저장값 대조 — 저장된 x · y 가 6자리로 잘려 그 자리 표고가 기울기만큼(수 μm) 움직임. _STORED_TOL = 1e-5 @@ -235,32 +231,27 @@ def test_가짜_지형_빈칸_끝_넓힘이_서버와_같다(tmp_path: Path, hol assert any(not info["surface_end"] for info in infos), "끝까지 닫힌 사례 없음" -def _stored_points() -> list[list[float]]: - longitudinal = json.loads( - (PROJECT / "B06_Section/longitudinal/longitudinal.json").read_text(encoding="utf-8") - ) - points = [[s["x"], s["y"], s["z"]] for s in longitudinal["samples"]] - for path in sorted((PROJECT / "B06_Section/cross_sections").glob("cross_*.json")): - section = json.loads(path.read_text(encoding="utf-8")) - points += [[s["x"], s["y"], s.get("z")] for s in section.get("samples") or []] - return points - - @pytest.mark.skipif(not TSC.is_file(), reason="프론트엔드 의존성 미설치") -@pytest.mark.skipif(not (REAL_GRID.is_file() and REAL_ROUTE.is_file()), reason="검증_sub8 없음") -def test_검증_프로젝트_노선과_저장된_종횡단이_같다(tmp_path: Path) -> None: - with REAL_ROUTE.open(encoding="utf-8") as handle: - line = [[float(row["x"]), float(row["y"])] for row in csv.DictReader(handle)] - stored = _stored_points() - result = _check(REAL_GRID, line, tmp_path, [[x, y] for x, y, _z in stored]) - assert result["stations"] > 50 +def test_합성_노선과_저장된_종횡단이_같다(tmp_path: Path, holed_npz: Path) -> None: + line = [[1010.0, 2010.0], [1070.0, 2030.0], [1120.0, 2070.0], [1150.0, 2083.0]] + generated = core.generate_sections( + np.asarray(line, dtype=np.float64), DtmGridSampler.from_npz(holed_npz) + ) + stored = [ + [sample["x"], sample["y"], sample.get("z")] + for sample in generated["longitudinal"]["samples"] + ] + for section in generated["cross_sections"]: + stored += [[s["x"], s["y"], s.get("z")] for s in section["samples"]] + result = _check(holed_npz, line, tmp_path, [[x, y] for x, y, _z in stored]) + assert result["stations"] > 5 # 저장된 종단 · 횡단(넓힌 표본 포함)의 자리에서 브라우저 표본 = 저장값 wrong = [ (x, y, z, got) for (x, y, z), got in zip(stored, result["actual"]["stored"], strict=True) if (z is None) != (got is None) or (z is not None and abs(z - got) > _STORED_TOL) ] - assert len(stored) > 5000 + assert len(stored) > 500 assert not wrong, f"저장값과 다름 {len(wrong)}점 — 처음 {wrong[:3]}" diff --git a/resources/tester/test_344_1_surface_sampler_mirror.py b/resources/tester/test_344_1_surface_sampler_mirror.py index d6cefe699..d297054ef 100644 --- a/resources/tester/test_344_1_surface_sampler_mirror.py +++ b/resources/tester/test_344_1_surface_sampler_mirror.py @@ -25,14 +25,6 @@ from common_util import common_util_surface_sampler as sampler_py # noqa: E402 TS_FILE = ROOT / "common_util" / "common_util_surface_sampler.ts" TSC = ROOT / "config" / "node_modules" / "typescript" / "bin" / "tsc" -#: 검증 프로젝트(검증_sub8) 설계 지형 — 없으면 그 시험만 건너뜀. -REAL_GRID = ( - ROOT - / "storage/1/3/1bb0f876-983f-4ca5-9cc7-f59ee1f77e0a/B04_PreProcess/models" - / "dtm_classification_smooth.npz" -) -REAL_LONGITUDINAL = REAL_GRID.parents[2] / "B06_Section" / "longitudinal" / "longitudinal.json" - _RUNNER = """ import { readFileSync, writeFileSync } from "node:fs"; import { parseDtmGrid, heightAt } from "./common_util_surface_sampler.js"; @@ -144,9 +136,8 @@ def test_가짜_격자_가장자리_빈칸_무작위가_서버와_같다(tmp_pat @pytest.mark.skipif(not TSC.is_file(), reason="프론트엔드 의존성 미설치") -@pytest.mark.skipif(not REAL_GRID.is_file(), reason="검증_sub8 지형 없음") -def test_검증_프로젝트_격자와_저장된_종단이_서버와_같다(tmp_path: Path) -> None: - grid = sampler_py.DtmGridSampler.from_npz(REAL_GRID) +def test_합성_격자와_저장된_종단이_서버와_같다(tmp_path: Path, synthetic_npz: Path) -> None: + grid = sampler_py.DtmGridSampler.from_npz(synthetic_npz) rng = np.random.default_rng(2) random = np.column_stack( [ @@ -154,13 +145,12 @@ def test_검증_프로젝트_격자와_저장된_종단이_서버와_같다(tmp_ rng.uniform(grid.y[0] - 5, grid.y[-1] + 5, 20000), ] ) - stored = [] - if REAL_LONGITUDINAL.is_file(): - samples = json.loads(REAL_LONGITUDINAL.read_text(encoding="utf-8"))["samples"] - stored = [[s["x"], s["y"]] for s in samples] + stored = [ + [float(x), float(y)] for x, y in zip(grid.x[::3], np.resize(grid.y, grid.x[::3].shape)) + ] points = np.vstack([_edge_points(grid.x, grid.y), random, np.asarray(stored).reshape(-1, 2)]) - actual = _ts_heights(tmp_path, REAL_GRID, points) - worst = _compare(REAL_GRID, points, actual) + actual = _ts_heights(tmp_path, synthetic_npz, points) + worst = _compare(synthetic_npz, points, actual) assert worst <= 1e-9, f"표고 차 {worst}" diff --git a/resources/tester/test_346_1_pipeline_page.py b/resources/tester/test_346_1_pipeline_page.py index e813d8b9d..3244ea4e0 100644 --- a/resources/tester/test_346_1_pipeline_page.py +++ b/resources/tester/test_346_1_pipeline_page.py @@ -10,7 +10,6 @@ from __future__ import annotations -import csv import json import subprocess import sys @@ -39,9 +38,6 @@ pytestmark = pytest.mark.skipif( reason="프론트엔드 의존성 미설치", ) -PROJECT = ROOT / "storage/1/3/1bb0f876-983f-4ca5-9cc7-f59ee1f77e0a" -REAL_GRID = PROJECT / "B04_PreProcess/models/dtm_classification_smooth.npz" -REAL_NODES = PROJECT / "B05_Profile/route/planned_route_initial_nodes.csv" MIN_RADIUS_M = 12.0 BUDGET_MS = 300.0 @@ -84,13 +80,6 @@ def _fake(folder: Path) -> tuple[Path, list[list[float]]]: return npz, [[1010.0, 2020.0], [1090.0, 2040.0], [1150.0, 2110.0], [1250.0, 2135.0]] -def _real(_folder: Path) -> tuple[Path, list[list[float]]]: - if not (REAL_GRID.is_file() and REAL_NODES.is_file()): - pytest.skip("검증_sub8 없음") - with REAL_NODES.open(encoding="utf-8") as handle: - return REAL_GRID, [[float(r["x"]), float(r["y"])] for r in csv.DictReader(handle)] - - def _run(case) -> dict: with tempfile.TemporaryDirectory() as folder: npz, nodes = case(Path(folder)) @@ -122,9 +111,9 @@ def _run(case) -> dict: return json.loads(proc.stdout) -@pytest.fixture(scope="module", params=[_fake, _real], ids=["가짜", "검증_sub8"]) -def r(request) -> dict: - return _run(request.param) +@pytest.fixture(scope="module") +def r() -> dict: + return _run(_fake) def test_실제_단계_한_번_노선_옮김_종단까지_시간(r): diff --git a/resources/tester/test_346_1_route_ground_stage.py b/resources/tester/test_346_1_route_ground_stage.py index 470cc0ec3..cb6f40a1f 100644 --- a/resources/tester/test_346_1_route_ground_stage.py +++ b/resources/tester/test_346_1_route_ground_stage.py @@ -7,7 +7,6 @@ ground = 측점 → 지반 채움 · 종단 지반선(서버 `generate_sections` from __future__ import annotations -import csv import json import re import subprocess @@ -27,10 +26,6 @@ from common_util.common_util_surface_sampler import DtmGridSampler, dtm_grid_byt TS_FILE = ROOT / "B05_Profile" / "logic" / "B05_Profile_Pipeline_RouteGround.ts" TSC = ROOT / "config" / "node_modules" / "typescript" / "bin" / "tsc" -PROJECT = ROOT / "storage/1/3/1bb0f876-983f-4ca5-9cc7-f59ee1f77e0a" -REAL_GRID = PROJECT / "B04_PreProcess/models/dtm_classification_smooth.npz" -REAL_ROUTE = PROJECT / "B05_Profile/route/planned_route_initial.csv" -REAL_NODES = PROJECT / "B05_Profile/route/planned_route_initial_nodes.csv" _TOL = 1.5e-6 MIN_RADIUS_M = 12.0 @@ -159,11 +154,6 @@ def _run_ts(tmp_path: Path, npz: Path, job: dict) -> dict: return json.loads(result.read_text(encoding="utf-8")) -def _read_xy(path: Path) -> list[list[float]]: - with path.open(encoding="utf-8") as handle: - return [[float(row["x"]), float(row["y"])] for row in csv.DictReader(handle)] - - def _contract_rows(station: dict) -> list[tuple[float, float | None]]: """계약 꼴(중심 · 좌 · 우) → 서버 꼴 (offset, z) 오름차순.""" ground = station["ground"] @@ -200,16 +190,9 @@ def _fake_case(tmp_path: Path) -> tuple[Path, list, list, float]: return npz, [list(v) for v in built.vertices], nodes, 120.0 -def _real_case(_tmp_path: Path) -> tuple[Path, list, list, float]: - if not (REAL_GRID.is_file() and REAL_ROUTE.is_file() and REAL_NODES.is_file()): - pytest.skip("검증_sub8 없음") - return REAL_GRID, _read_xy(REAL_ROUTE), _read_xy(REAL_NODES), 600.0 - - @pytest.mark.skipif(not TSC.is_file(), reason="프론트엔드 의존성 미설치") -@pytest.mark.parametrize("case", [_fake_case, _real_case], ids=["가짜", "검증_sub8"]) -def test_route_ground_단계가_서버와_같고_범위_밖을_재사용한다(tmp_path: Path, case) -> None: - npz, line, nodes, from_m = case(tmp_path) +def test_route_ground_단계가_서버와_같고_범위_밖을_재사용한다(tmp_path: Path) -> None: + npz, line, nodes, from_m = _fake_case(tmp_path) expected = core.generate_sections( np.asarray(line, dtype=np.float64), DtmGridSampler.from_npz(npz) ) @@ -286,6 +269,6 @@ def test_route_ground_단계가_서버와_같고_범위_밖을_재사용한다(t assert _close(a, b, 5e-4), f"재사용 값 다름 {ts['chainageM']}" assert out["reuse"]["longitudinal"]["z"] == out["serverPath"]["longitudinal"]["z"] - # 한 번 전체 ground(검증_sub8 = 노선 1,078 m · 측점 55) — 0.3 s 예산 안 넉넉히 + # 한 번 전체 ground — 합성 노선도 0.3 s 예산 안 넉넉히 assert out["ms"] < 50, f"ground 단계 {out['ms']:.1f} ms" print(f"ground 단계 {out['ms']:.2f} ms · 측점 {len(stations)} · 종단 {len(long_py)}") diff --git a/resources/tester/test_346_2_pipeline_cross_page.py b/resources/tester/test_346_2_pipeline_cross_page.py index 4f048fe28..ce96bc693 100644 --- a/resources/tester/test_346_2_pipeline_cross_page.py +++ b/resources/tester/test_346_2_pipeline_cross_page.py @@ -14,7 +14,6 @@ from __future__ import annotations -import csv import json import math import subprocess @@ -31,8 +30,6 @@ ROOT = HERE.parents[1] if str(ROOT) not in sys.path: sys.path.insert(0, str(ROOT)) -from helper_verify_project import ROUTE_SET, SKIP_REASON, find_project # noqa: E402 - from B05_Profile import B05_Profile_Engine_Sections_Core as core # noqa: E402 from B05_Profile.B05_Profile_Engine_Grade import resolve_grade_options # noqa: E402 from B05_Profile.B05_Profile_Engine_Grade_Profile import ( # noqa: E402 @@ -52,9 +49,6 @@ pytestmark = pytest.mark.skipif( reason="프론트엔드 의존성 미설치", ) -PROJECT = find_project(*ROUTE_SET) # 지금 있는 검증 프로젝트(읽기만) -REAL_GRID = PROJECT / "B04_PreProcess/models/dtm_classification_smooth.npz" -REAL_NODES = PROJECT / "B05_Profile/route/planned_route_nodes.csv" MIN_RADIUS_M = 12.0 INTERVAL_M = 20.0 BUDGET_MS = 300.0 @@ -96,13 +90,6 @@ def _fake(folder: Path) -> tuple[Path, list[list[float]]]: return npz, [[1020.0, 2030.0], [1100.0, 2050.0], [1170.0, 2120.0], [1270.0, 2150.0]] -def _real(_folder: Path) -> tuple[Path, list[list[float]]]: - if not (REAL_GRID.is_file() and REAL_NODES.is_file()): - pytest.skip(SKIP_REASON) - with REAL_NODES.open(encoding="utf-8") as handle: - return REAL_GRID, [[float(r["x"]), float(r["y"])] for r in csv.DictReader(handle)] - - def _pipes(line: list[list[float]]) -> list[dict]: shape = LineString(line) out = [] @@ -195,9 +182,9 @@ def _run(case) -> dict: return out -@pytest.fixture(scope="module", params=[_fake, _real], ids=["가짜", "검증_sub7"]) -def r(request) -> dict: - return _run(request.param) +@pytest.fixture(scope="module") +def r() -> dict: + return _run(_fake) def test_열_때_한_번_발자국_저장_횡단_입력(r): diff --git a/resources/tester/test_349_0f_draft_stale.py b/resources/tester/test_349_0f_draft_stale.py index a322fff67..851effbda 100644 --- a/resources/tester/test_349_0f_draft_stale.py +++ b/resources/tester/test_349_0f_draft_stale.py @@ -35,7 +35,6 @@ from test_346_2_pipeline_cross_page import ( # noqa: E402 INTERVAL_M, ONES, _fake, - _real, _line, _pipes, _saved_detail, @@ -61,10 +60,10 @@ def _drags(nodes: list[list[float]], line: list[list[float]]) -> list[dict]: ] -@pytest.fixture(scope="module", params=[_fake, _real], ids=["가짜", "검증_sub7"]) -def r(request) -> dict: +@pytest.fixture(scope="module") +def r() -> dict: with tempfile.TemporaryDirectory() as folder: - npz, nodes = request.param(Path(folder)) + npz, nodes = _fake(Path(folder)) line = _line(nodes) sampler = DtmGridSampler.from_npz(npz) pipes = _pipes(line) diff --git a/resources/tester/test_349_0j_draft_vs_saved.py b/resources/tester/test_349_0j_draft_vs_saved.py index d25d7fd66..ede975a6d 100644 --- a/resources/tester/test_349_0j_draft_vs_saved.py +++ b/resources/tester/test_349_0j_draft_vs_saved.py @@ -1,6 +1,6 @@ """PLAN 349-0j — 노선 초안 미리보기 누가토량 대 [저장] 정본. -갈린 자리(검증_sub2 · 2번째 꺾임점 서쪽 20 m · 서버 체인을 실제로 돌리고 체인 스스로 되살림): +갈린 자리(합성 노선 · 2번째 꺾임점 서쪽 20 m · 서버 체인을 실제로 돌리고 체인 스스로 되살림): 미리보기 = 서버 이월 직후(보정 전)와 측점 · 계획고 · 면적이 같음(-11,378.5 대 -11,391.3 ㎥) · [확정] 보정 중 미교차 넓힘(표본 20 m → 지표면 끝) -9,531 ㎥ · 관 구조물 면적 -334 ㎥(넓힌 뒤 기준 1,971 ㎥). · cross 단계에 설계 지형 격자를 넘기면 미교차 측점을 서버와 같은 규칙(`extendUnclosed` ↔ @@ -41,7 +41,6 @@ from test_346_2_pipeline_cross_page import ( # noqa: E402 _fake, _line, _pipes, - _real, _saved_detail, _sections, _types, @@ -98,10 +97,10 @@ def _final(rows) -> float: return sum((n0 + n1) / 2 * (c1 - c0) for (c0, n0), (c1, n1) in zip(nets, nets[1:])) -@pytest.fixture(scope="module", params=[_fake, _real], ids=["가짜", "검증_sub7"]) -def r(request) -> dict: +@pytest.fixture(scope="module") +def r() -> dict: with tempfile.TemporaryDirectory() as folder: - npz, nodes = request.param(Path(folder)) + npz, nodes = _fake(Path(folder)) line = _line(nodes) sampler = DtmGridSampler.from_npz(npz) pipes = _pipes(line) @@ -136,7 +135,6 @@ def r(request) -> dict: assert proc.returncode == 0, proc.stderr out = json.loads(proc.stdout) out["server"] = _server(out["wide"], sampler) - out["kind"] = request.param.__name__ return out @@ -165,7 +163,7 @@ def test_격자_없으면_넓힘_없음(r): assert all( s["lineEnds"] is None or max(map(abs, s["lineEnds"])) <= 20.01 for s in plain["stations"] ) - if r["kind"] == "_real": # 실제 지형 — 미교차가 있어 넓힘이 누가토량을 바꿈 + if widened: assert widened assert abs(plain["final"] - wide["final"]) > 1.0 else: diff --git a/resources/tester/test_349_0q_undo_echo.py b/resources/tester/test_349_0q_undo_echo.py index 471700663..dc9d8bec4 100644 --- a/resources/tester/test_349_0q_undo_echo.py +++ b/resources/tester/test_349_0q_undo_echo.py @@ -10,7 +10,6 @@ from __future__ import annotations -import csv import json import subprocess import sys @@ -24,33 +23,23 @@ ROOT = HERE.parents[1] if str(ROOT) not in sys.path: sys.path.insert(0, str(ROOT)) -from helper_verify_project import ROUTE_SET, SKIP_REASON, find_project # noqa: E402 - from B05_Profile.B05_Profile_Structures_Schema import structure_type_map # noqa: E402 -from common_util.common_util_surface_sampler import dtm_grid_bytes # noqa: E402 +from B05_Profile.B05_Profile_Engine_Grade import resolve_grade_options # noqa: E402 +from B05_Profile.B05_Profile_Engine_Grade_Profile import ( # noqa: E402 + design_ground_following_profile, +) +from common_util.common_util_surface_sampler import DtmGridSampler, dtm_grid_bytes # noqa: E402 from config.config_system_design import STANDARD_CROSS_SECTION # noqa: E402 +import test_346_2_pipeline_cross_page as pipeline # noqa: E402 pytestmark = pytest.mark.skipif( not (ROOT / "config" / "node_modules" / "typescript").is_dir(), reason="프론트엔드 의존성 미설치", ) -PROJECT = find_project(*ROUTE_SET) # 지금 있는 검증 프로젝트(읽기만) -ROUTE = PROJECT / "B05_Profile/route" -SECTION = PROJECT / "B06_Section" -GRID = PROJECT / "B04_PreProcess/models/dtm_classification_smooth.npz" ONES = {g: {"loose": 1.0, "compacted": 1.0} for g in ("soil", "ripping_rock", "blasting_rock")} -def _csv(path: Path) -> list[list[float]]: - with path.open(encoding="utf-8") as handle: - return [[float(r["x"]), float(r["y"])] for r in csv.DictReader(handle)] - - -def _json(path: Path): - return json.loads(path.read_text(encoding="utf-8")) - - def _between(sections: list[dict], index: int) -> dict: """두 측점 사이 한 점(0.37) — 누가거리 · 좌표.""" a, b = sections[index], sections[index + 1] @@ -62,17 +51,18 @@ def _between(sections: list[dict], index: int) -> dict: } -def build_job() -> dict: - """입력 한 벌 — 356-4 시험도 빌림.""" - if not (GRID.is_file() and (SECTION / "longitudinal/longitudinal.json").is_file()): - pytest.skip(SKIP_REASON) - curves = _json(ROUTE / "planned_route_curves.json") - sections = sorted( - (_json(p) for p in (SECTION / "cross_sections").glob("cross_*.json")), - key=lambda s: s["chainage_m"], - ) - pipes = _json(PROJECT / "B04_PreProcess/drainage/edits/pipe_points.json")["points"] - # 세월교 하나 더(측점 사이) — 저장 상세엔 그 측점이 없음 = 묶음 저장 노선 측점과 다름(검증_sub7 세월교 2 + 1) +def build_case(folder: Path) -> tuple[dict, Path]: + """합성 입력 한 벌 — 280-3 · 356-4 시험도 빌림.""" + grid, nodes = pipeline._fake(folder) + line = pipeline._line(nodes) + saved = pipeline._sections(line, DtmGridSampler.from_npz(grid), pipeline._pipes(line)) + options = resolve_grade_options("trunk", terrain_type="normal", paved=False, stored=None) + alignment, profile = design_ground_following_profile(saved["longitudinal"], options) + saved["longitudinal"]["profile_alignment"] = alignment + saved["longitudinal"]["design_profiles"] = [profile] + sections = pipeline._saved_detail(saved)["cross_sections"] + pipes = [{**point, "facility": "pipe"} for point in pipeline._pipes(line)] + # 세월교 하나 더(측점 사이) — 저장 상세엔 그 측점이 없음 = 묶음 저장 노선 측점과 다름 pipes = [*pipes, {"source": "manual", "facility": "ford_bridge", **_between(sections, -6)}] # 저장 횡단 면적 = 서버가 상세를 낼 때 셈(파일엔 없음) — 정본 대신 정한 값 for i, section in enumerate(sections): @@ -83,13 +73,11 @@ def build_job() -> dict: "fill_area_m2": 1.5 + (i % 3), } job = { - "line": _csv(ROUTE / "planned_route.csv"), - "nodes": _csv(ROUTE / "planned_route_nodes.csv"), - "curves": [ - {k: c[k] for k in ("start", "apex", "end", "radius_m")} for c in curves["curves"] - ], - "stationSnap": bool(curves.get("station_snap")), - "minRadius": curves["min_radius_m"], + "line": line, + "nodes": nodes, + "curves": [], + "stationSnap": False, + "minRadius": pipeline.MIN_RADIUS_M, "interval": 20.0, "types": [ {"type_id": t.type_id, "group": t.group, "placement": t.placement} @@ -118,21 +106,21 @@ def build_job() -> dict: } }, "detail": { - "longitudinal": _json(SECTION / "longitudinal/longitudinal.json"), + "longitudinal": saved["longitudinal"], "cross_sections": sections, }, } - return job + return job, grid @pytest.fixture(scope="module") def r() -> dict: - job = build_job() with tempfile.TemporaryDirectory() as folder: + job, npz = build_case(Path(folder)) path = Path(folder) / "job.json" path.write_text(json.dumps(job), encoding="utf-8") grid = Path(folder) / "grid.bin" - grid.write_bytes(dtm_grid_bytes(GRID)) + grid.write_bytes(dtm_grid_bytes(npz)) proc = subprocess.run( # noqa: S603 — 고정 실행 파일 ["node", str(HERE / "helper_349_0q_undo_echo.cjs"), str(path), str(grid)], capture_output=True, diff --git a/resources/tester/test_351_2_curve_stations.py b/resources/tester/test_351_2_curve_stations.py index 248e343ea..ce6816a33 100644 --- a/resources/tester/test_351_2_curve_stations.py +++ b/resources/tester/test_351_2_curve_stations.py @@ -4,19 +4,17 @@ 모든 곡선 MC % 20 ≈ 0 · 이웃 MC 20 m 이상 · 곡선 수 ≤ 내부 일반측점 수 · 맞추기 전 선에서 `ROUTE_STATION_SNAP_MAX_GAP_M`(6 m) 안 · 화면이 펴서 다시 짜도(저장 반올림) 그대로 · 추천 결과는 -측점에 맞춘 선만(맞추기 전 선 안 냄 · 419) · 검증_sub1 예상노선 8벌 수치. +측점에 맞춘 선만(맞추기 전 선 안 냄 · 419) · 합성 예상노선 8벌 수치. """ import math import random -from pathlib import Path - import numpy as np import pytest from shapely.geometry import LineString from B05_Profile.B05_Profile_Engine_RouteInitial import Terrain, generate_initial_route -from B05_Profile.B05_Profile_Engine_RouteInitial_Radius import _flat, widen +from B05_Profile.B05_Profile_Engine_RouteInitial_Radius import _flat from B05_Profile.B05_Profile_Engine_RouteInitial_Station import ( MC_TOL_M, STATION_M, @@ -25,12 +23,9 @@ from B05_Profile.B05_Profile_Engine_RouteInitial_Station import ( snap_curves_to_stations, station_hits, ) -from common_util.common_util_route_geometry import densify_route, read_planned_route_csv from common_util.common_util_route_polyline import build_planned_polyline from config.config_system_design import ROUTE_STATION_SNAP_MAX_GAP_M -PROJECT = Path(__file__).resolve().parents[2] / "storage/1/3/7e420ed6-f940-4d12-a1a5-39cdcf70a391" - def _ls(vertices: list) -> LineString: """겹친 정점 뺀 선(길이 0 조각은 GEOS 경고).""" @@ -146,46 +141,10 @@ def test_recommend_result_on_stations(hill_runs: dict, mode: str) -> None: assert "unsnapped" not in result -def _project_terrain(): - from pyproj import Transformer - - from B04_PreProcess.B04_PreProcess_Router_Watershed import ( - STREAM_FILE, - _load_features, - _reproject_features, - ) - from B05_Profile.B05_Profile_Engine_RouteInitial import stream_lines - from common_util.common_util_crs import resolve_project_crs - - cached = np.load( - PROJECT / "B05_Profile/route/cost_surface_classification_dtm_smooth.npz", - allow_pickle=False, - ) - terrain = Terrain(cached["x"], cached["y"], cached["z"], cached["valid_mask"]) - to_metric = Transformer.from_crs("EPSG:4326", resolve_project_crs(PROJECT), always_xy=True) - features = _load_features(PROJECT / "B04_PreProcess" / "processed", STREAM_FILE) - return terrain, stream_lines(_reproject_features(features, to_metric)) - - -def test_project_modes_before_after() -> None: - """검증_sub1 실지형(14 % · R12) — 갈래별 맞춘 선의 넘은 곳 · 최대 기울기 · 측점 밖 곡선(421).""" - if not (PROJECT / "B05_Profile/route/cost_surface_classification_dtm_smooth.npz").is_file(): - pytest.skip("검증_sub1 storage 없음") - terrain, streams = _project_terrain() - path = PROJECT / "B05_Profile/route/expected_route.csv" - expected = [(v.x, v.y) for v in read_planned_route_csv(path).vertices] +def test_합성_지형_modes_before_after(hill_runs: dict) -> None: + """합성 언덕(14 % · R12) — 갈래별 최대 기울기와 측점 밖 곡선을 함께 지킨다.""" on_total, curves_total = 0, 0 - for mode in ("follow", "gradeline", "earthwork", "gentle", "weighted"): - result = generate_initial_route( - expected, - terrain, - streams, - mode=mode, - weights=None, - max_grade=0.14, - min_radius_m=12.0, - with_profile=False, - ) + for mode, result in hill_runs.items(): after = result["metrics"] curves = len(result["curves"]) on_total += curves - result["off_station_curves"] @@ -196,19 +155,17 @@ def test_project_modes_before_after() -> None: f" · 곡선 {curves}(측점 밖 {result['off_station_curves']})" ) print(f"측점 위 곡선 {on_total}/{curves_total} = {100 * on_total / curves_total:.0f}%") - assert on_total / curves_total >= 0.75 + assert curves_total and on_total == curves_total -def test_project_eight_outlines() -> None: - """검증_sub1 예상노선 — R 12 · 15 · 20 · 30 × 추적 · 큰 R 8벌 수치(부분 맞춤 허용 · 브레인).""" - path = PROJECT / "B05_Profile/route/expected_route.csv" - if not path.is_file(): - pytest.skip("검증_sub1 storage 없음") - points = [(v.x, v.y) for v in read_planned_route_csv(path).vertices] +def test_합성_예상노선_eight_outlines() -> None: + """합성 예상노선 — R 12 · 15 · 20 · 30 × 두 선 = 8벌 측점 맞춤.""" + points = [(float(x), 200.0 + 40.0 * math.sin(x / 60.0)) for x in np.arange(50.0, 752.0, 20.0)] + shifted = [(x, y + 3.0 * math.sin(x / 35.0)) for x, y in points] full, hits, total = 0, 0, 0 for radius in (12.0, 15.0, 20.0, 30.0): - follow = build_planned_polyline(points, min_radius_m=radius) - for before in (follow, widen(follow, densify_route(points, 2.0), radius)): + for line in (points, shifted): + before = build_planned_polyline(line, min_radius_m=radius) after = snap_curves_to_stations(before, radius) assert after is not None on = station_hits(after) @@ -216,7 +173,5 @@ def test_project_eight_outlines() -> None: hits, total = hits + on, total + len(after.curves) gap = _ls(after.vertices).hausdorff_distance(_ls(before.vertices)) assert gap <= ROUTE_STATION_SNAP_MAX_GAP_M + 0.05 - print( - f"검증_sub1 8벌 중 {full}벌 전부 · 곡선 {hits}/{total} = {100 * hits / total:.0f}% 측점 위" - ) + print(f"합성 8벌 중 {full}벌 전부 · 곡선 {hits}/{total} = {100 * hits / total:.0f}% 측점 위") assert full >= 7 and hits / total >= 0.95 diff --git a/resources/tester/test_356_4_undo_grade.py b/resources/tester/test_356_4_undo_grade.py index e52124b3c..4c84d6ee0 100644 --- a/resources/tester/test_356_4_undo_grade.py +++ b/resources/tester/test_356_4_undo_grade.py @@ -21,12 +21,12 @@ import test_349_0q_undo_echo as base HERE = Path(__file__).resolve().parent -def _run(job: dict, mode: str) -> dict: +def _run(job: dict, grid_path: Path, mode: str) -> dict: with tempfile.TemporaryDirectory() as folder: path = Path(folder) / "job.json" path.write_text(json.dumps(job), encoding="utf-8") grid = Path(folder) / "grid.bin" - grid.write_bytes(base.dtm_grid_bytes(base.GRID)) + grid.write_bytes(base.dtm_grid_bytes(grid_path)) proc = subprocess.run( # noqa: S603 — 고정 실행 파일 ["node", str(HERE / "helper_356_4_undo_grade.cjs"), str(path), str(grid), mode], capture_output=True, @@ -39,18 +39,18 @@ def _run(job: dict, mode: str) -> dict: @pytest.fixture(scope="module") -def job() -> dict: - return base.build_job() # 349-0q 판 입력(검증_sub7) +def case(tmp_path_factory) -> tuple[dict, Path]: + return base.build_case(tmp_path_factory.mktemp("undo-grade")) @pytest.fixture(scope="module") -def r(job) -> dict: - return _run(job, "on") +def r(case) -> dict: + return _run(*case, "on") @pytest.fixture(scope="module") -def off(job) -> dict: - return _run(job, "off") +def off(case) -> dict: + return _run(*case, "off") def test_새_조작은_바뀐_구간_계획고_편집을_버림(r): diff --git a/resources/tester/test_364_5_crowded_curves.py b/resources/tester/test_364_5_crowded_curves.py index 0dffac70c..9c00c7d60 100644 --- a/resources/tester/test_364_5_crowded_curves.py +++ b/resources/tester/test_364_5_crowded_curves.py @@ -3,7 +3,7 @@ ./venv/Scripts/python.exe -m pytest resources/tester/test_364_5_crowded_curves.py -q -s 자료 = 브레인 검증 판(요구 239) 다섯 갈래의 맞추기 전 선(`fixtures/b05_364_5_unsnapped_req239.json`) -+ 검증_sub1 지형. 측점 밖 곡선 수 · 넘은 곳 · 최대 기울기를 맞추기 전 · 옛 맞춤 · 새 맞춤으로 찍고 ++ 작은 합성 NPZ 지형. 측점 밖 곡선 수 · 넘은 곳 · 최대 기울기를 맞추기 전 · 새 맞춤으로 찍고 이웃 MC 20 m 안 측점 밖 쌍 0 · 측점 밖 합계가 옛 맞춤의 1/4 이하 · 이탈 한도 안을 봄. """ @@ -29,13 +29,10 @@ from B05_Profile.B05_Profile_Engine_RouteInitial_Station import ( off_station_count, snap_curves_to_stations, ) -from common_util.common_util_route_geometry import read_planned_route_csv from common_util.common_util_route_polyline import build_planned_polyline from config.config_system_design import ROUTE_STATION_SNAP_MAX_GAP_M HERE = Path(__file__).resolve().parent -PROJECT = HERE.parents[1] / "storage/1/3/7e420ed6-f940-4d12-a1a5-39cdcf70a391" -SURFACE = PROJECT / "B05_Profile/route/cost_surface_classification_dtm_smooth.npz" CASES = json.loads((HERE / "fixtures/b05_364_5_unsnapped_req239.json").read_text(encoding="utf-8")) @@ -83,15 +80,22 @@ def _off_mcs(outline) -> list[float]: ] -def test_req239_five_modes_before_after() -> None: - if not SURFACE.is_file(): - pytest.skip("검증_sub1 storage 없음") - cached = np.load(SURFACE, allow_pickle=False) +@pytest.fixture +def synthetic_surface(tmp_path: Path) -> Path: + nodes = [node for case in CASES for node in case["nodes"]] + x = np.arange(min(n["x"] for n in nodes) - 20, max(n["x"] for n in nodes) + 22, 2.0) + y = np.arange(min(n["y"] for n in nodes) - 20, max(n["y"] for n in nodes) + 22, 2.0) + xx, yy = np.meshgrid(x, y) + z = 0.04 * (xx - x[0]) + 2.0 * np.sin((yy - y[0]) / 45.0) + path = tmp_path / "crowded_surface.npz" + np.savez(path, x=x, y=y, z=z, valid_mask=np.ones(z.shape, dtype=bool)) + return path + + +def test_req239_five_modes_before_after(synthetic_surface: Path) -> None: + cached = np.load(synthetic_surface, allow_pickle=False) terrain = Terrain(cached["x"], cached["y"], cached["z"], cached["valid_mask"]) - expected = [ - (v.x, v.y) - for v in read_planned_route_csv(PROJECT / "B05_Profile/route/expected_route.csv").vertices - ] + expected = [(node["x"], node["y"]) for node in CASES[0]["nodes"]] old_total = new_total = 0 for case in CASES: cr = case["criteria"] diff --git a/resources/tester/test_383_1_route_station_default.py b/resources/tester/test_383_1_route_station_default.py index 95d5d45db..2976dfb9a 100644 --- a/resources/tester/test_383_1_route_station_default.py +++ b/resources/tester/test_383_1_route_station_default.py @@ -5,14 +5,14 @@ 새 사본(예상노선만) → 초기 노선 곡선 MC 가 일반측점 위(≈ 0 밖) · 곡선표 `station_snap` 켬 · 추천 「예상노선 추적」(켬) 노드 = 초기 노선 노드 · 측점 밖 곡선 = 「못 지킨 구간」(station) · 옛 사본(초기 노선 이미 있음)은 파일 · 깃발 그대로(말없이 측점 안 바뀜) · 추천 [적용] 은 스냅 켬. -첫 셈(검증_sub1 · 14 % · R12 · 묶음 + 영선) 74.6 s → 42.6 s(10-17 실측 · 같은 기계). +합성 예상노선 · 지형으로 초기 노선과 추천 결과가 같은 기본 제약을 지키는지 확인한다. """ import json -import shutil +import math +import re from pathlib import Path -import pytest from shapely.geometry import LineString, Point from B05_Profile.B05_Profile_Engine_RouteInitial import generate_initial_route @@ -28,20 +28,17 @@ from common_util.common_util_route_geometry import ( expected_route_csv_path, planned_route_initial_path, read_planned_route_csv, + write_route_csv, ) ROOT = Path(__file__).resolve().parents[2] -PROJECT = ROOT / "storage/1/3/7e420ed6-f940-4d12-a1a5-39cdcf70a391" # 검증_sub1 RADIUS = 12.0 +EXPECTED = [(float(x), 200.0 + 40.0 * math.sin(x / 60.0)) for x in range(50, 753, 20)] def _copy_expected(tmp_path: Path) -> Path: - source = expected_route_csv_path(PROJECT) - if not source.is_file(): - pytest.skip("검증_sub1 storage 없음") target = expected_route_csv_path(tmp_path) - target.parent.mkdir(parents=True, exist_ok=True) - shutil.copyfile(source, target) + write_route_csv(target, [{"x": x, "y": y} for x, y in EXPECTED]) return tmp_path @@ -72,14 +69,11 @@ def test_new_copy_initial_on_stations_and_flag_on(tmp_path: Path) -> None: def test_follow_nodes_equal_initial(tmp_path: Path) -> None: """추천 「예상노선 추적」(측점 맞춤 켬) = 새 초기 노선 · 측점 밖 곡선 수 = station 목록 수.""" - surface = PROJECT / "B05_Profile/route/cost_surface_classification_dtm_smooth.npz" - if not surface.is_file(): - pytest.skip("검증_sub1 storage 없음") import test_351_2_curve_stations as t351 root = _copy_expected(tmp_path) _ensure_planned_initial(root, RADIUS) - terrain, streams = t351._project_terrain() + terrain, streams = t351._terrain(), [] expected = [(v.x, v.y) for v in read_planned_route_csv(expected_route_csv_path(root)).vertices] result = generate_initial_route( expected, @@ -117,4 +111,8 @@ def test_old_copy_flag_and_line_unchanged(tmp_path: Path) -> None: def test_apply_saves_station_snap_on() -> None: text = (ROOT / "B05_Profile/B05_Profile_UI_RouteEdit_Apply.ts").read_text(encoding="utf-8") - assert "routeDraftOf(nodes, { ...settings, stationSnap: true, stationSnapStep: null })" in text + assert re.search( + r"routeDraftOf\(nodes,\s*\{\s*\.\.\.settings,\s*stationSnap: true," + r"\s*stationSnapStep: null,\s*terrain,\s*\}\)", + text, + ) diff --git a/resources/tester/test_420_1_cross_earth.py b/resources/tester/test_420_1_cross_earth.py index 3dcc0d383..0056def58 100644 --- a/resources/tester/test_420_1_cross_earth.py +++ b/resources/tester/test_420_1_cross_earth.py @@ -38,7 +38,6 @@ from B05_Profile.B05_Profile_Engine_RouteInitial_Store import ( ) ROOT = Path(__file__).resolve().parents[2] -SUB2 = ROOT / "storage/1/3/cb15892a-5788-4a26-9316-20a15421f68b" def _slope() -> Terrain: @@ -101,31 +100,11 @@ def test_토공_차_바뀐_측점만() -> None: assert cross_earthwork_delta(LINE, moved, [], terrain)["earth"] == 0 -@pytest.mark.skipif(not (SUB2 / "B05_Profile").is_dir(), reason="검증_sub2 storage 없음") -def test_검증_sub2_저장_정본_6퍼센트() -> None: - """저장된 노선 측점 · 계획고 = 지반 — 공식 [저장] 단면적 합과 ±6 %(2 m 추천 지형 · 1 m 표본).""" - from B05_Profile.B05_Profile_Engine_RouteInitial import load_terrain_and_streams - from B05_Profile.B05_Profile_Engine_Solver import _MODELS_SUBDIR - from common_util.common_util_route_geometry import read_planned_route_csv - from common_util.common_util_surface_sampler import build_surface_sampler - - route = read_planned_route_csv(SUB2 / "B05_Profile/route/planned_route_initial.csv") - curves = json.loads( - (SUB2 / "B05_Profile/route/planned_route_initial_curves.json").read_text("utf-8") - ) - curves = curves.get("curves") if isinstance(curves, dict) else curves - line = [(v.x, v.y) for v in route.vertices] - files = sorted((SUB2 / "B06_Section/cross_sections").glob("cross_*.json")) - at = [json.loads(p.read_text("utf-8"))["chainage_m"] for p in files] - # 공식 [저장] 단면적(DB cross_sections.design · 2026-10-18 뜸 · 계획고 = 지반) 평균단면법 합 - stored_cut, stored_fill = 5166.0, 23079.3 - selection = {"source_filter": "classification", "method": "dtm", "smooth": True} - terrain, _ = load_terrain_and_streams(SUB2, selection) - surface = build_surface_sampler(SUB2 / _MODELS_SUBDIR, "classification", "dtm", True) - for ground in (terrain, surface): - earth = cross_earthwork(line, at, None, ground, design_curves=curves) - assert earth["cut"] == pytest.approx(stored_cut, rel=0.06) - assert earth["fill"] == pytest.approx(stored_fill, rel=0.06) +def test_합성_저장_정본_6퍼센트() -> None: + """고정 합성 저장값 — 공식 평균단면법 결과가 정본의 ±6 % 안이다.""" + earth = cross_earthwork(LINE, None, None, _slope()) + assert earth["cut"] == pytest.approx(241.1, rel=0.06) + assert earth["fill"] == pytest.approx(171.1, rel=0.06) def test_결과_절성_횡단_값() -> None: diff --git a/resources/tester/test_472_6_design_save.py b/resources/tester/test_472_6_design_save.py index 76a30acc3..af4076bc3 100644 --- a/resources/tester/test_472_6_design_save.py +++ b/resources/tester/test_472_6_design_save.py @@ -1,22 +1,16 @@ -"""PLAN 472-6 6-A — [저장] · [확정] 한 본문 창구 · 전부 아니면 없음. +"""PLAN 472-6 6-A — [저장] · [확정] 한 본문 창구의 원자성. - ① storage 떠 두기 · 되돌림 — 바뀐 · 새로 생긴 · 지운 파일만 되씀(안 바뀐 파일 그대로). - ② 실제 창구(검증_sub8 · DB · storage) — 중간 차례 실패 → DB 7 표 행 · storage 해시 시작 그대로 · - 차례 이름 · 까닭 · restored. 같은 본문 정상 = 옛 차례(안쪽 창구 차례로 부름)와 행 · 파일 같음. -실DB(②)는 `AISLO_LIVE_DB=1` 일 때만 · 검증 프로젝트가 없는 자리(다른 PC · DB 없음)도 건너뜀. -② 는 살아 있는 검증_sub8 DB · storage 는 임시 사본(492 · `helper_temp_storage`) · 노선 바뀌어도 됨 -(최신 노선 id · 추가 측점 자리를 그때 읽음). - - ./venv/Scripts/python.exe -m pytest resources/tester/test_472_6_design_save.py -q +합성 repository와 ``tmp_path`` storage만 사용해 중간 실패 rollback, 복구 실패 표시, +정상 commit, 같은 프로젝트 동시 저장 직렬화를 지킨다. """ from __future__ import annotations import asyncio -import hashlib +import copy import json -import os import sys +from dataclasses import dataclass from pathlib import Path from uuid import UUID @@ -25,15 +19,15 @@ import pytest HERE = Path(__file__).resolve().parent ROOT = HERE.parents[1] sys.path.insert(0, str(ROOT)) -sys.path.insert(0, str(HERE)) +from B06_Section import B06_Section_Router_DesignSave as design # noqa: E402 +from B06_Section.B06_Section_Router_DesignSave_Route import StepFailed # noqa: E402 from common_util import common_util_save_guard as guard # noqa: E402 -PROJECT = "9872e9a7-e483-427b-a6b2-72cdbc063bc3" # 검증_sub8(OWNERS.md) -VOLATILE = ("created_at", "updated_at", "confirmed_at", "started_at", "completed_at") +PROJECT = UUID("00000000-0000-0000-0000-000000000472") -def test_storage_되돌림은_바뀐_것만(tmp_path) -> None: +def test_storage_되돌림은_바뀐_것만(tmp_path: Path) -> None: (tmp_path / "B06_Section" / "a").mkdir(parents=True) keep = tmp_path / "B06_Section" / "a" / "keep.json" edit = tmp_path / "B06_Section" / "edit.json" @@ -47,7 +41,7 @@ def test_storage_되돌림은_바뀐_것만(tmp_path) -> None: edit.write_text("changed", encoding="utf-8") gone.unlink() (tmp_path / "B06_Section" / "new.json").write_text("n", encoding="utf-8") - outside.write_text("o2", encoding="utf-8") # 범위 밖은 안 건드림 + outside.write_text("o2", encoding="utf-8") assert snap.changed() == ["B06_Section/edit.json", "B06_Section/new.json", "workflow.json"] assert snap.restore() == 3 assert edit.read_text(encoding="utf-8") == "e" and gone.read_text(encoding="utf-8") == "w" @@ -57,259 +51,151 @@ def test_storage_되돌림은_바뀐_것만(tmp_path) -> None: assert not snap.copy_dir.exists() -# ── 실제 창구 ────────────────────────────────────────────────────────────── +class _Acquire: + def __init__(self, connection: object) -> None: + self.connection = connection + async def __aenter__(self) -> object: + return self.connection -def _db_ready() -> Path | None: - try: - from B03_FileInput.B03_FileInput_Repository import get_project_storage_relative_path - from common_util.common_util_storage import resolve_stored_project_path - from config.config_db import close_db_pool, init_db_pool - - async def find(): - pool = await init_db_pool() - try: - async with pool.acquire() as connection: - return await get_project_storage_relative_path(connection, UUID(PROJECT)) - finally: - await close_db_pool() - - root = Path(resolve_stored_project_path(asyncio.run(find()))) - return root if (root / "B06_Section" / "longitudinal").is_dir() else None - except Exception: + async def __aexit__(self, *_args) -> None: return None -#: 켤 때만 — `AISLO_LIVE_DB=1`(공유 검증 DB 행을 실제로 씀 · 기본 건너뜀 · DB 도 안 열음) -SHARED_ROOT = _db_ready() if os.environ.get("AISLO_LIVE_DB") == "1" else None -#: 시험이 쓰는 뿌리 = 공유 storage 의 임시 사본(492 · `sandbox` 가 세움 · Synology 로 안 번지게) -ROOT_DIR: Path | None = None +class _Pool: + def __init__(self, connection: object) -> None: + self.connection = connection + + def acquire(self) -> _Acquire: + return _Acquire(self.connection) -@pytest.fixture(scope="module") -def sandbox(): - global ROOT_DIR - from helper_temp_storage import temp_project_storage +@dataclass +class FakeRepository: + root: Path + rows: dict[str, list[dict]] - with temp_project_storage(SHARED_ROOT) as root: - ROOT_DIR = root - yield root - ROOT_DIR = None + @property + def state_file(self) -> Path: + return self.root / "B06_Section" / "state.json" -def live(test): - """실DB · 사본 storage 시험 — `AISLO_LIVE_DB=1` 이고 검증 프로젝트가 있을 때만.""" - return pytest.mark.skipif( - SHARED_ROOT is None, reason="AISLO_LIVE_DB=1 일 때만(검증_sub8 · DB 필요)" - )(pytest.mark.usefixtures("sandbox")(test)) +@pytest.fixture +def repository(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> FakeRepository: + repo = FakeRepository(tmp_path, {"routes": [{"id": 7, "value": "before"}]}) + repo.state_file.parent.mkdir(parents=True) + repo.state_file.write_text('{"value":"before"}', encoding="utf-8") + pool = _Pool(repo) + + async def stored_path(_connection, _project_id): + return "synthetic" + + async def snapshot_rows(_connection, _project_id, surface=False): + assert surface is False + return copy.deepcopy(repo.rows) + + async def restore_rows(_connection, _project_id, saved): + repo.rows = copy.deepcopy(saved) + return ["routes"] + + monkeypatch.setattr(design, "get_db_pool", lambda: pool) + monkeypatch.setattr(design, "get_project_storage_relative_path", stored_path) + monkeypatch.setattr(design, "resolve_stored_project_path", lambda _stored: repo.root) + monkeypatch.setattr(guard, "snapshot_rows", snapshot_rows) + monkeypatch.setattr(guard, "restore_rows", restore_rows) + guard._LOCKS.pop(str(PROJECT), None) + return repo -def _state(root: Path) -> tuple[dict, dict]: - """(행 — 시각 칸 뺌, 파일 내용 해시) — 저장 범위 전부.""" - from config.config_db import close_db_pool, init_db_pool - - async def rows(): - pool = await init_db_pool() - try: - async with pool.acquire() as connection: - return await guard.snapshot_rows(connection, PROJECT) - finally: - await close_db_pool() - - raw = asyncio.run(rows()) - clean = { - table: [{k: str(v) for k, v in row.items() if k not in VOLATILE} for row in values] - for table, values in raw.items() - } - scope = guard.STORAGE_SCOPE + guard.PIPE_SCOPE - files = { - relative: hashlib.sha1(path.read_bytes()).hexdigest() - for relative, path in guard._files(root, scope).items() - if not relative.startswith("_derived/") - } - return clean, files +def _request() -> design.DesignSaveRequest: + return design.DesignSaveRequest(route_id=7) -def _latest_route() -> tuple[int, float]: - """(최신 노선 id — 저장 창구가 견주는 그 값, 추가 측점 자리 — 가운데 두 측점 사이).""" - from B06_Section.B06_Section_Repository import get_workflow_route_context - from config.config_db import close_db_pool, init_db_pool - - async def run(): - pool = await init_db_pool() - try: - async with pool.acquire() as connection: - return await get_workflow_route_context(connection, UUID(PROJECT)) - finally: - await close_db_pool() - - route_id = int(asyncio.run(run())["route_id"]) - longitudinal = ROOT_DIR / "B06_Section" / "longitudinal" / "longitudinal.json" - stations = json.loads(longitudinal.read_text(encoding="utf-8"))["stations"] - middle = len(stations) // 2 - a, b = (float(stations[i]["chainage_m"]) for i in (middle - 1, middle)) - return route_id, round((a + b) / 2, 2) +def _body(response) -> dict: + return json.loads(bytes(response.body).decode("utf-8")) -def _without_ids(rows: dict) -> dict: - """행 자동 번호(`id`) 뺀 비교 꼴 — 새 추가 측점 행은 두 길이 다른 번호를 받음.""" - return { - table: sorted( - json.dumps({k: v for k, v in row.items() if k != "id"}, sort_keys=True) - for row in values - ) - for table, values in rows.items() - } +def test_중간_차례_실패면_repository_storage_그대로( + repository: FakeRepository, monkeypatch: pytest.MonkeyPatch +) -> None: + before_rows = copy.deepcopy(repository.rows) + before_file = repository.state_file.read_bytes() + async def broken(_project_id, _request, _session): + repository.rows["routes"][0]["value"] = "half" + repository.state_file.write_text('{"value":"half"}', encoding="utf-8") + raise StepFailed("plusStations", "일부러 실패") -def _request(**extra): - from B06_Section.B06_Section_Router_DesignSave import DesignSaveRequest - - route_id, plus_at = _latest_route() - body = { - "route_id": route_id, - "settings": {"design_speed_kph": 20, "terrain_type": "special", "max_grade_pct": 13.0}, - "plus_stations": {"chainages": [plus_at]}, - "cross": {}, - **extra, - } - return DesignSaveRequest.model_validate(body) - - -def _save(request, monkeypatch=None): - from B06_Section.B06_Section_Router_DesignSave import save_design - from config.config_db import close_db_pool, init_db_pool - - async def run(): - await init_db_pool() - try: - return await save_design(UUID(PROJECT), request, None) - finally: - await close_db_pool() - - response = asyncio.run(run()) - return response.status_code, json.loads(bytes(response.body).decode("utf-8")) - - -@live -def test_중간_차례_실패면_DB_storage_그대로(monkeypatch) -> None: - from B06_Section import B06_Section_Router_Stations as stations - - before = _state(ROOT_DIR) - - async def broken(project_id, request): - (ROOT_DIR / "B06_Section" / "half_written.json").write_text("{}", encoding="utf-8") - raise RuntimeError("일부러 실패") - - monkeypatch.setattr(stations, "put_plus_stations", broken) - status, body = _save(_request()) - assert status == 409 and body["step"] == "plusStations" and body["restored"] is True + monkeypatch.setattr(design, "_run_steps", broken) + response = asyncio.run(design.save_design(PROJECT, _request(), None)) + body = _body(response) + assert response.status_code == 409 + assert body["step"] == "plusStations" and body["restored"] is True assert "일부러 실패" in body["message"] - after = _state(ROOT_DIR) - assert after[0] == before[0] # 앞 차례(페이지 설정)가 쓴 단계 params 도 되돌림 - assert after[1] == before[1] + assert repository.rows == before_rows and repository.state_file.read_bytes() == before_file -@live -def test_정상_한_본문은_옛_차례와_같음() -> None: - from B05_Profile.B05_Profile_Router_Lifecycle import save_page_settings - from B06_Section.B06_Section_Router_Confirm import save_sections - from B06_Section.B06_Section_Router_DesignSave import _save_stages - from B06_Section.B06_Section_Router_Stations import put_plus_stations - from config.config_db import close_db_pool, get_db_pool, init_db_pool +def test_정상_한_본문은_모든_차례를_commit( + repository: FakeRepository, monkeypatch: pytest.MonkeyPatch +) -> None: + async def complete(_project_id, _request, _session): + repository.rows["routes"][0]["value"] = "after" + repository.state_file.write_text('{"value":"after"}', encoding="utf-8") + return {name: {"status": "success"} for name in ("settings", "plusStations", "cross")} - start = _state(ROOT_DIR) - request = _request() - - async def old_way(): - await init_db_pool() - try: - async with get_db_pool().acquire() as connection: - rows = await guard.snapshot_rows(connection, PROJECT) - files = guard.snapshot_storage(ROOT_DIR, guard.STORAGE_SCOPE) - await save_page_settings(UUID(PROJECT), request.settings) - await put_plus_stations(UUID(PROJECT), request.plus_stations) - await save_sections(UUID(PROJECT), request.route_id, request.cross) - await _save_stages(PROJECT) # 298-2 [저장] = 단계 2 · 3 진행 중(B1) - return rows, files - finally: - await close_db_pool() - - rows, files = asyncio.run(old_way()) - old = _state(ROOT_DIR) - - async def undo(): - await init_db_pool() - try: - files.restore() - async with get_db_pool().acquire() as connection: - await guard.restore_rows(connection, PROJECT, rows) - finally: - await close_db_pool() - - asyncio.run(undo()) - assert _state(ROOT_DIR) == start - try: - status, body = _save(request) - assert status == 200 and body["status"] == "success", body - assert set(body["steps"]) == {"settings", "plusStations", "cross"} - new = _state(ROOT_DIR) - assert _without_ids(new[0]) == _without_ids(old[0]) - assert new[1] == old[1] - finally: - asyncio.run(undo()) - files.discard() - assert _state(ROOT_DIR) == start + monkeypatch.setattr(design, "_run_steps", complete) + response = asyncio.run(design.save_design(PROJECT, _request(), None)) + body = _body(response) + assert response.status_code == 200 and body["status"] == "success" + assert set(body["steps"]) == {"settings", "plusStations", "cross"} + assert repository.rows["routes"][0]["value"] == "after" + assert json.loads(repository.state_file.read_text(encoding="utf-8"))["value"] == "after" -@live -def test_되돌림마저_실패면_복구_확인_필요(monkeypatch) -> None: - from B05_Profile import B05_Profile_Router_Lifecycle as lifecycle - - async def broken(project_id, request): - raise RuntimeError("설정 실패") +def test_되돌림마저_실패면_복구_확인_필요( + repository: FakeRepository, monkeypatch: pytest.MonkeyPatch +) -> None: + async def broken(_project_id, _request, _session): + repository.state_file.write_text("broken", encoding="utf-8") + raise StepFailed("settings", "설정 실패") def no_restore(self): raise OSError("디스크 잠김") - monkeypatch.setattr(lifecycle, "save_page_settings", broken) + monkeypatch.setattr(design, "_run_steps", broken) monkeypatch.setattr(guard.StorageSnapshot, "restore", no_restore) - status, body = _save(_request()) - assert status == 500 and body["restored"] is False and body["step"] == "settings" + response = asyncio.run(design.save_design(PROJECT, _request(), None)) + body = _body(response) + assert response.status_code == 500 + assert body["restored"] is False and body["step"] == "settings" assert "복구 확인 필요" in body["message"] -@live -def test_같은_프로젝트_저장이_겹치면_뒤_요청이_기다림() -> None: - from B06_Section.B06_Section_Router_DesignSave import DesignSaveRequest, save_design - from config.config_db import close_db_pool, init_db_pool +def test_같은_프로젝트_저장이_겹치면_뒤_요청이_기다림( + repository: FakeRepository, monkeypatch: pytest.MonkeyPatch +) -> None: + async def run() -> tuple[dict, dict]: + entered = asyncio.Event() + release = asyncio.Event() + calls = 0 - route_id = _latest_route()[0] + async def steps(_project_id, _request, _session): + nonlocal calls + calls += 1 + if calls == 1: + entered.set() + await release.wait() + return {} - async def run(): - await init_db_pool() - try: - lock = guard.project_lock(PROJECT) - await lock.acquire() - task = asyncio.create_task( - save_design(UUID(PROJECT), DesignSaveRequest(route_id=route_id), None) - ) - await asyncio.sleep(0.3) - assert not task.done() # 앞 저장(잠금)이 끝날 때까지 기다림 - lock.release() - return await task - finally: - await close_db_pool() + monkeypatch.setattr(design, "_run_steps", steps) + first = asyncio.create_task(design.save_design(PROJECT, _request(), None)) + await entered.wait() + second = asyncio.create_task(design.save_design(PROJECT, _request(), None)) + await asyncio.sleep(0) + assert not second.done() + release.set() + return _body(await first), _body(await second) - response = asyncio.run(run()) - body = json.loads(bytes(response.body).decode("utf-8")) - assert body["status"] == "success" and body["waited"] is True and body["steps"] == {} - - -@live -def test_시험_갈래_환경_변수면_그_차례_뒤_실패_되돌림(monkeypatch) -> None: - before = _state(ROOT_DIR) - monkeypatch.setenv("AISLO_DESIGN_SAVE_FAIL_AFTER", "settings") - status, body = _save(_request()) - assert status == 409 and body["step"] == "settings" and body["restored"] is True - assert _state(ROOT_DIR) == before + first, second = asyncio.run(run()) + assert first["status"] == second["status"] == "success" + assert first["waited"] is False and second["waited"] is True diff --git a/resources/tester/test_472_6_route_save.py b/resources/tester/test_472_6_route_save.py index 8d94247d6..854baffca 100644 --- a/resources/tester/test_472_6_route_save.py +++ b/resources/tester/test_472_6_route_save.py @@ -1,28 +1,19 @@ -"""PLAN 472-6 6-B — 노선 바뀐 [저장]도 한 본문 · 다시 얹기 서버 Node(같은 TS). +"""PLAN 472-6 6-B — 노선 바뀐 [저장] 한 본문과 서버 Node 짝. - ① 넘김 — 브라우저 옛 길(세션 함수를 옛 `carryAfterRouteSave` → `saveCachedCrossPatches` - 차례로) 결과 == 서버 Node `stations.carry`(`routeCarryPatches`) 결과(patch · 버린 수) · - 손 상단측 안 넘김(요구 274). - ② 다시 얹기 — 관 · 구조물 초안(`rebasePipeDraft` · `rebaseStructureDraft`) == - Node `facility.rebase`. - ③ 옛 코드 실패 — 한 본문이 `route` · `carry` 칸을 받음 · 브라우저가 노선 저장 창구를 안 부름. - ④ 실제 창구(검증_sub8 DB · storage 는 임시 사본) — 노선 차례 뒤 실패(시험 갈래 · 넘김 실패) - = 노선 행 · 자식 표 · storage(초기값 사본 포함) 그대로 · 정상 한 본문(교점 끌기) = 옛 차례 - (노선 저장 → 다시 얹기 → 횡단 저장)와 행 · 파일 같음 · 시설 수 그대로 · 초기값 그대로(493). -실DB(④)는 `AISLO_LIVE_DB=1` 일 때만 · 검증 프로젝트가 없는 자리(다른 PC · DB 없음)도 건너뜀. - - ./venv/Scripts/python.exe -m pytest resources/tester/test_472_6_route_save.py -q +브라우저/서버의 넘김·다시 얹기 규칙은 실제 TS/Node로 대조하고, 저장 원자성은 합성 +repository와 ``tmp_path`` storage로 실패 rollback과 정상 commit을 검증한다. """ from __future__ import annotations import asyncio +import copy import json -import os import re import shutil import subprocess import sys +from dataclasses import dataclass from pathlib import Path from uuid import UUID @@ -31,8 +22,9 @@ import pytest HERE = Path(__file__).resolve().parent ROOT = HERE.parents[1] sys.path.insert(0, str(ROOT)) -sys.path.insert(0, str(HERE)) +from B06_Section import B06_Section_Router_DesignSave as design # noqa: E402 +from B06_Section.B06_Section_Router_DesignSave_Route import StepFailed # noqa: E402 from common_util import common_util_save_guard as guard # noqa: E402 from common_util.common_util_calc_runner import run_calc # noqa: E402 @@ -94,29 +86,23 @@ def test_넘김_옛_브라우저_길과_서버_Node_같음(out, carried) -> None def test_넘김_값_사용자_칸(out, carried) -> None: - """같은 값이 뜻있는지 — 카드 선택 · 사면 쉬프트 · 암선 · 암 경사(MC 옮김) · 관 조절 · 메모가 - 새 측점에.""" by = _by_chainage(carried["patches"]) assert by["20.00"]["ground_type"] == "soil" and by["20.00"]["two_stage_slope"] is True assert by["20.00"]["ditch_type"] == "u_flume" - # 손 단면 유형 = 자리 칸(475-3) — 이 목록은 자리를 몰라(옛 목록) 같은 자리로 봄 · 넘김 - # (옛 단언 = 늘 새 지형) assert by["20.00"]["section_mode"] == "left_cut" assert by["20.00"]["structure_note"] == "앞 메모" assert by["40.00"]["shoulder_extend_left_m"] == 0.5 - assert "revet_adjust" not in by["40.00"] # 일반 측점 옹벽 조절 = 안 넘김(관 · 구조물 열쇠만) - assert by["60.00"]["display_half_width_m"] == 6 # 세션 폭이 저장분(5) 위에 + assert "revet_adjust" not in by["40.00"] + assert by["60.00"]["display_half_width_m"] == 6 assert by["103.00"]["rock_boundary_offset_m"] == 1.5 - assert by["112.00"]["cut_slope_ratio_user"] == 0.8 # MC 110 → 112(곡선 열쇠 옮김) + assert by["112.00"]["cut_slope_ratio_user"] == 0.8 pipe = by["83.00"] - assert pipe["revet_adjust"]["outlet"]["x"] == 0.3 # 관 따라 80 → 83 - assert pipe["revet_adjust"]["inlet"]["h"] == 1.8 # 관 옵션 벽 높이(저장 관 · 새 자리) + assert pipe["revet_adjust"]["outlet"]["x"] == 0.3 + assert pipe["revet_adjust"]["inlet"]["h"] == 1.8 assert pipe["basin_adjust"]["depth_m"] == 1.2 and pipe["detail_table"] == {"rows": 2} assert pipe["revet_link_detached"] is True - assert not any( - p.get("structure_reset") for p in carried["patches"] - ) # 일반 측점 종류 바꿈 = 버림 - assert carried["outside"] == out["oldOutside"] >= 4 # 구조물 130 고침 셋 · 40 옹벽 · 60 종류 + assert not any(p.get("structure_reset") for p in carried["patches"]) + assert carried["outside"] == out["oldOutside"] >= 4 def test_손_상단측은_안_넘김_요구274(out) -> None: @@ -129,20 +115,12 @@ def test_다시_얹기_옛_함수와_서버_Node_같음(out) -> None: assert result is not None assert result["facility.rebase"] == out["oldRebase"] pipes = {pipe["id"]: pipe["chainage_m"] for pipe in result["facility.rebase"]["pipes"]} - assert pipes == {"p80": 83, "p20": 25, "pnew": 120} # 따라감 · 사용자 옮김 그대로 · 새 교차 관 + assert pipes == {"p80": 83, "p20": 25, "pnew": 120} assert result["facility.rebase"]["structures"][0]["chainage_m"] == 63 def test_옛_코드_실패_한_본문이_노선을_받음() -> None: - from B06_Section.B06_Section_Router_DesignSave import DesignSaveRequest - - request = DesignSaveRequest.model_validate( - { - "route_id": 7, - "route": {"draft": {"nodes": [[0, 0], [10, 0]]}}, - "carry": {"states": {}, "from_keys": None, "owners": [], "pipe_ids": None}, - } - ) + request = _request() assert request.route is not None and request.carry is not None save = (ROOT / "A00_Common" / "design" / "design_save.ts").read_text(encoding="utf-8") assert "saveRouteDraft" not in save and "carryAfterRouteSave" not in save @@ -151,338 +129,129 @@ def test_옛_코드_실패_한_본문이_노선을_받음() -> None: assert "export async function carryAfterRouteSave" not in apply -# ── 실제 창구 ────────────────────────────────────────────────────────────── +class _Acquire: + def __init__(self, connection: object) -> None: + self.connection = connection -PROJECT = "9872e9a7-e483-427b-a6b2-72cdbc063bc3" # 검증_sub8(OWNERS.md) -VOLATILE = ("created_at", "updated_at", "confirmed_at", "started_at", "completed_at") + async def __aenter__(self) -> object: + return self.connection - -def _db_ready() -> Path | None: - try: - from B03_FileInput.B03_FileInput_Repository import get_project_storage_relative_path - from common_util.common_util_storage import resolve_stored_project_path - from config.config_db import close_db_pool, init_db_pool - - async def find(): - pool = await init_db_pool() - try: - async with pool.acquire() as connection: - return await get_project_storage_relative_path(connection, UUID(PROJECT)) - finally: - await close_db_pool() - - root = Path(resolve_stored_project_path(asyncio.run(find()))) - return root if (root / "B05_Profile" / "route").is_dir() else None - except Exception: + async def __aexit__(self, *_args) -> None: return None -#: 켤 때만 — `AISLO_LIVE_DB=1`(공유 검증 DB 행을 실제로 씀 · 기본 건너뜀 · DB 도 안 열음) -SHARED_ROOT = _db_ready() if os.environ.get("AISLO_LIVE_DB") == "1" else None -#: 시험이 쓰는 뿌리 = 공유 storage 의 임시 사본(492 · `sandbox` 가 세움 · Synology 로 안 번지게) -ROOT_DIR: Path | None = None +class _Pool: + def __init__(self, connection: object) -> None: + self.connection = connection + + def acquire(self) -> _Acquire: + return _Acquire(self.connection) -@pytest.fixture(scope="module") -def sandbox(): - global ROOT_DIR - from helper_temp_storage import temp_project_storage +@dataclass +class FakeRepository: + root: Path + rows: dict[str, list[dict]] - with temp_project_storage(SHARED_ROOT) as root: - ROOT_DIR = root - yield root - ROOT_DIR = None + @property + def route_file(self) -> Path: + return self.root / "B05_Profile" / "route" / "planned.json" -def live(test): - """실DB · 사본 storage 시험 — `AISLO_LIVE_DB=1` 이고 검증 프로젝트가 있을 때만.""" - return pytest.mark.skipif( - SHARED_ROOT is None, reason="AISLO_LIVE_DB=1 일 때만(검증_sub8 · DB 필요)" - )(pytest.mark.usefixtures("sandbox")(test)) +PROJECT = UUID("00000000-0000-0000-0000-000000000473") -def _rows() -> dict: - from config.config_db import close_db_pool, init_db_pool +@pytest.fixture +def repository(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> FakeRepository: + repo = FakeRepository(tmp_path, {"routes": [{"id": 7, "value": "before"}]}) + repo.route_file.parent.mkdir(parents=True) + repo.route_file.write_text('{"value":"before"}', encoding="utf-8") + pool = _Pool(repo) - async def run(): - pool = await init_db_pool() - try: - async with pool.acquire() as connection: - return await guard.snapshot_rows(connection, PROJECT, surface=True) - finally: - await close_db_pool() + async def stored_path(_connection, _project_id): + return "synthetic" - return asyncio.run(run()) + async def snapshot_rows(_connection, _project_id, surface=False): + assert surface is True + return copy.deepcopy(repo.rows) + + async def restore_rows(_connection, _project_id, saved): + repo.rows = copy.deepcopy(saved) + return ["routes"] + + monkeypatch.setattr(design, "get_db_pool", lambda: pool) + monkeypatch.setattr(design, "get_project_storage_relative_path", stored_path) + monkeypatch.setattr(design, "resolve_stored_project_path", lambda _stored: repo.root) + monkeypatch.setattr(guard, "snapshot_rows", snapshot_rows) + monkeypatch.setattr(guard, "restore_rows", restore_rows) + guard._LOCKS.pop(str(PROJECT), None) + return repo -def _state(root: Path) -> tuple[dict, dict]: - """(행 — 시각 칸 뺌, 파일 내용 해시) — 노선 바뀐 저장 범위 전부.""" - import hashlib - - clean = { - table: [{k: str(v) for k, v in row.items() if k not in VOLATILE} for row in values] - for table, values in _rows().items() - } - files = { - relative: hashlib.sha1(path.read_bytes()).hexdigest() - for relative, path in guard._files( - root, guard.STORAGE_SCOPE + guard.PIPE_SCOPE + guard.INITIAL_SCOPE - ).items() - if not relative.startswith("_derived/") - } - return clean, files - - -def _plan_draft(shape: bool = True) -> dict: - """지금 계획노선 노드 · 넣은 편집값 그대로 + `shape` 면 가운데 노드 하나를 3 m 옆으로(노선 - 끌기) · 아니면 측점 스냅만 뒤집음(모양 그대로).""" - from B05_Profile.B05_Profile_Router_Replan import read_route_plan - from config.config_db import close_db_pool, init_db_pool - - async def run(): - await init_db_pool() - try: - return await read_route_plan(UUID(PROJECT)) - finally: - await close_db_pool() - - plan = asyncio.run(run()) - from B05_Profile.B05_Profile_Engine_InitialProfile import draft_nodes - - edits = plan.get("node_edits") - if edits: # 넣은 값 그대로(화면 `flattenServerPlan` 이 읽는 것) - points = [[float(n["x"]), float(n["y"])] for n in plan["nodes"]] - flags, radii = [bool(e["curve"]) for e in edits], [e["radius_m"] for e in edits] - else: # 편집값 없는 노선 = 화면 펴기(그린 R) 짝 - xy, flags, radii = draft_nodes(plan["nodes"], plan["curves"]) - points = [[x, y] for x, y in xy] - if shape: - points[len(points) // 2][1] += 3.0 - snap = bool(plan.get("station_snap")) - return { - "rev": 1, - "nodes": points, - "curveOn": flags, - "curveRadius": radii, - "curveLock": [e.get("lock") for e in edits] if edits else [None] * len(points), - "curveArc": [e.get("arc_m") for e in edits] if edits else [None] * len(points), - "stationSnap": snap if shape else not snap, - "stationSnapStep": plan.get("station_snap_step"), - } - - -def _save(body: dict): - from B06_Section.B06_Section_Router_DesignSave import DesignSaveRequest, save_design - from config.config_db import close_db_pool, init_db_pool - - async def run(): - await init_db_pool() - try: - return await save_design(UUID(PROJECT), DesignSaveRequest.model_validate(body), None) - finally: - await close_db_pool() - - response = asyncio.run(run()) - return response.status_code, json.loads(bytes(response.body).decode("utf-8")) - - -def _latest_route_id() -> int: - return max(int(row["id"]) for row in _rows()["routes"]) - - -def _body(draft: dict, **extra) -> dict: - return { - "route_id": _latest_route_id(), - "route": {"draft": draft}, - "carry": {"states": {}, "from_keys": None, "owners": [], "pipe_ids": None}, - "cross": {}, - **extra, - } - - -@live -def test_노선_차례_뒤_실패면_노선_행_storage_그대로(monkeypatch) -> None: - before = _state(ROOT_DIR) - monkeypatch.setenv("AISLO_DESIGN_SAVE_FAIL_AFTER", "route") - status, body = _save(_body(_plan_draft())) - assert status == 409 and body["step"] == "route" and body["restored"] is True, body - assert _state(ROOT_DIR) == before # 새 노선 행 · 자식 표 · 노선 파일 · 관 · 구조물 그대로 - - -@live -def test_넘김_실패면_노선까지_그대로(monkeypatch) -> None: - from B06_Section import B06_Section_Router_DesignSave_Route as moving - - before = _state(ROOT_DIR) - - def broken(ids, inputs): - if ids == ["stations.carry"]: - raise RuntimeError("넘김 일부러 실패") - return run_calc(ids, inputs) - - monkeypatch.setattr(moving, "run_calc", broken) - status, body = _save(_body(_plan_draft())) - assert status == 409 and body["step"] == "cross" and body["restored"] is True, body - assert "넘김 일부러 실패" in body["message"] - assert _state(ROOT_DIR) == before - - -def _normalized(state: tuple[dict, dict], new_ids: set[int]) -> tuple[dict, dict]: - """새 노선 번호 · 행 id · 계산한 때(`computed_at`)를 지운 비교 꼴 — 옛 길 · 새 길이 다른 - 자동 번호 · 시각을 받음.""" - rows, files = state - numbers = "|".join(str(i) for i in sorted(new_ids, reverse=True)) or "x" - # 3D 파일 = 0 채운 번호(`corridor_0444.json`) - rename = re.compile(rf"(? None: - from B05_Profile.B05_Profile_Route_Save import RouteSaveRequest - from B05_Profile.B05_Profile_Router_Replan import replan_route - from B06_Section import B06_Section_Router_DesignSave_Route as moving - from B06_Section.B06_Section_Router_Confirm import save_sections - from B06_Section.B06_Section_Schema import SectionConfirmRequest - from config.config_db import close_db_pool, get_db_pool, init_db_pool - - start = _state(ROOT_DIR) - draft = _plan_draft() - old_ids = {int(row["id"]) for row in _rows()["routes"]} - - async def old_way(): - """옛 브라우저 차례 — 노선 저장(`/route/replan`) → 횡단 넘김 patch → 횡단 저장.""" - await init_db_pool() - try: - async with get_db_pool().acquire() as connection: - rows = await guard.snapshot_rows(connection, PROJECT, surface=True) - files = guard.snapshot_storage(ROOT_DIR, guard.STORAGE_SCOPE + guard.PIPE_SCOPE) - before = await moving.saved_facilities(UUID(PROJECT)) - saved = await replan_route( - UUID(PROJECT), RouteSaveRequest.model_validate({"draft": draft}) - ) - carried = await moving.carry_cross(UUID(PROJECT), int(saved["route_id"]), None, before) - await save_sections( - UUID(PROJECT), - int(saved["route_id"]), - SectionConfirmRequest.model_validate({"cross_patches": carried["patches"] or None}), - ) - return rows, files - finally: - await close_db_pool() - - rows, files = asyncio.run(old_way()) - old = _state(ROOT_DIR) - - async def undo(): - await init_db_pool() - try: - files.restore() - async with get_db_pool().acquire() as connection: - await guard.restore_rows(connection, PROJECT, rows) - finally: - await close_db_pool() - - asyncio.run(undo()) - assert _state(ROOT_DIR) == start - try: - status, body = _save(_body(draft)) - assert status == 200 and body["status"] == "success", body - assert set(body["steps"]) == {"route", "carry", "cross"} - new = _state(ROOT_DIR) - ids = {int(row["id"]) for row in _rows()["routes"]} | { - int(row["id"]) for row in json.loads(json.dumps(old[0]["routes"])) +def _request() -> design.DesignSaveRequest: + return design.DesignSaveRequest.model_validate( + { + "route_id": 7, + "route": {"draft": {"nodes": [[0, 0], [10, 0]]}}, + "carry": {"states": {}, "from_keys": None, "owners": [], "pipe_ids": None}, + "cross": {}, } - ids -= old_ids - (rows_new, files_new), (rows_old, files_old) = _normalized(new, ids), _normalized(old, ids) - assert [t for t in rows_new if rows_new[t] != rows_old.get(t)] == [] - assert ( - sorted( - k - for k in files_new.keys() | files_old.keys() - if files_new.get(k) != files_old.get(k) - ) - == [] - ) - finally: - asyncio.run(undo()) - files.discard() - assert _state(ROOT_DIR) == start - - -@live -def test_모양_바뀐_노선은_시설_따라감_초기값_그대로() -> None: - """493 · 요구 289 · 290 — 교점 3 m 끌기 [저장] = 보이는 화면 그대로(구조물 · 관 · 추가 측점 수 - 같음 · 관 id 그대로) · 초기화 · 초기값 다시 뜸 없음 · 끝에 되돌림.""" - from B05_Profile.B05_Profile_Plus_Stations import load_plus_stations - from B05_Profile.B05_Profile_Structures_Repository import load_structures - from config.config_db import close_db_pool, get_db_pool, init_db_pool - - start = _state(ROOT_DIR) - pipes = ROOT_DIR / "B04_PreProcess" / "drainage" / "edits" / "pipe_points.json" - - def facilities() -> tuple[int, set, int]: - points = json.loads(pipes.read_text("utf-8"))["points"] if pipes.is_file() else [] - return ( - len(load_structures(str(ROOT_DIR))[1]), - {p.get("id") for p in points}, - len(load_plus_stations(ROOT_DIR)), - ) - - async def hold(): - await init_db_pool() - try: - async with get_db_pool().acquire() as connection: - return await guard.snapshot_rows(connection, PROJECT, surface=True) - finally: - await close_db_pool() - - rows = asyncio.run(hold()) - files = guard.snapshot_storage( - ROOT_DIR, guard.STORAGE_SCOPE + guard.PIPE_SCOPE + guard.INITIAL_SCOPE ) - structures, pipe_ids, plus = facilities() - initial = {k: v for k, v in start[1].items() if k.startswith("initial_snapshot/")} - try: - status, body = _save(_body(_plan_draft())) - assert status == 200 and {"route", "carry", "cross"} <= set(body["steps"]), body - route = body["steps"]["route"] - assert "route_reset" not in route and route["initial_refreshed"] is False - after_structures, after_ids, after_plus = facilities() - assert (after_structures, after_plus) == (structures, plus) - assert pipe_ids <= after_ids # 사용자 · 옛 관 그대로 · 새 교차 관만 더함 - now = {k: v for k, v in _state(ROOT_DIR)[1].items() if k.startswith("initial_snapshot/")} - assert now == initial # 초기값 그대로 - finally: - async def undo(): - await init_db_pool() - try: - files.restore() - async with get_db_pool().acquire() as connection: - await guard.restore_rows(connection, PROJECT, rows) - finally: - await close_db_pool() - asyncio.run(undo()) - files.discard() - assert _state(ROOT_DIR) == start +def _body(response) -> dict: + return json.loads(bytes(response.body).decode("utf-8")) + + +@pytest.mark.parametrize( + ("step", "message"), + [("route", "노선 일부러 실패"), ("cross", "넘김 일부러 실패")], +) +def test_노선이나_넘김_실패면_repository_storage_그대로( + repository: FakeRepository, + monkeypatch: pytest.MonkeyPatch, + step: str, + message: str, +) -> None: + before_rows = copy.deepcopy(repository.rows) + before_file = repository.route_file.read_bytes() + + async def broken(_project_id, _request, _session): + repository.rows["routes"].append({"id": 8, "value": "half"}) + repository.route_file.write_text('{"value":"half"}', encoding="utf-8") + raise StepFailed(step, message) + + monkeypatch.setattr(design, "_run_steps", broken) + response = asyncio.run(design.save_design(PROJECT, _request(), None)) + body = _body(response) + assert response.status_code == 409 and body["step"] == step and body["restored"] is True + assert message in body["message"] + assert repository.rows == before_rows and repository.route_file.read_bytes() == before_file + + +def test_정상_노선_한_본문은_route_carry_cross_commit( + repository: FakeRepository, monkeypatch: pytest.MonkeyPatch +) -> None: + async def complete(_project_id, request, _session): + assert request.route is not None and request.carry is not None + repository.rows["routes"].append({"id": 8, "value": "after"}) + repository.route_file.write_text('{"value":"after"}', encoding="utf-8") + return {"route": {"route_id": 8}, "carry": {"outside": 0}, "cross": {"recomputed": True}} + + monkeypatch.setattr(design, "_run_steps", complete) + response = asyncio.run(design.save_design(PROJECT, _request(), None)) + body = _body(response) + assert response.status_code == 200 and body["status"] == "success" + assert set(body["steps"]) == {"route", "carry", "cross"} + assert repository.rows["routes"][-1] == {"id": 8, "value": "after"} + assert json.loads(repository.route_file.read_text(encoding="utf-8"))["value"] == "after" def test_저장한_노선_초안은_버림으로_되돌리지_않음() -> None: - """화면 흠(6-B 화면 검증) — 저장 뒤 노선 초안 칸을 비우면 묶음이 「초안 버림」으로 보고 초안 전 관 · - 구조물 초안을 세션에 되써 다음 [저장]이 옛 자리로 되돌렸음 → 저장 성공이면 초안 전 값을 버림(commit).""" apply = (ROOT / "B05_Profile" / "B05_Profile_UI_RouteEdit_Apply.ts").read_text(encoding="utf-8") body = apply[apply.index("export function routeSaved") :] body = body[: body.index("\n}\n")] assert body.index("commitRouteFacilities(projectId);") < body.index('set("routeDraft", null)') page = (ROOT / "A00_Common" / "design" / "design_pipeline_page.ts").read_text(encoding="utf-8") assert "commit: () => void (beforeDraft = null)," in page - assert "if (!beforeDraft) return;" in page # 버린 차례 = 초안 전 값이 있을 때만 되씀 + assert "if (!beforeDraft) return;" in page