Aislo/resources/tester/test_452_server_keys_prune_migrate.py

# -*- coding: utf-8 -*-
"""452 서버 — 구 구조물 이관 · 옛 횡단 파일 · 시종점 열쇠 (433-17① · 433-10 · 433-18).

못박는 것
  - 이관 원천 = 저장 종단 파일의 비정규 측점 — 화면 본문(초안 관 자리 섞임)은 안 믿음 ·
    저장 관 자리면 structures.json · pipe_points.json 그대로.
  - 관 PUT 끝 측점 다시 뜸(`_regenerate_structure_stations`) = 옛 자리 cross_*.json 도 지움.
  - 시종점 열쇠 한 규칙 — 맨 앞 · 맨 끝은 구조물이 얹혀도(병합이 종류를 irregular 로 갈음해도)
    `end:BP/EP` · 구간형 시작 · 끝 = `:start` · `:end` · 서버 `station_keys` = B06 `stationKeysOf`.
"""

import asyncio
import json
import shutil
import subprocess
import sys
from pathlib import Path
from types import SimpleNamespace

import pytest
from fastapi import FastAPI
from fastapi.testclient import TestClient

HERE = Path(__file__).resolve().parent
ROOT = HERE.parents[1]
if str(ROOT) not in sys.path:
    sys.path.insert(0, str(ROOT))

import B05_Profile.B05_Profile_Structures_Router as router_module  # noqa: E402
import B06_Section.B06_Section_Repository as b06_repo  # noqa: E402
from B05_Profile.B05_Profile_Engine_Sections import cross_filename  # noqa: E402
from B05_Profile.B05_Profile_Route_Save_Parts import station_keys  # noqa: E402
from common_util.common_util_drainage_pipes import pipe_points_path_in  # noqa: E402

PROJECT_ID = "45245245-2452-4524-5245-245245245245"
LONG_REL = "B06_Section/longitudinal/longitudinal.json"


def _write_longitudinal(project: Path, stations):
    path = project / LONG_REL
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(json.dumps({"stations": stations}), encoding="utf-8")
    return path


# ── 433-17① 이관 = 저장 종단 파일 ─────────────────────────────────────────


@pytest.fixture()
def migrate_client(tmp_path, monkeypatch):
    project = tmp_path / "project"

    async def fake_root(project_id):
        return str(project)

    async def fake_length(project_id):
        return 400.0

    async def fake_latest(connection, project_id):
        return {"id": 7}

    async def fake_longitudinal(connection, project_id, route_id):
        assert route_id == 7
        return {"longitudinal_file_path": LONG_REL}

    class _Acquire:
        async def __aenter__(self):
            return object()

        async def __aexit__(self, *exc):
            return False

    monkeypatch.setattr(router_module, "_project_root", fake_root)
    monkeypatch.setattr(router_module, "_route_length", fake_length)
    monkeypatch.setattr(router_module, "get_latest_route", fake_latest)
    monkeypatch.setattr(router_module, "get_db_pool", lambda: SimpleNamespace(acquire=_Acquire))
    monkeypatch.setattr(b06_repo, "get_longitudinal_section", fake_longitudinal)
    app = FastAPI()
    app.include_router(router_module.router)
    return TestClient(app), project


def test_초안_측점_본문은_버리고_저장_관_자리는_이관_안_함(migrate_client):
    client, project = migrate_client
    pipes = pipe_points_path_in(project)
    pipes.parent.mkdir(parents=True, exist_ok=True)
    pipes.write_text(
        json.dumps({"route_signature": "s", "points": [{"chainage_m": 351.0, "id": "p1"}]}),
        encoding="utf-8",
    )
    _write_longitudinal(
        project,
        [
            {"chainage_m": 0.0, "kind": "bp"},
            {"chainage_m": 351.0, "kind": "irregular", "structure": "파형강관 D800"},
            {"chainage_m": 400.0, "kind": "ep"},
        ],
    )
    before = pipes.read_bytes()
    # 화면이 초안(관 351 → 364) 상세를 들고 와도 — 옛 본문 꼴을 보내 봄
    body = {"route_id": 7, "stations": [{"chainage_m": 364.0, "structure": "낙석방지책"}]}
    response = client.post(f"/api/projects/{PROJECT_ID}/route/structures/migrate", json=body)
    assert response.status_code == 200, response.text
    assert response.json()["migrated"] == 0
    assert not (project / "B05_Profile" / "route" / "structures.json").exists()
    assert pipes.read_bytes() == before


