Merge remote-tracking branch 'origin/sub_laptop_3' into main_laptop_1
This commit is contained in:
@@ -1,4 +1,5 @@
|
||||
"""D 분석 API(PLAN 530-2) — GET 이은 누가 · PUT [저장] = 서버 저장본으로 다시 셈 · 모르는 갈래 거절.
|
||||
"""D 분석 API(PLAN 530-2 · 530-3) — GET 이은 누가 · PUT [저장] = 서버 저장본으로 다시 셈 · 모르는 갈래
|
||||
거절 · 끈 지점(points) = 그 지점으로 보기 · 안 이어지면 거절 · [저장] 때만 지점 정본(산림청만).
|
||||
|
||||
./venv/Scripts/python.exe -m pytest resources/tester/test_530_2_plan_router.py -q
|
||||
"""
|
||||
@@ -9,6 +10,7 @@ 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_Router as router_mod
|
||||
@@ -86,16 +88,26 @@ def client(tmp_path, monkeypatch):
|
||||
)
|
||||
project = {"road_type": "work", "design_speed_kph": None, "route_start_m": 1000.0}
|
||||
|
||||
async def inputs(project_id):
|
||||
return tmp_path, project, segments
|
||||
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)
|
||||
test = TestClient(app)
|
||||
test.log = log
|
||||
test.session = session
|
||||
test.stored = stored
|
||||
return test
|
||||
|
||||
|
||||
@@ -125,3 +137,39 @@ def test_put_rejects_unknown_or_empty_mode(client) -> None:
|
||||
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
|
||||
|
||||
@@ -0,0 +1,142 @@
|
||||
"""D03 화면 계산 거울(PLAN 530-3) — 파이썬 `D03_PlanAnalysis_Analyze` ↔ TS `logic/D03_PlanAnalysis_Plan.ts`.
|
||||
|
||||
짝: `pick_mode` · `chain` · `joined_line` ↔ `pickMode` · `chainRows` · `joinedLine`.
|
||||
TS 만: 측점 누가(경계 = 측점 · 간격 배수) · 앞붙임 선(선 위 누가 = 등록 시작 누가) · 받은 지형 판정.
|
||||
|
||||
./venv/Scripts/python.exe -m pytest resources/tester/test_530_3_plan_logic.py -q
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import re
|
||||
import subprocess
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
ROOT = Path(__file__).resolve().parents[2]
|
||||
if str(ROOT) not in sys.path:
|
||||
sys.path.insert(0, str(ROOT))
|
||||
|
||||
from D03_PlanAnalysis import D03_PlanAnalysis_Analyze as analyze # noqa: E402
|
||||
|
||||
TS_FILE = ROOT / "D03_PlanAnalysis" / "logic" / "D03_PlanAnalysis_Plan.ts"
|
||||
TSC = ROOT / "config" / "node_modules" / "typescript" / "bin" / "tsc"
|
||||
|
||||
_RUNNER = """
|
||||
import { readFileSync, writeFileSync } from "node:fs";
|
||||
import * as plan from "./D03_PlanAnalysis/logic/D03_PlanAnalysis_Plan.js";
|
||||
const job = JSON.parse(readFileSync(process.argv[2], "utf8"));
|
||||
const out = job.cases.map((c) => {
|
||||
const rows = plan.chainRows(job.results, c.choices, c.start);
|
||||
return {
|
||||
picks: job.results.map((r) => plan.pickMode(r, c.choices[r.year])),
|
||||
rows,
|
||||
line: plan.joinedLine(job.results, rows),
|
||||
stations: plan.stationChainages(rows, 20),
|
||||
};
|
||||
});
|
||||
const lead = plan.leadInLine([[0, 0], [3, 4], [10, 4]], 1234.5);
|
||||
const reach = plan.nodesInReach(
|
||||
[[500, 500], [900, 500], [1700, 500]],
|
||||
[{ x0: 0, y0: 0, x1: 1300, y1: 1000 }, null],
|
||||
);
|
||||
writeFileSync(process.argv[3], JSON.stringify({ out, lead, reach }));
|
||||
"""
|
||||
|
||||
_P0, _P1, _P2 = [0.0, 0.0], [300.0, 400.0], [900.0, 400.0]
|
||||
_RESULTS = [
|
||||
{
|
||||
"year": 2026,
|
||||
"start": _P0,
|
||||
"end": _P1,
|
||||
"routes": {
|
||||
"gradeline": {"planned": [_P0, [100.0, 200.0], _P1], "metrics": {"length_m": 530.0}},
|
||||
"earthwork": {"error": "길 없음"},
|
||||
"gentle": {"planned": [_P0, _P1], "metrics": {"length_m": 500.004}},
|
||||
},
|
||||
},
|
||||
{
|
||||
"year": 2027,
|
||||
"start": _P1,
|
||||
"end": _P2,
|
||||
"routes": {
|
||||
"gradeline": {"error": "길 없음"},
|
||||
"gentle": {"planned": [_P1, [600.0, 500.0], _P2], "metrics": {"length_m": 640.333}},
|
||||
"weighted": {"planned": [_P1, _P2], "metrics": {"length_m": 600.0}},
|
||||
},
|
||||
},
|
||||
]
|
||||
_CASES = [
|
||||
{"choices": {}, "start": 1000.0},
|
||||
{"choices": {"2026": "gentle", "2027": "weighted"}, "start": 17.25},
|
||||
{"choices": {"2026": "earthwork", "2027": "follow"}, "start": 0.0},
|
||||
]
|
||||
|
||||
|
||||
def _run_ts(tmp_path: Path) -> dict:
|
||||
out = tmp_path / "js"
|
||||
subprocess.run( # noqa: S603 — 고정 실행 파일
|
||||
[
|
||||
"node",
|
||||
str(TSC),
|
||||
str(TS_FILE),
|
||||
"--outDir",
|
||||
str(out),
|
||||
"--rootDir",
|
||||
str(ROOT),
|
||||
"--module",
|
||||
"esnext",
|
||||
"--target",
|
||||
"es2022",
|
||||
"--moduleResolution",
|
||||
"bundler",
|
||||
"--ignoreConfig",
|
||||
],
|
||||
cwd=str(ROOT),
|
||||
check=True,
|
||||
capture_output=True,
|
||||
)
|
||||
for emitted in out.rglob("*.js"):
|
||||
emitted.write_text(
|
||||
re.sub(r'(from\s+"\.\.?/[^"]+)"', r'\1.js"', emitted.read_text(encoding="utf-8")),
|
||||
encoding="utf-8",
|
||||
)
|
||||
(out / "runner.mjs").write_text(_RUNNER, encoding="utf-8")
|
||||
job = tmp_path / "job.json"
|
||||
result = tmp_path / "out.json"
|
||||
job.write_text(json.dumps({"results": _RESULTS, "cases": _CASES}), encoding="utf-8")
|
||||
subprocess.run( # noqa: S603
|
||||
["node", str(out / "runner.mjs"), str(job), str(result)],
|
||||
cwd=str(ROOT),
|
||||
check=True,
|
||||
capture_output=True,
|
||||
)
|
||||
return json.loads(result.read_text(encoding="utf-8"))
|
||||
|
||||
|
||||
def test_ts_matches_python_chain(tmp_path: Path) -> None:
|
||||
got = _run_ts(tmp_path)
|
||||
for case, ts in zip(_CASES, got["out"]):
|
||||
choices = {int(k): v for k, v in case["choices"].items()}
|
||||
rows = analyze.chain(_RESULTS, choices, case["start"])
|
||||
assert ts["picks"] == [analyze.pick_mode(r, choices.get(r["year"])) for r in _RESULTS]
|
||||
assert ts["rows"] == rows
|
||||
assert ts["line"] == analyze.joined_line(_RESULTS, rows)
|
||||
|
||||
# 측점 = 경계(1530) · 20 배수 · 끝 — 첫 연도 시작 1000 · 끝 2170.333
|
||||
stations = got["out"][0]["stations"]
|
||||
assert stations[:3] == [1000.0, 1020.0, 1040.0]
|
||||
assert 1530.0 in stations and 1520.0 in stations and 1540.0 in stations
|
||||
assert stations[-2:] == [2160.0, 2170.333]
|
||||
# 시작 17.25 → 첫 규칙 측점 20
|
||||
assert got["out"][1]["stations"][:2] == [17.25, 20.0]
|
||||
|
||||
|
||||
def test_lead_in_and_reach(tmp_path: Path) -> None:
|
||||
got = _run_ts(tmp_path)
|
||||
back = got["lead"][0]
|
||||
assert abs(back[0] + 1234.5 * 0.6) < 1e-9 and abs(back[1] + 1234.5 * 0.8) < 1e-9
|
||||
assert got["lead"][1:] == [[0, 0], [3, 4], [10, 4]]
|
||||
# 마디 0 = 첫 연도 안 · 마디 1 = 첫 연도 안 + 둘째 연도 DEM 없음 = 밖 · 마디 2 = 밖
|
||||
assert got["reach"] == [True, False, False]
|
||||
Reference in New Issue
Block a user