eomsangdonandClaude Opus 5.5
17534a01ad
- B05 지형 로더 models_dir · cache_dir=None · 비용면 cache_dir=None · 탐색 복도 corridor_m(기본 100 m · 기본이면 묶음 열쇠 그대로) - D03_PlanAnalysis_Terrain — 지점 + 1 km 표본 3×3 도엽 · prune 없음 · 병합 옛 ∪ 새 · 위성 합집합 · 연도별 models_<연도> DEM · plan.prj - D03_PlanAnalysis_Analyze — 연도 구간마다 갈래 넷 · 법정 기준 · 누가 이어 셈 · 파생 저장본(입력 지문) - D03_PlanAnalysis_Router — POST plan/terrain · plan/analyze(뒤에서 · 진행률) · GET plan/progress · plan/analysis · PUT [저장] - 지킴 시험 test_530_1_plan_terrain · test_530_2_plan_analyze · test_530_2_plan_router Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
128 lines
4.2 KiB
Python
128 lines
4.2 KiB
Python
"""D 분석 API(PLAN 530-2) — GET 이은 누가 · PUT [저장] = 서버 저장본으로 다시 셈 · 모르는 갈래 거절.
|
|
|
|
./venv/Scripts/python.exe -m pytest resources/tester/test_530_2_plan_router.py -q
|
|
"""
|
|
|
|
import json
|
|
|
|
import pytest
|
|
from fastapi import FastAPI
|
|
from fastapi.testclient import TestClient
|
|
|
|
from common_util.common_util_json import atomic_write_json
|
|
from D03_PlanAnalysis import D03_PlanAnalysis_Analyze as analyze
|
|
from D03_PlanAnalysis import D03_PlanAnalysis_Router as router_mod
|
|
|
|
PID = "12953f5a-5ba1-42ad-a427-8ac857a58bc2"
|
|
P0, P1, P2 = (0.0, 0.0), (300.0, 400.0), (900.0, 400.0)
|
|
|
|
|
|
def _route(points: list, length: float) -> dict:
|
|
return {"planned": points, "nodes": [], "curves": [], "metrics": {"length_m": length}}
|
|
|
|
|
|
class _Cursor:
|
|
def __init__(self, log):
|
|
self.log = log
|
|
|
|
async def __aenter__(self):
|
|
return self
|
|
|
|
async def __aexit__(self, *exc):
|
|
return False
|
|
|
|
async def execute(self, sql, args=None):
|
|
self.log.append((sql, args))
|
|
|
|
|
|
class _Connection(_Cursor):
|
|
async def begin(self):
|
|
pass
|
|
|
|
async def commit(self):
|
|
self.log.append(("COMMIT", None))
|
|
|
|
async def rollback(self):
|
|
pass
|
|
|
|
def cursor(self, *args):
|
|
return _Cursor(self.log)
|
|
|
|
|
|
class _Pool:
|
|
def __init__(self, log):
|
|
self.log = log
|
|
|
|
def acquire(self):
|
|
return _Connection(self.log)
|
|
|
|
|
|
@pytest.fixture
|
|
def client(tmp_path, monkeypatch):
|
|
segments = [
|
|
{"year": 2026, "start": P0, "end": P1, "epsg": 5186, "route_json": None},
|
|
{"year": 2027, "start": P1, "end": P2, "epsg": 5186, "route_json": '{"mode": "gentle"}'},
|
|
]
|
|
for seg, routes in (
|
|
(segments[0], {"gradeline": _route([[0, 0], [100, 200], [300, 400]], 530.0)}),
|
|
(
|
|
segments[1],
|
|
{
|
|
"gradeline": _route([[300, 400], [900, 400]], 600.0),
|
|
"gentle": _route([[300, 400], [600, 500], [900, 400]], 640.0),
|
|
"earthwork": {"error": "길 없음"},
|
|
},
|
|
),
|
|
):
|
|
atomic_write_json(
|
|
analyze.derived_dir(tmp_path) / f"route_{seg['year']}.json",
|
|
{
|
|
"year": seg["year"],
|
|
"start": list(seg["start"]),
|
|
"end": list(seg["end"]),
|
|
"fingerprint": "f",
|
|
"routes": routes,
|
|
},
|
|
)
|
|
project = {"road_type": "work", "design_speed_kph": None, "route_start_m": 1000.0}
|
|
|
|
async def inputs(project_id):
|
|
return tmp_path, project, segments
|
|
|
|
log: list = []
|
|
monkeypatch.setattr(router_mod, "_plan_inputs", inputs)
|
|
monkeypatch.setattr(router_mod, "get_db_pool", lambda: _Pool(log))
|
|
app = FastAPI()
|
|
app.include_router(router_mod.router)
|
|
test = TestClient(app)
|
|
test.log = log
|
|
return test
|
|
|
|
|
|
def test_get_chains_saved_choice(client) -> None:
|
|
body = client.get(f"/api/projects/{PID}/plan/analysis").json()
|
|
chain = body["chain"]
|
|
assert [r["mode"] for r in chain] == ["gradeline", "gentle"] # 저장값 · 없으면 첫 성공
|
|
assert chain[0]["start_chainage_m"] == 1000.0
|
|
assert chain[1]["start_chainage_m"] == chain[0]["end_chainage_m"] == 1530.0
|
|
assert chain[1]["end_chainage_m"] == 2170.0
|
|
assert body["line"] == [[0, 0], [100, 200], [300, 400], [600, 500], [900, 400]]
|
|
|
|
|
|
def test_put_saves_server_routes(client) -> None:
|
|
got = client.put(f"/api/projects/{PID}/plan/analysis", json={"choices": {"2027": "gradeline"}})
|
|
assert got.status_code == 200
|
|
assert got.json()["chain"][1]["end_chainage_m"] == 2130.0
|
|
updates = [args for sql, args in client.log if sql.startswith("UPDATE plan_route_years")]
|
|
saved = {args[3]: json.loads(args[0]) for args in updates}
|
|
assert saved[2027]["mode"] == "gradeline" and saved[2027]["planned"] == [[300, 400], [900, 400]]
|
|
assert saved[2026]["start_chainage_m"] == 1000.0 and updates[0][1] == "f"
|
|
assert client.log[-1][0] == "COMMIT"
|
|
|
|
|
|
def test_put_rejects_unknown_or_empty_mode(client) -> None:
|
|
url = f"/api/projects/{PID}/plan/analysis"
|
|
assert client.put(url, json={"choices": {"2027": "follow"}}).status_code == 400
|
|
assert client.put(url, json={"choices": {"2027": "earthwork"}}).status_code == 409
|
|
assert not [sql for sql, _ in client.log if sql.startswith("UPDATE")]
|