Aislo/resources/tester/test_343_2_persist_keys.py

"""343-2 도엽 3D 면 열쇠 · 지표면 모델 id 유지 · 비용면 · 추천 노선 열쇠 · 횡단 상세 ETag.

같은 입력 두 번째 = 다시 안 셈 · 열쇠 바뀌면 다시 셈 · 지울 때.
"""

from __future__ import annotations

import asyncio
import json
import os
import time
from pathlib import Path

import numpy as np
import pytest
from pyproj import Transformer

import B04_PreProcess.B04_PreProcess_Engine_SheetSurface as sheet
from B04_PreProcess.B04_PreProcess_Engine_SheetMethods import SHEET_METHOD_BUILDERS

EPSG = 5186
X0, Y0 = 200000.0, 500000.0
ROUTE = np.array([[X0 - 50.0, Y0], [X0 + 50.0, Y0]], dtype=np.float64)


@pytest.fixture()
def sheet_project(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> dict:
    processed = tmp_path / "B04_PreProcess" / "processed"
    models = tmp_path / "B04_PreProcess" / "models"
    processed.mkdir(parents=True)
    models.mkdir(parents=True)
    to_wgs84 = Transformer.from_crs(f"EPSG:{EPSG}", "EPSG:4326", always_xy=True)
    features = []
    for half, elev in ((450.0, 100.0), (300.0, 110.0), (150.0, 120.0)):
        ring = [
            list(to_wgs84.transform(x, y))
            for x, y in (
                (X0 - half, Y0 - half),
                (X0 + half, Y0 - half),
                (X0 + half, Y0 + half),
                (X0 - half, Y0 + half),
                (X0 - half, Y0 - half),
            )
        ]
        features.append(
            {
                "type": "Feature",
                "properties": {"등고수치": elev},
                "geometry": {"type": "LineString", "coordinates": ring},
            }
        )
    (processed / "도엽_등고선.geojson").write_text(
        json.dumps({"type": "FeatureCollection", "features": features}, ensure_ascii=False),
        encoding="utf-8",
    )
    calls: list[str] = []
    real = SHEET_METHOD_BUILDERS["laplace"]

    def counting(*args, **kwargs):  # type: ignore[no-untyped-def]
        calls.append("laplace")
        return real(*args, **kwargs)

    monkeypatch.setitem(SHEET_METHOD_BUILDERS, "laplace", counting)
    return {"root": tmp_path, "processed": processed, "models": models, "calls": calls}


def _build(project: dict, route: np.ndarray = ROUTE) -> list[dict]:
    return sheet.build_sheet_surface_model(
        project["root"], project["processed"], project["models"], route, EPSG, ["laplace"]
    )


def test_도엽_3D_면_같은_입력_두번째는_안_셈(sheet_project: dict) -> None:
    first = _build(sheet_project)
    assert sheet_project["calls"] == ["laplace"]
    assert (sheet_project["models"] / "dtm_sheet_laplace.key.json").is_file()
    started = time.perf_counter()
    second = _build(sheet_project)
    assert time.perf_counter() - started < 0.5
    assert sheet_project["calls"] == ["laplace"]
    assert second == first
    assert first[0]["generation_params"]["file_stamp"]


@pytest.mark.parametrize("change", ["route", "contour", "algo", "key_gone", "file_gone"])
def test_도엽_3D_면_열쇠가_바뀌면_다시_셈(
    sheet_project: dict, monkeypatch: pytest.MonkeyPatch, change: str
) -> None:
    _build(sheet_project)
    route = ROUTE
    if change == "route":
        route = ROUTE + np.array([5.0, 0.0])
    elif change == "contour":
        path = sheet_project["processed"] / "도엽_등고선.geojson"
        stat = path.stat()
        os.utime(path, ns=(stat.st_atime_ns, stat.st_mtime_ns + 5_000_000_000))
    elif change == "algo":
        monkeypatch.setattr(sheet, "SHEET_SURFACE_ALGO", sheet.SHEET_SURFACE_ALGO + 1)
    elif change == "key_gone":
        (sheet_project["models"] / "dtm_sheet_laplace.key.json").unlink()
    else:
        (sheet_project["models"] / "dtm_sheet_laplace_preview.glb").unlink()
    _build(sheet_project, route)
    assert sheet_project["calls"] == ["laplace", "laplace"]


# ── 지표면 모델 id 유지(R2) ─────────────────────────────────────────────────────


class _FakeCursor:
    def __init__(self, db: "_FakeDb") -> None:
        self.db = db
        self.rowcount = 0
        self.lastrowid = 0
        self._rows: list[tuple] = []

    async def __aenter__(self) -> "_FakeCursor":
        return self

    async def __aexit__(self, *_: object) -> None:
        return None

    async def execute(self, sql: str, params: tuple = ()) -> None:
        text = " ".join(sql.split())
        self.db.log.append(text)
        if text.startswith("SELECT id, model_type"):
            self._rows = [
                (row["id"], row["model_type"], row["path"], json.dumps(row["params"]))
                for row in self.db.models.values()
            ]
        elif text.startswith("INSERT INTO surface_models"):
            self.db.next_id += 1
            self.lastrowid = self.db.next_id
            self.db.models[self.lastrowid] = {
                "id": self.lastrowid,
                "model_type": params[1],
                "path": params[7],
                "params": json.loads(params[8]),
            }
        elif text.startswith("UPDATE surface_models"):
            self.db.models[params[-1]]["params"] = json.loads(params[4])
        elif text.startswith("DELETE FROM surface_models WHERE id IN"):
            for model_id in params:
                self.db.models.pop(model_id, None)
        elif text.startswith("DELETE FROM surface_models WHERE id ="):
            self.db.models.pop(params[0], None)
        elif text.startswith("DELETE FROM surface_models WHERE project_id"):
            for model_id in [k for k, v in self.db.models.items() if v["path"] == params[1]]:
                self.db.models.pop(model_id)
        elif text.startswith("INSERT INTO terrain_layers"):
            self.db.next_id += 1
            self.lastrowid = self.db.next_id

    async def fetchall(self) -> list[tuple]:
        return self._rows

    async def fetchone(self) -> tuple | None:
        return None


class _FakeDb:
    def __init__(self) -> None:
        self.log: list[str] = []
        self.models: dict[int, dict] = {}
        self.next_id = 100

    def cursor(self, *_: object) -> _FakeCursor:
        return _FakeCursor(self)


def _model(path: str, stamp: str) -> dict:
    return {
        "model_type": "dtm",
        "model_file_path": path,
        "resolution_m": 1.0,
        "generation_params": {"source_filter": "ptd", "file_stamp": stamp},
        "layers": [
            {
                "layer_name": "dtm_ptd_preview",
                "geometry_type": "MESH",
                "file_path": "B04_PreProcess/models/dtm_ptd_preview.glb",
                "file_format": "glb",
            }
        ],
    }


def _save(db: _FakeDb, models: list[dict]) -> list[int]:
    from B04_PreProcess import B04_PreProcess_Repository as repo

    async def fake_input(*_a):  # type: ignore[no-untyped-def]
        return {"crs_epsg": 5186}

    async def fake_cloud(*_a, **_k):  # type: ignore[no-untyped-def]
        return 9

    repo_input, repo_cloud = repo.get_input_file, repo.create_processed_point_cloud
    repo.get_input_file, repo.create_processed_point_cloud = fake_input, fake_cloud
    try:
        return asyncio.run(
            repo.save_surface_analysis_to_db(
                db,  # type: ignore[arg-type]
                project_id="00000000-0000-0000-0000-000000000001",  # type: ignore[arg-type]
                input_file_id=1,
                analysis_result={
                    "processed": {
                        "processed_file_path": "B04_PreProcess/processed/structured.npz",
                        "converted_file_path": None,
                        "point_count": 1,
                        "bounds": {},
                        "statistics": {},
                    },
                    "models": models,
                },
                source_filters=["ptd"],
            )
        )
    finally:
        repo.get_input_file, repo.create_processed_point_cloud = repo_input, repo_cloud


def test_파일_지문이_같으면_모델_id_유지() -> None:
    db = _FakeDb()
    path = "B04_PreProcess/models/dtm_ptd.npz"
    first = _save(
        db, [_model(path, "1:2"), _model("B04_PreProcess/models/implicit_ptd.npz", "3:4")]
    )
    second = _save(db, [_model(path, "1:2")])
    assert second == [first[0]]
    # 이번 결과에 없는 옛 행은 지움 · 같은 파일 행은 하나
    assert list(db.models) == [first[0]]
    third = _save(db, [_model(path, "9:9")])
    assert third != second and list(db.models) == third


def test_도엽_한_방식_요청도_id_유지() -> None:
    from B04_PreProcess import B04_PreProcess_Repository as repo

    db = _FakeDb()
    path = "B04_PreProcess/models/dtm_sheet_laplace.npz"
    entry = {**_model(path, "5:6"), "generation_params": {"source_filter": "sheet_laplace"}}
    entry["generation_params"]["file_stamp"] = "5:6"
    first = asyncio.run(repo.save_sheet_surface_models(db, "p", [entry]))  # type: ignore[arg-type]
    second = asyncio.run(repo.save_sheet_surface_models(db, "p", [entry]))  # type: ignore[arg-type]
    assert first == second and len(db.models) == 1


# ── 비용면 · 계류 메모리 기억(R17) · 추천 노선 열쇠(R18) ──────────────────────────


def test_비용면_계류_메모리_기억(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
    import B04_PreProcess.B04_PreProcess_Router_Watershed as watershed
    import B05_Profile.B05_Profile_Engine_RouteInitial as engine
    import B05_Profile.B05_Profile_Engine_Solver as solver
    import common_util.common_util_crs as crs

    loads: list[int] = []
    signature = {"value": "sig-1"}
    grid = np.arange(4.0)

    def fake_load(*_a):  # type: ignore[no-untyped-def]
        loads.append(1)
        return grid, grid, np.zeros((4, 4)), np.ones((4, 4), bool), None, None, 1.0

    monkeypatch.setattr(engine, "_TERRAINS", {})
    monkeypatch.setattr(solver, "_load_or_build_cost_surface", fake_load)
    monkeypatch.setattr(solver, "_cost_surface_signature", lambda *_a: signature["value"])
    monkeypatch.setattr(watershed, "_load_features", lambda *_a: [])
    monkeypatch.setattr(crs, "resolve_project_crs", lambda _root: "EPSG:5186")
    selection = {"source_filter": "ptd", "method": "dtm", "smooth": False}
    terrain, _ = engine.load_terrain_and_streams(tmp_path, selection)
    again, _ = engine.load_terrain_and_streams(tmp_path, selection)
    assert loads == [1] and again is terrain and terrain.signature
    signature["value"] = "sig-2"
    changed, _ = engine.load_terrain_and_streams(tmp_path, selection)
    assert loads == [1, 1] and changed.signature != terrain.signature
    # 추천 열쇠 = 서명 — 배열 통째 해시 안 함 · 서명이 다르면 다른 열쇠
    key_a = engine._pool_key([(0.0, 0.0), (3.0, 3.0)], terrain, [], 0.14)
    key_b = engine._pool_key([(0.0, 0.0), (3.0, 3.0)], changed, [], 0.14)
    assert key_a != key_b
    assert key_a == engine._pool_key([(0.0, 0.0), (3.0, 3.0)], terrain, [[(9.0, 9.0)]], 0.14)


def test_추천_저장본_종류당_최근_두_벌(tmp_path: Path) -> None:
    from B05_Profile.B05_Profile_Engine_RouteInitial_Store import (
        KEEP_PER_KIND,
        read_stored,
        store_dir,
        write_stored,
    )

    assert KEEP_PER_KIND == 2
    request = {"mode": "earthwork", "terrain": None}
    write_stored(tmp_path, "a", "p", request, {"n": 1}, True)
    time.sleep(0.02)
    write_stored(tmp_path, "b", "p", request, {"n": 2}, False)
    write_stored(tmp_path, "g", "p", {"mode": "gentle", "terrain": None}, {}, True)
    time.sleep(0.02)
    assert read_stored(tmp_path, "a") is not None  # 꺼내면 최근으로
    time.sleep(0.02)
    write_stored(tmp_path, "c", "p", request, {"n": 3}, False)
    time.sleep(0.02)
    write_stored(tmp_path, "d", "p", request, {"n": 4}, False)
    names = sorted(path.stem for path in store_dir(tmp_path).glob("*.json"))
    # 기본값(a)은 따로 셈 — 사용자 시도 b · c · d 중 옛 b 만 지움(351-1)
    assert names == ["a", "c", "d", "g"]


# ── 횡단 상세 — GET 은 안 지움 · ETag(R26) ───────────────────────────────────────


def test_횡단_상세_조회는_파일을_지우지_않음(tmp_path: Path) -> None:
    from B05_Profile.B05_Profile_Engine_Sections import prune_stale_cross_files, valid_cross_names

    for name in ("cross_00000m.json", "cross_00020m.json", "cross_00040m.json"):
        (tmp_path / name).write_text("{}", encoding="utf-8")
    stations = [{"chainage_m": 0.0}, {"chainage_m": 20.0}]
    names = valid_cross_names(tmp_path, stations)
    assert len(names) == 2 and (tmp_path / "cross_00040m.json").is_file()
    assert prune_stale_cross_files(tmp_path, stations) == names
    assert not (tmp_path / "cross_00040m.json").exists()


def test_횡단_상세_ETag_같으면_304() -> None:
    from starlette.requests import Request

    from B06_Section.B06_Section_Router import _etag_json_response

    def request(etag: str | None) -> Request:
        headers = [] if etag is None else [(b"if-none-match", etag.encode())]
        return Request({"type": "http", "headers": headers})

    body = b'{"cross_sections": []}'
    first = _etag_json_response(request(None), body)
    assert first.status_code == 200 and first.body == body
    etag = first.headers["etag"]
    assert _etag_json_response(request(etag), body).status_code == 304
    assert _etag_json_response(request(etag), body + b" ").status_code == 200


def test_횡단_상세_GET_소스에_지우기_없음() -> None:
    source = Path("B06_Section/B06_Section_Router.py").read_text(encoding="utf-8")
    start = source.index("def _read_section_detail(")
    end = source.index("\ndef ", start + 10)
    assert "prune_stale_cross_files" not in source[start:end]
    assert "unlink" not in source[start:end]