Aislo/resources/tester/test_763_3_surface_change.py

"""763-3 (요구 605) B04 분석 · 조합 짓기 = 보기만 · 다른 서피스 [확정] = 업로드와 같은 전처리 체인 ·
같은 서피스 다시 [확정] = 다시 셈 없음 · 체인 실패에도 추천 미리 셈 · B05~B09 는 서피스를 안 바꿈.

    ./venv/Scripts/python.exe -m pytest resources/tester/test_763_3_surface_change.py -q
"""

import asyncio
import re
from contextlib import asynccontextmanager
from pathlib import Path
from uuid import uuid4

import pytest

from B03_FileInput import B03_FileInput_Service_SurfaceChange as change
from B04_PreProcess import B04_PreProcess_Router as b04
from B04_PreProcess.B04_PreProcess_Schema import SurfaceAnalyzeRequest, SurfaceConfirmRequest

ROOT = Path(__file__).resolve().parents[2]
CONFIRMED = {
    "source_filter": "classification",
    "method": "implicit",
    "smooth": False,
    "contour_interval_m": 1.0,
}


class _Cursor:
    def __init__(self, log):
        self.log = log

    async def __aenter__(self):
        return self

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

    async def execute(self, sql, *_a):
        self.log.append(" ".join(str(sql).split()))

    async def fetchone(self):
        return None


class _Connection:
    def __init__(self, log):
        self.log = log

    def cursor(self, *_a):
        return _Cursor(self.log)

    async def begin(self):
        self.log.append("BEGIN")

    async def commit(self):
        self.log.append("COMMIT")

    async def rollback(self):
        self.log.append("ROLLBACK")


class _Pool:
    def __init__(self, log):
        self.log = log

    @asynccontextmanager
    async def acquire(self):
        yield _Connection(self.log)


def _async(value):
    async def inner(*_a, **_k):
        return value

    return inner


@pytest.fixture
def b04_world(tmp_path, monkeypatch):
    log: list[str] = []
    calls: dict[str, list] = {"progress": [], "chain": [], "confirm": [], "contour": []}
    las = tmp_path / "raw.las"
    las.write_bytes(b"")
    monkeypatch.setattr(b04, "get_db_pool", lambda: _Pool(log))
    monkeypatch.setattr(b04, "get_project_storage_relative_path", _async("x"))
    monkeypatch.setattr(b04, "resolve_stored_project_path", lambda _p: str(tmp_path))
    monkeypatch.setattr(b04, "get_input_file", _async({"raw_file_path": "raw.las"}))
    monkeypatch.setattr(b04, "list_project_point_cloud_paths", _async([las]))
    monkeypatch.setattr(
        b04,
        "run_surface_analysis",
        lambda *_a, **_k: {"ground_summary": {}, "manifest": {"status": "ok"}},
    )
    monkeypatch.setattr(b04, "save_surface_analysis_to_db", _async([1, 2]))
    monkeypatch.setattr(b04, "confirmed_surface_params", _async(dict(CONFIRMED)))
    monkeypatch.setattr(
        b04, "write_surface_progress", lambda _r, p, stage, _m: calls["progress"].append(stage)
    )
    monkeypatch.setattr(
        b04,
        "list_surface_models",
        _async(
            [
                {
                    "id": 7,
                    "model_type": "implicit",
                    "generation_params": {"source_filter": "classification"},
                },
                {"id": 8, "model_type": "implicit", "generation_params": {"source_filter": "ptd"}},
            ]
        ),
    )

    async def confirm(_c, _pid, model_id, selection):
        calls["confirm"].append((model_id, selection["source_filter"]))

    async def contour(_c, _pid, value):
        calls["contour"].append(value)

    async def chain(pid, model_id):
        calls["chain"].append(model_id)

    monkeypatch.setattr(b04, "confirm_surface_selection", confirm)
    monkeypatch.setattr(b04, "update_contour_interval_param", contour)
    monkeypatch.setattr(change, "run_surface_change_chain", chain)
    calls["log"] = log
    return calls


def test_분석은_보기만_단계_확정_그대로(b04_world) -> None:
    request = SurfaceAnalyzeRequest(input_file_id=3, source_filters=["ptd"], methods=["implicit"])
    answer = asyncio.run(b04.analyze_surface(uuid4(), request))
    assert answer.surface_model_ids == [1, 2]
    sql = " | ".join(b04_world["log"])
    assert "project_workflow_stages" not in sql and "surface_models" not in sql
    assert b04_world["progress"][-1] == "completed"  # 확정 그대로 — 「확정하세요」 아님
    source = (ROOT / "B04_PreProcess/B04_PreProcess_Router.py").read_text(encoding="utf-8")
    body = source.split("async def analyze_surface(", 1)[1].split("\n@router.", 1)[0]
    for name in ("start_stage", "fail_stage", "update_stage_progress", "schedule_precompute"):
        assert name not in body, name


def test_조합_다시_적어도_확정_행_그대로() -> None:
    source = (ROOT / "B04_PreProcess/B04_PreProcess_Repository.py").read_text(encoding="utf-8")
    assert "status = IF(status = 'CONFIRMED', 'CONFIRMED', 'COMPLETE')" in source
    assert "clear_confirmed_surface_models" not in source


def _confirm(model_id: int, contour: float | None = None):
    request = SurfaceConfirmRequest(model_id=model_id, smooth=False, contour_interval_m=contour)

    async def run():
        answer = await b04.confirm_surface(uuid4(), request, {})
        await asyncio.sleep(0)  # 백그라운드 체인 한 번 돌림
        return answer

    return asyncio.run(run())


