Aislo/resources/tester/test_312_empty_pipes.py

"""312 — 관 목록 「빈 목록 = 관 0개」 · 「None = 자동 배치」.

마지막 관을 지우면 화면이 자동 배치로 다시 차고 새로고침 · [저장] 뒤 저장분 하나가 되살아나던 것.
서버는 검증_sub5 storage **사본** + 읽기 전용 DB 로 부름(원본 · DB 안 바뀜).
"""

from __future__ import annotations

import json
import re
import shutil
from contextlib import asynccontextmanager
from pathlib import Path

import pytest

ROOT = Path(__file__).resolve().parents[2]
PROJECT_ID = "bec7fe41-83e7-4df7-86a3-fdddaa1f8fec"  # OWNERS.md — laptop-sub5 `검증_sub5`
SOURCE = ROOT / "storage" / "1" / "3" / PROJECT_ID
PIPES_FILE = "B04_PreProcess/drainage/edits/pipe_points.json"
BASE = f"/api/projects/{PROJECT_ID}"


def _read(rel: str) -> str:
    return (ROOT / rel).read_text(encoding="utf-8")


def _copy_project(dest: Path) -> None:
    for rel in ("B04_PreProcess/drainage", "B05_Profile"):
        shutil.copytree(SOURCE / rel, dest / rel)
    for rel in ("B04_PreProcess/models", "B04_PreProcess/processed"):
        (dest / rel).mkdir(parents=True, exist_ok=True)
        for item in (SOURCE / rel).iterdir():
            if item.is_file() and item.stat().st_size <= 20 * 1024 * 1024:
                shutil.copy2(item, dest / rel / item.name)
    for item in SOURCE.iterdir():
        if item.is_file():
            shutil.copy2(item, dest / item.name)


@pytest.fixture()
def api(tmp_path, monkeypatch):
    if not (SOURCE / PIPES_FILE).is_file():
        pytest.skip("배정 storage 프로젝트가 없음")
    aiomysql = pytest.importorskip("aiomysql")
    from fastapi import FastAPI
    from fastapi.testclient import TestClient

    import B04_PreProcess.B04_PreProcess_Router_Basins as basins
    import common_util.common_util_storage as storage
    import config.config_db as config_db
    from config import config_system as cfg

    storage_root = tmp_path / "storage"
    root = storage_root / "1" / "3" / PROJECT_ID
    _copy_project(root)
    monkeypatch.setattr(storage, "STORAGE_BASE_DIR", str(storage_root))

    @asynccontextmanager
    async def lifespan(_app):
        try:
            pool = await aiomysql.create_pool(
                host=cfg.DB_HOST,
                port=cfg.DB_PORT,
                db=cfg.DB_NAME,
                user=cfg.DB_USER,
                password=cfg.DB_PASSWORD,
                minsize=1,
                maxsize=4,
                autocommit=True,
                charset="utf8mb4",
                init_command="SET time_zone='+00:00', SESSION tx_read_only=1",
            )
        except Exception as exc:  # DB 가 안 닿는 곳
            pytest.skip(f"DB 에 안 닿음: {exc}")
        config_db.db_pool = pool
        try:
            yield
        finally:
            config_db.db_pool = None
            pool.close()
            await pool.wait_closed()

    app = FastAPI(lifespan=lifespan)
    app.include_router(basins.router)
    with TestClient(app) as client:
        client.root = root  # type: ignore[attr-defined]
        yield client


def _file_points(api) -> list[dict]:
    return json.loads((api.root / PIPES_FILE).read_text(encoding="utf-8"))["points"]


def _stored(api) -> list[dict]:
    got = api.get(f"{BASE}/drainage/pipe-points").json()["pipe_points"]
    return [{k: v for k, v in p.items() if k != "lonlat"} for p in got]


