eomsangdonandClaude Opus 5
c59dbba482
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_016VBGFXB9AbJBwXP19z75Qq
296 lines
12 KiB
Python
296 lines
12 KiB
Python
"""B05 구조물 라우터 테스트 (지적 5 반영: 404·STALE 실패 정합 포함)."""
|
|
|
|
import json
|
|
|
|
import pytest
|
|
from fastapi import FastAPI
|
|
from fastapi.testclient import TestClient
|
|
|
|
import B05_Profile.B05_Profile_Structures_Router as router_module
|
|
from common_util.common_util_drainage_pipes import pipe_points_path_in
|
|
|
|
PROJECT_ID = "11111111-1111-1111-1111-111111111111"
|
|
|
|
|
|
@pytest.fixture()
|
|
def invalidated_calls():
|
|
return []
|
|
|
|
|
|
def _make_client(tmp_path, monkeypatch, invalidated_calls, *, stale_ok=True, route_length=None):
|
|
async def fake_project_root(project_id):
|
|
return str(tmp_path / "project")
|
|
|
|
async def fake_invalidate(project_id):
|
|
invalidated_calls.append(str(project_id))
|
|
return stale_ok
|
|
|
|
async def fake_route_length(project_id):
|
|
return route_length
|
|
|
|
monkeypatch.setattr(router_module, "_project_root", fake_project_root)
|
|
monkeypatch.setattr(router_module, "_invalidate_downstream", fake_invalidate)
|
|
monkeypatch.setattr(router_module, "_route_length", fake_route_length)
|
|
app = FastAPI()
|
|
app.include_router(router_module.router)
|
|
return TestClient(app)
|
|
|
|
|
|
@pytest.fixture()
|
|
def client(tmp_path, monkeypatch, invalidated_calls):
|
|
return _make_client(tmp_path, monkeypatch, invalidated_calls)
|
|
|
|
|
|
def _point_payload(base_revision=0, chainage=100.0, type_id="erosion_check"):
|
|
return {
|
|
"base_revision": base_revision,
|
|
"structures": [
|
|
{
|
|
"type_id": type_id,
|
|
"placement": "point",
|
|
"chainage_m": chainage,
|
|
"options": {"form": "돌"},
|
|
}
|
|
],
|
|
}
|
|
|
|
|
|
def test_structure_types_endpoint_returns_registry(client):
|
|
response = client.get("/api/projects/structure-types")
|
|
assert response.status_code == 200
|
|
body = response.json()
|
|
assert body["schema_version"] >= 1
|
|
assert len(body["types"]) >= 30
|
|
# 2026-08-19 C군 개편: 높이는 사용자 확정 기본값 2.5를 갖고 필수 강제는 풀렸다
|
|
# (기본값 원칙 ② — 구 저장분 저장 거부 재발 방지). 상세 정책은 registry_policy 테스트.
|
|
wall = next(item for item in body["types"] if item["type_id"] == "retaining_wall")
|
|
height = next(option for option in wall["options"] if option["key"] == "height_m")
|
|
# 2026-09-14 A4 — 옹벽 제안값은 자료가 있는 2.0(반중력식 H=2.0 한 벌뿐).
|
|
assert height["required"] is False and height["default"] == 2.0
|
|
|
|
|
|
def test_read_structures_empty_project(client):
|
|
response = client.get(f"/api/projects/{PROJECT_ID}/route/structures")
|
|
assert response.status_code == 200
|
|
assert response.json()["revision"] == 0
|
|
|
|
|
|
def test_save_then_read_roundtrip(client):
|
|
saved = client.put(f"/api/projects/{PROJECT_ID}/route/structures", json=_point_payload())
|
|
assert saved.status_code == 200
|
|
assert saved.json()["revision"] == 1
|
|
read = client.get(f"/api/projects/{PROJECT_ID}/route/structures")
|
|
assert read.json()["structures"][0]["structure_id"]
|
|
|
|
|
|
def test_stale_revision_returns_409(client):
|
|
client.put(f"/api/projects/{PROJECT_ID}/route/structures", json=_point_payload())
|
|
conflict = client.put(
|
|
f"/api/projects/{PROJECT_ID}/route/structures", json=_point_payload(base_revision=0)
|
|
)
|
|
assert conflict.status_code == 409
|
|
assert conflict.json()["revision"] == 1
|
|
|
|
|
|
def test_pipe_type_returns_400(client):
|
|
response = client.put(
|
|
f"/api/projects/{PROJECT_ID}/route/structures", json=_point_payload(type_id="pipe")
|
|
)
|
|
assert response.status_code == 400
|
|
|
|
|
|
def test_invalid_placement_returns_422(client):
|
|
response = client.put(
|
|
f"/api/projects/{PROJECT_ID}/route/structures",
|
|
json={
|
|
"base_revision": 0,
|
|
"structures": [
|
|
{
|
|
"type_id": "ditch_side",
|
|
"placement": "interval",
|
|
"start_m": 50.0,
|
|
"end_m": 10.0,
|
|
"options": {"form": "L형"},
|
|
}
|
|
],
|
|
},
|
|
)
|
|
assert response.status_code == 422
|
|
|
|
|
|
def test_registry_placement_mismatch_returns_400(client):
|
|
"""레지스트리와 다른 배치형태(옹벽을 point로)는 400."""
|
|
response = client.put(
|
|
f"/api/projects/{PROJECT_ID}/route/structures",
|
|
json=_point_payload(type_id="retaining_wall"),
|
|
)
|
|
assert response.status_code == 400
|
|
|
|
|
|
def test_missing_project_returns_404(tmp_path, monkeypatch, invalidated_calls):
|
|
"""없는 프로젝트 — 저장 경로 조회가 LookupError를 던져도 404로 응답해야 한다."""
|
|
client = _make_client(tmp_path, monkeypatch, invalidated_calls)
|
|
|
|
async def raising_project_root(project_id):
|
|
raise LookupError("프로젝트 또는 프로젝트 저장 경로를 찾을 수 없습니다.")
|
|
|
|
monkeypatch.setattr(router_module, "_project_root", raising_project_root)
|
|
assert client.get(f"/api/projects/{PROJECT_ID}/route/structures").status_code == 404
|
|
assert (
|
|
client.put(
|
|
f"/api/projects/{PROJECT_ID}/route/structures", json=_point_payload()
|
|
).status_code
|
|
== 404
|
|
)
|
|
|
|
|
|
def test_design_change_invalidates_downstream(client, invalidated_calls):
|
|
client.put(f"/api/projects/{PROJECT_ID}/route/structures", json=_point_payload())
|
|
moved = client.put(
|
|
f"/api/projects/{PROJECT_ID}/route/structures",
|
|
json=_point_payload(base_revision=1, chainage=180.0),
|
|
)
|
|
assert moved.json()["invalidated_downstream"] is True
|
|
assert invalidated_calls == [PROJECT_ID, PROJECT_ID]
|
|
|
|
|
|
def test_unchanged_save_does_not_invalidate_downstream(client, invalidated_calls):
|
|
saved = client.put(f"/api/projects/{PROJECT_ID}/route/structures", json=_point_payload())
|
|
stored = client.get(f"/api/projects/{PROJECT_ID}/route/structures").json()["structures"]
|
|
invalidated_calls.clear()
|
|
again = client.put(
|
|
f"/api/projects/{PROJECT_ID}/route/structures",
|
|
json={"base_revision": saved.json()["revision"], "structures": stored},
|
|
)
|
|
assert again.json()["invalidated_downstream"] is False
|
|
assert invalidated_calls == []
|
|
|
|
|
|
def test_stale_failure_is_not_reported_as_success(tmp_path, monkeypatch, invalidated_calls):
|
|
"""STALE 갱신이 실패하면 invalidated_downstream=false — 성공한 척 금지(지적 5).
|
|
|
|
화면이 "필요했는데 못 했다"를 구분해 알릴 수 있도록 needs 플래그도 함께 준다
|
|
(2차 크로스체크 지적 3).
|
|
"""
|
|
client = _make_client(tmp_path, monkeypatch, invalidated_calls, stale_ok=False)
|
|
response = client.put(f"/api/projects/{PROJECT_ID}/route/structures", json=_point_payload())
|
|
body = response.json()
|
|
assert response.status_code == 200 # 저장 자체는 성공
|
|
assert body["invalidated_downstream"] is False
|
|
assert body["needs_downstream_invalidation"] is True
|
|
|
|
|
|
def test_unchanged_save_needs_no_invalidation_flag(tmp_path, monkeypatch, invalidated_calls):
|
|
"""설계 영향이 없으면 needs 플래그도 false — 실패 안내가 뜨지 않아야 한다."""
|
|
client = _make_client(tmp_path, monkeypatch, invalidated_calls, stale_ok=False)
|
|
saved = client.put(f"/api/projects/{PROJECT_ID}/route/structures", json=_point_payload())
|
|
stored = client.get(f"/api/projects/{PROJECT_ID}/route/structures").json()["structures"]
|
|
again = client.put(
|
|
f"/api/projects/{PROJECT_ID}/route/structures",
|
|
json={"base_revision": saved.json()["revision"], "structures": stored},
|
|
)
|
|
assert again.json()["needs_downstream_invalidation"] is False
|
|
|
|
|
|
def test_route_length_limits_position(tmp_path, monkeypatch, invalidated_calls):
|
|
"""노선 연장이 있으면 범위 밖 배치는 400."""
|
|
client = _make_client(tmp_path, monkeypatch, invalidated_calls, route_length=500.0)
|
|
ok = client.put(
|
|
f"/api/projects/{PROJECT_ID}/route/structures", json=_point_payload(chainage=499.0)
|
|
)
|
|
assert ok.status_code == 200
|
|
bad = client.put(
|
|
f"/api/projects/{PROJECT_ID}/route/structures",
|
|
json=_point_payload(base_revision=1, chainage=999.0),
|
|
)
|
|
assert bad.status_code == 400
|
|
|
|
|
|
# ── 구 비정규 측점 이관 (2026-08-17 컨테이너 병합 3단계) ────────────────────
|
|
|
|
|
|
def _migrate_payload():
|
|
return {
|
|
"stations": [
|
|
{"chainage_m": 60.0, "structure": "파형강관 D800"}, # 배관 — 제외돼야 한다
|
|
{"chainage_m": 120.0, "structure": "기성막이"},
|
|
{"chainage_m": 200.0, "structure": "대피로 2.0m"},
|
|
{"chainage_m": 260.0, "structure": "낙석방지책"}, # 기타
|
|
]
|
|
}
|
|
|
|
|
|
def _pipe_points_file(tmp_path, points=()):
|
|
"""관 지점 정본을 미리 깔아 둔다 — 기슭막이 이관은 이 파일이 있어야 나간다."""
|
|
path = pipe_points_path_in(tmp_path / "project")
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
path.write_text(
|
|
json.dumps({"route_signature": "test-signature", "points": list(points)}),
|
|
encoding="utf-8",
|
|
)
|
|
return path
|
|
|
|
|
|
def test_migrate_converts_legacy_stations(client, tmp_path):
|
|
"""기슭막이는 관 지점 정본으로, 나머지는 구조물 정본으로 갈린다(2026-09-01 확정)."""
|
|
path = _pipe_points_file(tmp_path)
|
|
response = client.post(
|
|
f"/api/projects/{PROJECT_ID}/route/structures/migrate", json=_migrate_payload()
|
|
)
|
|
assert response.status_code == 200
|
|
body = response.json()
|
|
assert body["migrated"] == 3 # 배관 제외 (구조물 2 + 관 시설 1)
|
|
assert body["pipe_facilities"] == 1
|
|
stored = client.get(f"/api/projects/{PROJECT_ID}/route/structures").json()["structures"]
|
|
type_ids = sorted(item["type_id"] for item in stored)
|
|
assert type_ids == ["etc", "refuge"] # 기슭막이는 구조물 목록에 없다
|
|
(pipe,) = json.loads(path.read_text(encoding="utf-8"))["points"]
|
|
assert pipe["facility"] == "revetment" and pipe["chainage_m"] == 120.0
|
|
# 구간·제원은 레지스트리 기본값 승계 — 기준측점 전/후 5m.
|
|
assert (pipe["start_m"], pipe["end_m"]) == (115.0, 125.0)
|
|
# 높이는 설계자 입력 — 등록부 기본 2.5 걷음(2026-09-14 브레인 판정)
|
|
assert pipe["options"]["form"] == "돌쌓기(메)" and "height_m" not in pipe["options"]
|
|
|
|
|
|
def test_migrate_without_pipe_points_defers_revetment(client):
|
|
"""관 지점 정본이 없으면 기슭막이만 미룬다 — 원천 측점이 남아 다음에 다시 이관된다."""
|
|
response = client.post(
|
|
f"/api/projects/{PROJECT_ID}/route/structures/migrate", json=_migrate_payload()
|
|
)
|
|
assert response.status_code == 200
|
|
assert response.json()["migrated"] == 2 and response.json()["pipe_facilities"] == 0
|
|
stored = client.get(f"/api/projects/{PROJECT_ID}/route/structures").json()["structures"]
|
|
assert sorted(item["type_id"] for item in stored) == ["etc", "refuge"]
|
|
|
|
|
|
def test_migrate_is_idempotent(client, tmp_path):
|
|
path = _pipe_points_file(tmp_path)
|
|
client.post(f"/api/projects/{PROJECT_ID}/route/structures/migrate", json=_migrate_payload())
|
|
again = client.post(
|
|
f"/api/projects/{PROJECT_ID}/route/structures/migrate", json=_migrate_payload()
|
|
)
|
|
assert again.status_code == 200
|
|
assert again.json()["migrated"] == 0
|
|
stored = client.get(f"/api/projects/{PROJECT_ID}/route/structures").json()["structures"]
|
|
assert len(stored) == 2
|
|
assert len(json.loads(path.read_text(encoding="utf-8"))["points"]) == 1
|
|
|
|
|
|
def test_migrate_keeps_existing_structures(client, tmp_path):
|
|
_pipe_points_file(tmp_path)
|
|
client.put(f"/api/projects/{PROJECT_ID}/route/structures", json=_point_payload())
|
|
response = client.post(
|
|
f"/api/projects/{PROJECT_ID}/route/structures/migrate", json=_migrate_payload()
|
|
)
|
|
assert response.json()["migrated"] == 3
|
|
stored = client.get(f"/api/projects/{PROJECT_ID}/route/structures").json()["structures"]
|
|
assert len(stored) == 3 # 기존 1 + 이관 2(기슭막이는 관 지점으로)
|
|
|
|
|
|
def test_migrate_empty_list_is_noop(client):
|
|
response = client.post(
|
|
f"/api/projects/{PROJECT_ID}/route/structures/migrate", json={"stations": []}
|
|
)
|
|
assert response.status_code == 200
|
|
assert response.json()["migrated"] == 0
|