def test_같은_서피스_다시_확정은_다시_셈_없음(b04_world) -> None:
    _confirm(7, 2.0)
    assert b04_world["confirm"] == [] and b04_world["chain"] == []
    assert b04_world["contour"] == [2.0]


def test_다른_서피스_확정은_전처리_체인(b04_world) -> None:
    _confirm(8)
    assert b04_world["confirm"] == [(8, "ptd")]
    assert b04_world["chain"] == [8]


def test_서피스_바꿈_체인_차례(tmp_path, monkeypatch) -> None:
    import B03_FileInput.B03_FileInput_Repository as repo
    import B03_FileInput.B03_FileInput_Service_Chain as chain
    import B05_Profile.B05_Profile_Router_RouteInitial as router
    import common_util.common_util_initial_snapshot as snapshot
    import common_util.common_util_project_reset as reset
    import common_util.common_util_storage as storage
    import common_util.common_util_workflow_state as state
    import config.config_db as db

    log: list = []

    async def purge(_c, _pid, _root, keep_preprocess=False):
        log.append(("purge", keep_preprocess))

    async def stages(_cursor, _pid, first_stage=1):
        log.append(("stages", first_stage))

    async def auto(_pid, surface_model_id=None):
        log.append(("preprocess", surface_model_id))

    monkeypatch.setattr(db, "get_db_pool", lambda: _Pool([]))
    monkeypatch.setattr(repo, "get_project_storage_relative_path", _async("x"))
    monkeypatch.setattr(storage, "resolve_stored_project_path", lambda _p: str(tmp_path))
    monkeypatch.setattr(snapshot, "discard_initial_snapshot", lambda _r: log.append("snapshot"))
    monkeypatch.setattr(reset, "purge_project_outputs", purge)
    monkeypatch.setattr(state, "reset_stages_after_input_change", stages)
    monkeypatch.setattr(chain, "run_preprocess_chain", auto)
    monkeypatch.setattr(router, "schedule_precompute", lambda pid: log.append("precompute"))
    asyncio.run(change.run_surface_change_chain(uuid4(), 8))
    assert log == ["snapshot", ("purge", True), ("stages", 2), ("preprocess", 8)]

    async def broken(*_a, **_k):
        raise RuntimeError("db")

    log.clear()
    monkeypatch.setattr(reset, "purge_project_outputs", broken)
    asyncio.run(change.run_surface_change_chain(uuid4(), 8))
    assert log == ["snapshot", "precompute"]  # 초기화 실패 = 체인 없음 · 추천만 새 서피스로


@pytest.mark.parametrize("name", ["run_auto_design_chain", "run_preprocess_chain"])
def test_체인_실패_길에도_미리_셈(monkeypatch, name) -> None:
    import B03_FileInput.B03_FileInput_Service_Chain as chain
    import B05_Profile.B05_Profile_Router_RouteInitial as router
    import config.config_db as db

    class _Broken:
        def acquire(self):
            raise RuntimeError("db down")

    scheduled: list = []
    monkeypatch.setattr(db, "get_db_pool", lambda: _Broken())
    monkeypatch.setattr(router, "schedule_precompute", scheduled.append)
    project = uuid4()
    asyncio.run(getattr(chain, name)(project))
    assert scheduled == [project]


def test_LAS_나중_올리기_끝에_미리_셈() -> None:
    source = (ROOT / "B05_Profile/B05_Profile_Router_Lidar.py").read_text(encoding="utf-8")
    work = source.split("async def work(", 1)[1].split("return lidar_upload.start_job", 1)[0]
    assert "schedule_precompute(project_id)" in work


#: B05~B09 에서 확정 서피스를 바꿀 수 있는 부름 — 남은 곳은 까닭 있는 둘만(요구 605-2)
_SURFACE_WRITES = re.compile(
    r"\b(confirm_surface_selection|confirm_surface_model|merge_surface_confirmation_params|"
    r"switch_design_terrain|stash_las_selection|run_surface_change_chain)\("
)
_ALLOWED = {
    # 지형 칩(626-1) — 사용자 답 기다림(763 ①) · 추천 [적용]은 지형을 안 넘김(아래 시험)
    "B05_Profile/B05_Profile_Router_Replan.py",
    # 라이다 확정값만 적어 둠 · 설계 지형 그대로(637-2)
    "B05_Profile/B05_Profile_Router_Lidar.py",
    # 지형 칩 = 벌 맞바꿈 · LAS 나중 올리기 뒤 전처리 다시(765-4 · 5 · 요구 605-1 · 606-1)
    "B05_Profile/B05_Profile_Router_TerrainSwitch.py",
}


def test_B05_B09_확정은_서피스를_안_바꿈() -> None:
    found = set()
    for folder in (
        "B05_Profile",
        "B06_Section",
        "B07_DesignDetail",
        "B08_Quantity",
        "B09_Estimation",
    ):
        for path in (ROOT / folder).glob("*.py"):
            if _SURFACE_WRITES.search(path.read_text(encoding="utf-8")):
                found.add(path.relative_to(ROOT).as_posix())
    assert found <= _ALLOWED, found - _ALLOWED
    replan = (ROOT / "B05_Profile/B05_Profile_Router_Replan.py").read_text(encoding="utf-8")
    save = replan.split("async def _save_route(", 1)[1].split("\nasync def ", 1)[0]
    assert "request.terrain" not in save  # 추천 [적용] = 선만 · 설계 지형 그대로(605-2)