def test_PUT_빈_목록은_관_0개로_저장(api) -> None:
    keep = _stored(api)
    assert len(keep) >= 2
    one = api.put(f"{BASE}/drainage/pipe-points", json={"points": keep[:1]}).json()
    assert one["saved_count"] == 1
    res = api.put(f"{BASE}/drainage/pipe-points", json={"points": []}).json()
    assert res["saved_count"] == 0 and res["pipe_points"] == [] and res["basins"] == []
    assert _file_points(api) == []
    got = api.get(f"{BASE}/drainage/pipe-points").json()
    assert got["pipe_points"] == [] and got["saved"] is True  # 새로고침 = 0 · 되살아나지 않음


def test_POST_빈_목록은_관_0_유역_0(api) -> None:
    res = api.post(f"{BASE}/drainage/detail-basins", json={"points": []}).json()
    assert res["pipe_points"] == [] and res["basins"] == []
    assert len(_file_points(api)) > 0  # 계산만 · 저장 안 함


def test_칸_없음_파일_없음_DELETE_노선_바뀜은_자동_배치(api) -> None:
    auto = api.post(f"{BASE}/drainage/detail-basins", json={}).json()["pipe_points"]
    assert len(auto) > 0  # 요청에 칸 없음 = 자동 배치
    api.put(f"{BASE}/drainage/pipe-points", json={"points": []})
    reset = api.delete(f"{BASE}/drainage/pipe-points").json()
    assert len(reset["pipe_points"]) == len(auto) and reset["saved"] is False  # [초기화]
    assert not (api.root / PIPES_FILE).exists()
    got = api.get(f"{BASE}/drainage/pipe-points").json()  # 파일 없음
    assert len(got["pipe_points"]) == len(auto) and got["saved"] is False
    # 노선 바뀜(지문 다름 · 좌표 없는 옛 저장분) = None → 자동 배치
    (api.root / PIPES_FILE).parent.mkdir(parents=True, exist_ok=True)
    (api.root / PIPES_FILE).write_text(
        json.dumps({"route_signature": "old", "points": [{"chainage_m": 50.0}]}),
        encoding="utf-8",
    )
    got = api.get(f"{BASE}/drainage/pipe-points").json()
    assert len(got["pipe_points"]) == len(auto) and got["saved"] is False


def test_빈_목록과_None_을_가르는_자리() -> None:
    src = _read("common_util/common_util_drainage_detail.py")
    assert "if confirmed_chainages is not None:" in src
    router = _read("B04_PreProcess/B04_PreProcess_Router_Basins.py")
    assert "points = requested if requested is not None else stored" in router
    assert "requested or stored" not in router


def _body(rel: str, start: str, end: str) -> str:
    text = _read(rel)
    return text[text.index(start) : text.index(end, text.index(start))]


def test_화면_초안_빈_목록은_저장분으로_안_넘어감() -> None:
    load = _body(
        "B05_Profile/B05_Profile_UI_Drainage_Panel.ts", "const loadSaved = ", "const resetPipes"
    )
    assert "pending !== null" in load and "pending?.length" not in load
    # B06 목록도 같은 규칙(`??`)
    b06 = _read("B06_Section/B06_Section_UI_Page_Structures_Panel.ts")
    assert "(pendingPipes ?? pipeResponse.pipe_points)" in b06


def test_횡단_벽_초안_빈_목록은_다_지움_그리고_옛_결함_막이() -> None:
    store = _body(
        "B06_Section/B06_Section_Section_Store.ts", "async function withDraftWalls(", "\n}\n"
    )
    assert "if (!drafts) return withStructureAreas(detail);" in store
    assert "!drafts.length" not in store
    # 옛 「놓아도 안 보임」(09-09) — 옛 배열 참조가 `[]` 를 알리던 것. 정본 주입은 배열 안을
    # 채우고(갈아끼우지 않음) onChange 를 울리지 않아야 함.
    inject = _body(
        "B05_Profile/B05_Profile_UI_Structures_Panel.ts", "    setStructures(next) {", "    },\n"
    )
    assert "structures.splice(0, structures.length" in inject
    assert not re.search(r"\bemit\(|onChange\(", inject)
    # 알림은 지금 목록 그대로를 실어 보냄(빈 배열을 따로 만들지 않음)
    commit = _read("B05_Profile/B05_Profile_UI_Structures_Panel_Commit.ts")
    assert "ctx.callbacks.onChange([...ctx.structures]);" in commit