def test_저장_종단의_옛_자유글_측점은_이관(migrate_client):
    client, project = migrate_client
    _write_longitudinal(
        project,
        [
            {"chainage_m": 0.0, "kind": "bp"},
            {"chainage_m": 260.0, "kind": "irregular", "structure": "낙석방지책"},
            {"chainage_m": 400.0, "kind": "ep"},
        ],
    )
    response = client.post(
        f"/api/projects/{PROJECT_ID}/route/structures/migrate", json={"route_id": None}
    )
    assert response.status_code == 200, response.text
    assert response.json()["migrated"] == 1
    stored = client.get(f"/api/projects/{PROJECT_ID}/route/structures").json()["structures"]
    assert [(item["type_id"], item["chainage_m"]) for item in stored] == [("etc", 260.0)]


def test_저장_종단이_없으면_아무것도_안_함(migrate_client):
    client, project = migrate_client
    response = client.post(f"/api/projects/{PROJECT_ID}/route/structures/migrate", json={})
    assert response.status_code == 200 and response.json()["migrated"] == 0
    assert not (project / "B05_Profile" / "route" / "structures.json").exists()


# ── 433-10 관 PUT 측점 다시 뜸 = 옛 cross 파일 지움 ─────────────────────────


def test_관_옮김_측점_다시_뜸이_옛_자리_횡단_파일을_지움(tmp_path, monkeypatch):
    import B05_Profile.B05_Profile_Engine_Sections as engine
    import B06_Section.B06_Section_Router_Stations as stations_module

    _write_longitudinal(
        tmp_path,
        [
            {"chainage_m": 0.0, "kind": "bp"},
            {"chainage_m": 351.0, "kind": "irregular", "structure": "파형강관 D800"},
            {"chainage_m": 400.0, "kind": "ep"},
        ],
    )
    cross_dir = tmp_path / "B06_Section" / "cross_sections"
    cross_dir.mkdir(parents=True)
    for at in (0.0, 351.0, 400.0):
        (cross_dir / cross_filename(at)).write_text("{}", encoding="utf-8")

    def fake_generate(*args, **kwargs):
        (cross_dir / cross_filename(361.0)).write_text("{}", encoding="utf-8")
        return [{"chainage_m": 361.0, "kind": "irregular", "structure": "파형강관 D800"}]

    async def fake_inputs(project_id, route, snapshot):
        return None, None, {"longitudinal_file_path": LONG_REL}

    async def fake_run(fn, *args):
        return None  # DB 행 맞춤 · 등급 요약은 이 시험 밖

    monkeypatch.setattr(engine, "_load_route_polyline", lambda *a: [[0.0, 0.0], [400.0, 0.0]])
    monkeypatch.setattr(engine, "_load_pipe_points", lambda *a: [])
    monkeypatch.setattr(engine, "resolve_extra_stations", lambda *a: [])
    monkeypatch.setattr(engine, "generate_irregular_sections", fake_generate)
    monkeypatch.setattr(stations_module, "_generation_inputs", fake_inputs)
    monkeypatch.setattr(stations_module, "run_with_connection", fake_run)

    made = asyncio.run(
        stations_module._regenerate_structure_stations(
            PROJECT_ID,
            tmp_path,
            {"id": 7, "route_data_path": "route.json"},
            {"filter_key": "f", "method": "m"},
        )
    )
    assert made == 1
    assert sorted(p.name for p in cross_dir.glob("cross_*.json")) == sorted(
        cross_filename(at) for at in (0.0, 361.0, 400.0)
    )


# ── 433-18 시종점 열쇠 한 규칙 ─────────────────────────────────────────────


