258 lines
9.5 KiB
Python
258 lines
9.5 KiB
Python
"""PLAN 472-6 6-B — 노선 바뀐 [저장] 한 본문과 서버 Node 짝.
|
|
|
|
브라우저/서버의 넘김·다시 얹기 규칙은 실제 TS/Node로 대조하고, 저장 원자성은 합성
|
|
repository와 ``tmp_path`` storage로 실패 rollback과 정상 commit을 검증한다.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import copy
|
|
import json
|
|
import re
|
|
import shutil
|
|
import subprocess
|
|
import sys
|
|
from dataclasses import dataclass
|
|
from pathlib import Path
|
|
from uuid import UUID
|
|
|
|
import pytest
|
|
|
|
HERE = Path(__file__).resolve().parent
|
|
ROOT = HERE.parents[1]
|
|
sys.path.insert(0, str(ROOT))
|
|
|
|
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
|
|
|
|
pytestmark = pytest.mark.skipif(
|
|
not (ROOT / "config" / "node_modules" / "vite").is_dir(),
|
|
reason="프론트엔드 의존성 미설치",
|
|
)
|
|
|
|
|
|
def _node(*args: str, timeout: int = 300) -> str:
|
|
proc = subprocess.run( # noqa: S603 — 고정 실행 파일
|
|
["node", *args], cwd=ROOT, capture_output=True, text=True, encoding="utf-8", timeout=timeout
|
|
)
|
|
assert proc.returncode == 0, proc.stderr
|
|
return proc.stdout
|
|
|
|
|
|
@pytest.fixture(scope="module")
|
|
def out(tmp_path_factory) -> dict:
|
|
"""vite ssr 로 진짜 모듈을 묶어 옛 길을 돌림(417 꼴)."""
|
|
work = tmp_path_factory.mktemp("r6b")
|
|
_node(
|
|
str(ROOT / "config" / "node_modules" / "vite" / "bin" / "vite.js"),
|
|
"build",
|
|
"--configLoader",
|
|
"runner",
|
|
"--ssr",
|
|
"../resources/tester/helper_472_6_route_entry.ts",
|
|
"--outDir",
|
|
str(work),
|
|
"--logLevel",
|
|
"error",
|
|
)
|
|
bundle = work / "bundle.mjs"
|
|
shutil.copy(work / "helper_472_6_route_entry.js", bundle)
|
|
return json.loads(_node(str(HERE / "helper_472_6_route_carry.mjs"), str(bundle)))
|
|
|
|
|
|
@pytest.fixture(scope="module")
|
|
def carried(out) -> dict:
|
|
inputs = {
|
|
**out["carry"],
|
|
"detail": out["detail"],
|
|
"pipes_before": out["pipesBefore"],
|
|
"pipes_after": out["pipesAfter"],
|
|
}
|
|
result = run_calc(["stations.carry"], {"carry": inputs})
|
|
assert result is not None
|
|
return result["stations.carry"]
|
|
|
|
|
|
def _by_chainage(patches: list[dict]) -> dict[str, dict]:
|
|
return {f"{patch['chainage_m']:.2f}": patch for patch in patches}
|
|
|
|
|
|
def test_넘김_옛_브라우저_길과_서버_Node_같음(out, carried) -> None:
|
|
assert carried["patches"] == out["oldPatches"]
|
|
assert carried["outside"] == out["oldOutside"]
|
|
|
|
|
|
def test_넘김_값_사용자_칸(out, carried) -> None:
|
|
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"
|
|
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
|
|
assert by["103.00"]["rock_boundary_offset_m"] == 1.5
|
|
assert by["112.00"]["cut_slope_ratio_user"] == 0.8
|
|
pipe = by["83.00"]
|
|
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
|
|
|
|
|
|
def test_손_상단측은_안_넘김_요구274(out) -> None:
|
|
assert "uphill" not in out["carry"]["states"]
|
|
assert out["carry"]["states"]["kindreset"] == {"60.00": "culvert"}
|
|
|
|
|
|
def test_다시_얹기_옛_함수와_서버_Node_같음(out) -> None:
|
|
result = run_calc(["facility.rebase"], {"rebase": out["rebaseInput"]})
|
|
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 result["facility.rebase"]["structures"][0]["chainage_m"] == 63
|
|
|
|
|
|
def test_옛_코드_실패_한_본문이_노선을_받음() -> 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
|
|
assert re.search(r"postDesignSave\([^;]*route \? \{ route, carry", save, re.S)
|
|
apply = (ROOT / "B05_Profile" / "B05_Profile_UI_RouteEdit_Apply.ts").read_text(encoding="utf-8")
|
|
assert "export async function carryAfterRouteSave" not in apply
|
|
|
|
|
|
class _Acquire:
|
|
def __init__(self, connection: object) -> None:
|
|
self.connection = connection
|
|
|
|
async def __aenter__(self) -> object:
|
|
return self.connection
|
|
|
|
async def __aexit__(self, *_args) -> None:
|
|
return None
|
|
|
|
|
|
class _Pool:
|
|
def __init__(self, connection: object) -> None:
|
|
self.connection = connection
|
|
|
|
def acquire(self) -> _Acquire:
|
|
return _Acquire(self.connection)
|
|
|
|
|
|
@dataclass
|
|
class FakeRepository:
|
|
root: Path
|
|
rows: dict[str, list[dict]]
|
|
|
|
@property
|
|
def route_file(self) -> Path:
|
|
return self.root / "B05_Profile" / "route" / "planned.json"
|
|
|
|
|
|
PROJECT = UUID("00000000-0000-0000-0000-000000000473")
|
|
|
|
|
|
@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 stored_path(_connection, _project_id):
|
|
return "synthetic"
|
|
|
|
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 _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": {},
|
|
}
|
|
)
|
|
|
|
|
|
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:
|
|
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
|