eomsangdonandClaude Opus 5.5
4753d88983
- 분석본 = route_<연도>_<지점 열쇠>.json · GET 은 DB 지점 분석본 · 끈 지점 GET 은 그 분석 - [저장] 뒤 정본 = 끈 지점 분석본(DB 지점 따라감) - 쌓임 거르기 = 연도마다 최근 3 개 + 저장 지점 것 - 옛 꼴 route_<연도>.json 은 읽기만(공유 storage · 임시 배포 옛 판 서버가 읽음 — 옮기지 않음) - 시험 test_530_2 끈 지점 분석 · 저장 · 쌓임 Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
304 lines
13 KiB
Python
304 lines
13 KiB
Python
"""D 분석 API(PLAN 530-2 · 530-3) — 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
|