eomsangdonandClaude Opus 5.5
f17db59ab6
- 지형 받기 — 연도 구간마다 models_<연도> 대신 한 벌 models_plan · 범위 = 전체 시점 · 종점(+ 그은 점) 상자 + 여유 ∪ 받은 범위(줄지 않음) · 받은 범위 plan_terrain.json
- 도엽 = 범위 상자에 닿는 도엽 모두(3×3 이웃 대신) · 위성 = 그 도엽 도곽
- 새 GET …/plan/terrain/extent — 상자 네 귀 경위도 · 도엽 도곽 · 칸 수 어림 · 받은 범위 · covered(더 받기 판정) · 내려받기 없음
- POST …/plan/terrain 몸 = {margin_m, extra} · 여유 처음 값 350 m(옛 DEM 폭) · 바닥 300 m · 실패 글에 병합 도엽 매수
- 전체 시점 · 종점 = 임시로 첫 연도 시점 · 끝 연도 종점(_plan_ends · 560-2 받은 뒤 바꿈)
- 시험 530_1 다시 씀 · 530_2_plan_router 범위 · 받기 둘 더함 · 530_2_plan_analyze 옛 연도 DEM 기대 한 줄 고침(한 벌로 바뀜)
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
355 lines
15 KiB
Python
355 lines
15 KiB
Python
"""D 분석 API(PLAN 530-2 · 530-3 · 561-2) — GET 이은 누가 · PUT [저장] = 서버 저장본으로 다시 셈 ·
|
|
모르는 갈래 거절 · 끈 지점(points) = 그 지점으로 보기 · 안 이어지면 거절 ·
|
|
[저장] 때만 지점 정본(산림청만) ·
|
|
꺾임점 고침(543-3) = 서버가 마디로 선 · 지표를 다시 셈 · 누가가 고친 길이로 · GET 은 지문 같은
|
|
고침만 · 지형 = 전체 시점 · 종점 한 벌 · 범위 보기.
|
|
|
|
./venv/Scripts/python.exe -m pytest resources/tester/test_530_2_plan_router.py -q
|
|
"""
|
|
|
|
import json
|
|
import os
|
|
|
|
import pytest
|
|
from fastapi import FastAPI
|
|
from fastapi.testclient import TestClient
|
|
|
|
from common_util.common_util_auth import require_company
|
|
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_Edit as plan_edit
|
|
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",
|
|
"criteria": {"min_radius_m": 12.0, "max_grade_pct": 14.0},
|
|
"routes": routes,
|
|
},
|
|
)
|
|
project = {
|
|
"road_type": "work",
|
|
"design_speed_kph": None,
|
|
"route_start_m": 1000.0,
|
|
"route_end_m": 2500.0,
|
|
}
|
|
|
|
async def inputs(project_id, points=None):
|
|
return tmp_path, project, router_mod._override(segments, points) if points else 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)
|
|
session = {"role": "USER", "email": "a@forest.go.kr"}
|
|
app.dependency_overrides[require_company] = lambda: session
|
|
stored: list = []
|
|
|
|
async def store_points(project_id, segs):
|
|
stored.append([(s["year"], s["start"], s["end"]) for s in segs])
|
|
|
|
monkeypatch.setattr(router_mod, "_store_points", store_points)
|
|
# 꺾임점 고침 지표 — 지형 없이(선 길이만)
|
|
loads: list = []
|
|
monkeypatch.setattr(
|
|
analyze, "_load", lambda root, year, loader: loads.append(year) or (None, [])
|
|
)
|
|
monkeypatch.setattr(
|
|
plan_edit,
|
|
"edit_metrics",
|
|
lambda line, result, terrain, streams: {
|
|
"length_m": round(analyze.line_length(line.vertices), 2),
|
|
"max_grade_pct": 9.5,
|
|
"over_limit": {"total": 1},
|
|
},
|
|
)
|
|
test = TestClient(app)
|
|
test.segments = segments
|
|
test.loads = loads
|
|
test.log = log
|
|
test.session = session
|
|
test.stored = stored
|
|
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]]
|
|
assert body["end_chainage_target_m"] == 2500.0 # 등록 목표 — 화면이 차이를 보임
|
|
|
|
|
|
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")]
|
|
|
|
|
|
def _points(p1) -> list:
|
|
return [
|
|
{"year": 2026, "start": list(P0), "end": list(p1)},
|
|
{"year": 2027, "start": list(p1), "end": list(P2)},
|
|
]
|
|
|
|
|
|
def test_points_override_sees_matching_store_only(client) -> None:
|
|
url = f"/api/projects/{PID}/plan/analysis"
|
|
same = client.get(url, params={"points": json.dumps(_points(P1))}).json()
|
|
assert same["chain"] is not None and same["line"][-1] == [900, 400]
|
|
moved = client.get(url, params={"points": json.dumps(_points((310.0, 400.0)))}).json()
|
|
assert moved["chain"] is None and moved["years"][0]["analysis"] is None # 끈 지점 = 저장본 없음
|
|
|
|
|
|
def test_points_must_chain_and_cover_years(client) -> None:
|
|
url = f"/api/projects/{PID}/plan/analysis"
|
|
broken = _points(P1)
|
|
broken[1]["start"] = [301.0, 400.0]
|
|
assert client.get(url, params={"points": json.dumps(broken)}).status_code == 400
|
|
assert client.get(url, params={"points": json.dumps(_points(P1)[:1])}).status_code == 400
|
|
|
|
|
|
def test_put_with_points_stores_points_forest_only(client) -> None:
|
|
url = f"/api/projects/{PID}/plan/analysis"
|
|
body = {"choices": {"2027": "gradeline"}, "points": _points(P1)}
|
|
assert client.put(url, json=body).status_code == 200
|
|
assert client.stored == [[(2026, P0, P1), (2027, P1, P2)]]
|
|
assert client.put(url, json={"choices": {}}).status_code == 200 and len(client.stored) == 1
|
|
client.session["email"] = "a@gmail.com"
|
|
assert client.put(url, json=body).status_code == 403
|
|
moved = {"choices": {}, "points": _points((310.0, 400.0))}
|
|
client.session["email"] = "a@forest.go.kr"
|
|
assert client.put(url, json=moved).status_code == 409 and len(client.stored) == 1
|
|
|
|
|
|
def _edit(nodes, fingerprint="f") -> dict:
|
|
return {
|
|
"base_mode": "gentle",
|
|
"base_fingerprint": fingerprint,
|
|
"nodes": nodes,
|
|
"curve_flags": [True] * len(nodes),
|
|
"radii": [None] * len(nodes),
|
|
}
|
|
|
|
|
|
def test_put_edit_server_redraws_and_get_returns(client) -> None:
|
|
url = f"/api/projects/{PID}/plan/analysis"
|
|
edit = _edit([[300, 400], [600, 560], [900, 400]])
|
|
got = client.put(url, json={"choices": {}, "edits": {"2027": edit}})
|
|
assert got.status_code == 200, got.text
|
|
updates = {args[3]: json.loads(args[0]) for sql, args in client.log if sql.startswith("UPDATE")}
|
|
body = updates[2027]
|
|
line = plan_edit.edited_line(edit, 12.0)
|
|
assert len(body["planned"]) == len(line.vertices) and body["edit"]["nodes"] == edit["nodes"]
|
|
assert body["planned"][0] == [300, 400] and body["planned"][-1] == [900, 400]
|
|
assert body["metrics"]["max_grade_pct"] == 9.5 and len(body["curves"]) == 1
|
|
length = analyze.line_length(body["planned"])
|
|
assert got.json()["chain"][1]["length_m"] == round(length, 3) != 640.0
|
|
assert "edit" not in updates[2026] and client.loads == [2027]
|
|
|
|
# GET — 저장된 고침 + 고친 길이로 이은 누가 · 지문이 바뀌면 안 돌려줌
|
|
client.segments[1]["route_json"] = json.dumps(body)
|
|
view = client.get(url).json()
|
|
assert view["years"][1]["edit"]["nodes"] == edit["nodes"]
|
|
assert view["years"][1]["edit"]["metrics"]["over_limit"]["total"] == 1
|
|
assert view["chain"][1]["length_m"] == round(length, 3)
|
|
assert view["years"][1]["analysis"]["routes"]["gentle"]["metrics"]["length_m"] == 640.0
|
|
# 같은 입력 다시 [저장] = 저장된 선 · 지표 그대로(지형 안 읽음)
|
|
assert client.put(url, json={"choices": {}, "edits": {"2027": edit}}).status_code == 200
|
|
assert client.loads == [2027]
|
|
client.segments[1]["route_json"] = json.dumps(
|
|
{**body, "edit": {**body["edit"], "base_fingerprint": "old"}}
|
|
)
|
|
stale = client.get(url).json()
|
|
assert stale["years"][1]["edit"] is None and stale["chain"][1]["length_m"] == 640.0
|
|
|
|
|
|
def test_put_edit_rejects(client) -> None:
|
|
url = f"/api/projects/{PID}/plan/analysis"
|
|
off_end = _edit([[300, 400], [600, 560], [901, 400]])
|
|
stale = _edit([[300, 400], [600, 560], [900, 400]], "old")
|
|
other = {**stale, "base_fingerprint": "f", "base_mode": "gradeline"}
|
|
for bad in (off_end, stale, other):
|
|
assert client.put(url, json={"choices": {}, "edits": {"2027": bad}}).status_code == 409
|
|
client.session["email"] = "a@gmail.com"
|
|
good = _edit([[300, 400], [600, 560], [900, 400]])
|
|
assert client.put(url, json={"choices": {}, "edits": {"2027": good}}).status_code == 403
|
|
assert not [sql for sql, _ in client.log if sql.startswith("UPDATE")]
|
|
|
|
|
|
def _analyze_moved(client, tmp_path, monkeypatch, p1, length=700.0) -> None:
|
|
"""끈 지점 분석(`POST …/plan/analyze` 와 같은 길) — 엔진 · 지문은 가짜."""
|
|
monkeypatch.setattr(analyze, "segment_fingerprint", lambda seg, *a, **k: "g")
|
|
monkeypatch.setattr(
|
|
analyze,
|
|
"generate_initial_route",
|
|
lambda expected, *a, mode, **k: _route([list(expected[0]), list(expected[1])], length),
|
|
)
|
|
points = [router_mod.PlanPoint(**p) for p in _points(p1)]
|
|
segs = router_mod._override(client.segments, points)
|
|
analyze.analyze_plan(tmp_path, segs, analyze.plan_criteria("work", None), workers=1)
|
|
|
|
|
|
def test_moved_analysis_keeps_saved_store(client, tmp_path, monkeypatch) -> None:
|
|
"""543-4 — 끈 지점 분석은 지점 열쇠 파일로 · 저장 지점 GET 그대로 · 끈 지점 GET 은 그 분석 ·
|
|
[저장] 뒤 정본 = 끈 지점 분석본 · 옛 꼴 파일은 읽기만 · 쌓임 거르기는 저장 지점 것을 남김."""
|
|
url = f"/api/projects/{PID}/plan/analysis"
|
|
before = client.get(url).json()
|
|
moved = (310.0, 400.0)
|
|
_analyze_moved(client, tmp_path, monkeypatch, moved)
|
|
folder = analyze.derived_dir(tmp_path)
|
|
assert (folder / "route_2026.json").is_file() and (folder / "route_2027.json").is_file()
|
|
|
|
assert client.get(url).json() == before # 저장 지점 = 옛 분석 그대로
|
|
view = client.get(url, params={"points": json.dumps(_points(moved))}).json()
|
|
assert [y["analysis"]["fingerprint"] for y in view["years"]] == ["g", "g"]
|
|
assert view["chain"][1]["end_chainage_m"] == 2400.0
|
|
|
|
# [저장] = 끈 지점 정본 → 저장 지점 GET 이 그 분석
|
|
body = {"choices": {"2027": "gentle"}, "points": _points(moved)}
|
|
assert client.put(url, json=body).status_code == 200
|
|
for seg, (start, end) in zip(client.segments, ((P0, moved), (moved, P2))):
|
|
seg.update(start=start, end=end)
|
|
after = client.get(url).json()
|
|
assert [y["analysis"]["fingerprint"] for y in after["years"]] == ["g", "g"]
|
|
|
|
# 쌓임 — 저장 지점 것은 가장 옛것이어도 남음 · 연도마다 최근 KEEP_PER_YEAR 개
|
|
for path in folder.glob("route_*_*.json"):
|
|
os.utime(path, (1000.0, 1000.0))
|
|
for n in range(analyze.KEEP_PER_YEAR + 1):
|
|
_analyze_moved(client, tmp_path, monkeypatch, (320.0 + n, 400.0))
|
|
keyed = sorted(p.name for p in folder.glob("route_2026_*.json"))
|
|
assert len(keyed) == analyze.KEEP_PER_YEAR + 1
|
|
assert analyze.segment_path(tmp_path, client.segments[0]).name in keyed
|
|
assert client.get(url).json() == after
|
|
|
|
|
|
def test_terrain_extent_uses_whole_ends(client) -> None:
|
|
got = client.get(f"/api/projects/{PID}/plan/terrain/extent?margin_m=400").json()
|
|
m = 400.0
|
|
assert got["margin_m"] == m and got["have"] is None and got["covered"] is False
|
|
# 전체 시점 = 첫 연도 시점 · 전체 종점 = 끝 연도 종점(연도 구간 아님)
|
|
assert got["box_m"] == [P0[0] - m, P0[1] - m, P2[0] + m, P2[1] + m]
|
|
extra = client.get(
|
|
f"/api/projects/{PID}/plan/terrain/extent", params={"extra": json.dumps([[2000, 0]])}
|
|
).json()
|
|
assert extra["box_m"][2] == 2000 + extra["margin_m"]
|
|
bad = client.get(f"/api/projects/{PID}/plan/terrain/extent", params={"extra": "[1]"})
|
|
assert bad.status_code == 400
|
|
|
|
|
|
def test_terrain_post_runs_one_set(client, monkeypatch) -> None:
|
|
import asyncio
|
|
|
|
calls: list = []
|
|
saved: list = []
|
|
works: list = []
|
|
|
|
def ensure(root, points, epsg, margin_m, report):
|
|
calls.append((points, epsg, margin_m))
|
|
return {"sheets": {"downloaded": ["1"]}, "models": [{"model_file_path": "m"}]}
|
|
|
|
async def save(project_id, models, epsg):
|
|
saved.append((models, epsg))
|
|
|
|
def start(project_id, kind, work):
|
|
works.append(work)
|
|
return {"status": "accepted", "kind": kind}
|
|
|
|
def post(body):
|
|
res = client.post(f"/api/projects/{PID}/plan/terrain", json=body)
|
|
asyncio.run(works.pop()(lambda percent, message: None))
|
|
return res
|
|
|
|
monkeypatch.setattr(router_mod.terrain, "ensure_plan_terrain", ensure)
|
|
monkeypatch.setattr(router_mod, "_save_models", save)
|
|
monkeypatch.setattr(router_mod, "_start_job", start)
|
|
res = post({"margin_m": 500, "extra": [[5, 6]]})
|
|
assert res.json() == {"status": "accepted", "kind": "terrain"}
|
|
assert calls == [([P0, P2, (5.0, 6.0)], 5186, 500.0)]
|
|
assert saved == [([{"model_file_path": "m"}], 5186)]
|
|
# 옛 화면 꼴(points 만 · 빈 몸) — 여유 처음 값 · 전체 시점 · 종점
|
|
post({"points": []})
|
|
assert calls[-1] == ([P0, P2], 5186, None)
|