Merge remote-tracking branch 'origin/sub_laptop_2' into main_laptop_1
This commit is contained in:
@@ -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:
|
||||
|
||||
@@ -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]}"
|
||||
|
||||
|
||||
|
||||
@@ -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}"
|
||||
|
||||
|
||||
|
||||
@@ -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):
|
||||
|
||||
@@ -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)}")
|
||||
|
||||
@@ -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):
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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:
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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):
|
||||
|
||||
@@ -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"]
|
||||
|
||||
@@ -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,
|
||||
)
|
||||
|
||||
@@ -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:
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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"(?<!\d)0*({numbers})(?!\d)")
|
||||
keep = {}
|
||||
for table, values in rows.items():
|
||||
out = []
|
||||
for row in values:
|
||||
row = {k: v for k, v in row.items() if k not in ("id", "route_id", "computed_at")}
|
||||
out.append(rename.sub("N", json.dumps(row, sort_keys=True, ensure_ascii=False)))
|
||||
keep[table] = sorted(out)
|
||||
return keep, {rename.sub("N", k): v for k, v in files.items()}
|
||||
|
||||
|
||||
@live
|
||||
def test_정상_한_본문은_옛_차례와_같음() -> 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
|
||||
|
||||
Reference in New Issue
Block a user