STRUCTURES = [
    {"structure_id": "w1", "type_id": "retaining_wall", "chainage_m": 80.0, "start_m": 60.0},
]
# 구간형 옹벽 끝 = 노선 끝(EP) — 병합이 EP 행 종류를 irregular 로 갈음한 저장분 꼴
SECTIONS = [
    {"chainage_m": 0.0, "kind": "bp"},
    {"chainage_m": 20.0, "kind": "regular"},
    {"chainage_m": 40.0, "kind": "regular"},
    {"chainage_m": 60.0, "kind": "irregular", "structure": "옹벽"},
    {"chainage_m": 80.0, "kind": "irregular", "structure": "옹벽"},
    {"chainage_m": 100.0, "kind": "irregular", "structure": "옹벽"},
]


def _structures_ns():
    return [SimpleNamespace(**item, end_m=100.0, options={}) for item in STRUCTURES]


def test_서버_시종점_열쇠가_구조물보다_먼저():
    keys = station_keys(SECTIONS, [], _structures_ns(), 20.0)
    assert keys == {
        "end:BP": 0.0,
        "grid:1": 20.0,
        "grid:2": 40.0,
        "structure:w1:start": 60.0,
        "structure:w1": 80.0,
        "end:EP": 100.0,
    }


def test_B06_열쇠가_서버와_같음(tmp_path):
    work = tmp_path / "b452"
    build = subprocess.run(
        [
            "node",
            str(ROOT / "config" / "node_modules" / "vite" / "bin" / "vite.js"),
            "build",
            "--configLoader",
            "runner",
            "--ssr",
            "../resources/tester/helper_452_entry.ts",
            "--outDir",
            str(work),
            "--logLevel",
            "error",
        ],
        cwd=ROOT,
        capture_output=True,
        text=True,
        encoding="utf-8",
        timeout=300,
    )
    assert build.returncode == 0, build.stderr
    bundle = work / "bundle.mjs"
    shutil.copy(work / "helper_452_entry.js", bundle)
    source = work / "input.json"
    source.write_text(
        json.dumps(
            {
                "sections": SECTIONS,
                "structures": [{**item, "end_m": 100.0} for item in STRUCTURES],
                "interval": 20.0,
            }
        ),
        encoding="utf-8",
    )
    proc = subprocess.run(
        ["node", str(HERE / "helper_452_keys.mjs"), str(bundle), str(source)],
        cwd=ROOT,
        capture_output=True,
        text=True,
        encoding="utf-8",
        timeout=120,
    )
    assert proc.returncode == 0, proc.stderr
    ts = {entry["key"]: entry["chainageM"] for entry in json.loads(proc.stdout)}
    assert ts == station_keys(SECTIONS, [], _structures_ns(), 20.0)


def test_노선_저장_이월_EP_고침이_새_EP_에_남음():
    from B05_Profile.B05_Profile_Route_Save_Parts import carry_cross_inputs

    old_keys = station_keys(SECTIONS, [], _structures_ns(), 20.0)
    # 노선이 늘어 옹벽이 따라감 — 끝(EP) 110 · 시작 70 · 기준 90
    moved = [
        SimpleNamespace(
            structure_id="w1",
            type_id="retaining_wall",
            chainage_m=90.0,
            start_m=70.0,
            end_m=110.0,
            options={},
        )
    ]
    new_sections = [
        {"chainage_m": at, "kind": "regular"} for at in (0.0, 20.0, 40.0, 60.0, 70.0, 80.0, 90.0)
    ] + [{"chainage_m": 100.0, "kind": "regular"}, {"chainage_m": 110.0, "kind": "irregular"}]
    new_keys = station_keys(new_sections, [], moved, 20.0)
    designs = {100.0: {"ground_type": "rock"}, 60.0: {"ground_type": "weathered"}}
    inputs, _, dropped = carry_cross_inputs(old_keys, new_keys, designs, ("ground_type",), ())
    assert inputs[110.0] == {"ground_type": "rock"}  # EP 고침 = 새 EP
    assert inputs[70.0] == {"ground_type": "weathered"}  # 구간 시작 고침 = 따라간 시작
    assert dropped == 0