# -*- coding: utf-8 -*-
"""765-2 지형마다 설계 한 벌 — 떠 두기 · 꺼내 끼우기 · 맞바꿈(`common_util_design_sets`).

DB = 메모리 sqlite 를 aiomysql 꼴로 감싼 가짜(노선 · 자식 넷 · 단계 · 프로젝트 · 지표면 모델) ·
storage = 시험 안 임시 폴더. 지킴:
  ① 떠 둔 뒤 꺼내면 값 같음 — 노선 · 자식 · 단계 · 상태 · 벌 몫 파일 · 초기값 · 코리도
  ② 꺼낼 때 새 노선 id · 자식 다시 잇기 · 코리도 이름 새 id
  ③ 공유 캐시(비용면 · 예정노선 · 배수 하강장 · 도엽 좌표 · 강우)는 안 옮기고 안 지움
  ④ 지문 어긋난(묵은) 벌은 버리고 그 자리에서 지음 · 실패면 통째 되돌림
  ⑤ 벌 예정노선(765-8) = 받은 노선을 그 벌 지형으로 자름 · 고친 예정노선과 함께 벌 몫 · 초기
     폴리라인 바탕 · 공유 예상노선은 설계 지형 하나(추천 열쇠)
"""

from __future__ import annotations

import asyncio
import json
import re
import sqlite3
import sys
from pathlib import Path
from types import SimpleNamespace
from uuid import uuid4

import pytest

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

import aiomysql  # noqa: E402

from common_util import common_util_design_sets as sets  # noqa: E402

PID = str(uuid4())

_SCHEMA = """
CREATE TABLE projects (id TEXT PRIMARY KEY, status TEXT, name TEXT);
CREATE TABLE surface_models (id INTEGER PRIMARY KEY, project_id TEXT, status TEXT);
CREATE TABLE project_workflow_stages (
  id INTEGER PRIMARY KEY AUTOINCREMENT, project_id TEXT, stage_no INT, stage_key TEXT,
  state TEXT, progress_percent INT, params TEXT, message TEXT, started_at TEXT, completed_at TEXT);
CREATE TABLE routes (
  id INTEGER PRIMARY KEY AUTOINCREMENT, project_id TEXT, surface_model_id INT, status TEXT,
  total_length_m REAL, route_data_path TEXT, created_at TEXT);
CREATE TABLE route_points (
  id INTEGER PRIMARY KEY AUTOINCREMENT,
  route_id INT REFERENCES routes(id) ON DELETE CASCADE, chainage_m REAL, sequence_num INT);
CREATE TABLE route_statistics (
  id INTEGER PRIMARY KEY AUTOINCREMENT,
  route_id INT REFERENCES routes(id) ON DELETE CASCADE, cut_volume_m3 REAL);
CREATE TABLE longitudinal_sections (
  id INTEGER PRIMARY KEY AUTOINCREMENT, project_id TEXT,
  route_id INT REFERENCES routes(id) ON DELETE CASCADE, data TEXT);
CREATE TABLE cross_sections (
  id INTEGER PRIMARY KEY AUTOINCREMENT, project_id TEXT,
  route_id INT REFERENCES routes(id) ON DELETE CASCADE, chainage_m REAL, data TEXT);
"""


class _Cursor:
    def __init__(self, db: sqlite3.Connection, dict_rows: bool) -> None:
        self.db, self.dict_rows, self.rows, self.lastrowid, self.rowcount = db, dict_rows, [], 0, 0

    async def __aenter__(self):
        return self

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

    @staticmethod
    def _sql(sql: str) -> str:
        return re.sub(r"%s", "?", sql)

    async def execute(self, sql: str, args=()) -> None:
        cur = self.db.execute(self._sql(sql), tuple(args or ()))
        self.lastrowid, self.rowcount = cur.lastrowid, cur.rowcount
        names = [d[0] for d in cur.description or []]
        rows = cur.fetchall()
        self.rows = [dict(zip(names, r)) for r in rows] if self.dict_rows else rows

    async def executemany(self, sql: str, seq) -> None:
        self.db.executemany(self._sql(sql), [tuple(a) for a in seq])

    async def fetchall(self):
        return self.rows

    async def fetchone(self):
        return self.rows[0] if self.rows else None


class _Conn:
    def __init__(self, db: sqlite3.Connection) -> None:
        self.db = db

    def cursor(self, kind=None):
        return _Cursor(self.db, kind is aiomysql.DictCursor)

    async def begin(self):
        pass

    async def commit(self):
        self.db.commit()

    async def rollback(self):
        self.db.rollback()


class _Pool:
    def __init__(self, conn: _Conn) -> None:
        self.conn = conn

    def acquire(self):
        conn = self.conn

        class _Ctx:
            async def __aenter__(self):
                return conn

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

        return _Ctx()


def _seed_db() -> sqlite3.Connection:
    db = sqlite3.connect(":memory:")
    db.execute("PRAGMA foreign_keys = ON")
    db.executescript(_SCHEMA)
    db.execute("INSERT INTO projects VALUES (?, 'WF3_ANALYZING', 'p')", (PID,))
    db.executemany(
        "INSERT INTO surface_models VALUES (?, ?, ?)",
        [(5, PID, "CONFIRMED"), (6, PID, "COMPLETE")],
    )
    for no in range(7):
        db.execute(
            "INSERT INTO project_workflow_stages (project_id, stage_no, stage_key, state,"
            " progress_percent, params, message) VALUES (?, ?, ?, ?, ?, ?, ?)",
            (PID, no, f"S{no}", "IN_PROGRESS", 50, json.dumps({"no": no, "k": "las"}), f"m{no}"),
        )
    for rid in (10, 11):
        db.execute(
            "INSERT INTO routes (id, project_id, surface_model_id, status, total_length_m,"
            " route_data_path, created_at) VALUES (?, ?, 5, 'CONFIRMED', ?, 'r.geojson', 't')",
            (rid, PID, rid * 100.0),
        )
        for seq in range(3):
            db.execute(
                "INSERT INTO route_points (route_id, chainage_m, sequence_num) VALUES (?, ?, ?)",
                (rid, seq * 20.0, seq),
            )
            db.execute(
                "INSERT INTO cross_sections (project_id, route_id, chainage_m, data)"
                " VALUES (?, ?, ?, ?)",
                (PID, rid, seq * 20.0, json.dumps({"cut": seq})),
            )
        db.execute("INSERT INTO route_statistics (route_id, cut_volume_m3) VALUES (?, 7)", (rid,))
        db.execute(
            "INSERT INTO longitudinal_sections (project_id, route_id, data) VALUES (?, ?, '{}')",
            (PID, rid),
        )
    db.commit()
    return db


_SET_FILES = {
    "B05_Profile/route/planned_route.csv": "x,y\n1,2\n",
    "B05_Profile/route/terrain_route.csv": "t",
    "B05_Profile/route/expected_route_edit.csv": "e2",
    "B05_Profile/route/structures.json": '{"items": [1]}',
    "B05_Profile/route/route_main.geojson": "{}",
    "B05_Profile/corridor/corridor_0011.json": '{"route": 11}',
    "B06_Section/longitudinal/longitudinal.json": '{"z": [1, 2]}',
    "B06_Section/sampling.json": '{"surface_model_id": 5}',
    "B04_PreProcess/drainage/01_primary_region.geojson": "{}",
    "B04_PreProcess/drainage/edits/pipe_points.json": "[1]",
    "B04_PreProcess/drainage/manifest.json": "{}",
    "initial_snapshot/db.json": '{"routes": []}',
}
_SHARED_FILES = {
    "B05_Profile/route/cost_surface_x_implicit.npz": "c",
    "B05_Profile/route/expected_route.csv": "e",
    "B05_Profile/route/initial_choice.json": "{}",
    "B04_PreProcess/drainage/descent_ab_cd.npz": "d",
    "B04_PreProcess/drainage/00_sheet_metric.pkl": "s",
    "B04_PreProcess/drainage/rainfall_table.json": "r",
    "B04_PreProcess/models/dtm_x.npz": "m",
}


def _seed_files(root: Path) -> None:
    for relative, text in {**_SET_FILES, **_SHARED_FILES}.items():
        path = root / relative
        path.parent.mkdir(parents=True, exist_ok=True)
        path.write_text(text, encoding="utf-8")


def _tree(root: Path) -> dict[str, str]:
    return {
        p.relative_to(root).as_posix(): p.read_text(encoding="utf-8")
        for p in sorted(root.rglob("*"))
        if p.is_file() and SETS_DIR not in p.relative_to(root).parts
    }


SETS_DIR = sets.SETS_DIRNAME
FP = {
    "terrain": "las",
    "source_filter": "classification",
    "method": "implicit",
    "smooth": False,
    "model_id": 5,
    "model_digest": "aa",
}


def _rows(db: sqlite3.Connection, sql: str, args=()) -> list[tuple]:
    return db.execute(sql, args).fetchall()


def _no_ids(db: sqlite3.Connection) -> dict[str, list]:
    """id 를 뺀 값(자식은 노선 차례로) — 왕복 비교용."""
    order = {rid: i for i, (rid,) in enumerate(_rows(db, "SELECT id FROM routes ORDER BY id"))}
    out = {
        "routes": _rows(
            db,
            "SELECT project_id, surface_model_id, status, total_length_m, route_data_path,"
            " created_at FROM routes ORDER BY id",
        ),
        "stages": _rows(
            db,
            "SELECT stage_no, state, progress_percent, params, message FROM"
            " project_workflow_stages ORDER BY stage_no",
        ),
        "status": _rows(db, "SELECT status FROM projects"),
    }
    for table, cols in (
        ("route_points", "chainage_m, sequence_num"),
        ("route_statistics", "cut_volume_m3"),
        ("longitudinal_sections", "project_id, data"),
        ("cross_sections", "project_id, chainage_m, data"),
    ):
        out[table] = sorted(
            (order[r[0]], *r[1:])
            for r in _rows(db, f"SELECT route_id, {cols} FROM {table} ORDER BY id")
        )
    return out


def test_벌_몫과_공유를_가름(tmp_path: Path) -> None:
    _seed_files(tmp_path)
    entries = sets.set_entries(tmp_path)
    assert "B06_Section" in entries and "initial_snapshot" in entries
    assert "B05_Profile/route/structures.json" in entries
    assert "B04_PreProcess/drainage/edits" in entries
    joined = " ".join(entries)
    for shared in ("cost_surface_", "expected_route.csv", "initial_choice", "descent_", "00_sheet"):
        assert shared not in joined
    # 벌 예정노선 · 고친 예정노선 = 벌 몫(765-8)
    assert "B05_Profile/route/terrain_route.csv" in entries
    assert "B05_Profile/route/expected_route_edit.csv" in entries
    assert "rainfall_" not in joined and "models" not in joined


def test_떠_두고_꺼내면_값_같음_새_id(tmp_path: Path) -> None:
    db = _seed_db()
    conn = _Conn(db)
    _seed_files(tmp_path)
    before_db, before_files = _no_ids(db), _tree(tmp_path)

    meta = asyncio.run(sets.park_active_set(conn, tmp_path, PID, FP))
    assert meta["routes"] == [10, 11] and meta["built"]
    assert _rows(db, "SELECT COUNT(*) FROM routes") == [(0,)]
    assert _rows(db, "SELECT COUNT(*) FROM cross_sections") == [(0,)]
    states = _rows(db, "SELECT stage_no, state, params FROM project_workflow_stages ORDER BY 1")
    assert states[:2] == [(0, "IN_PROGRESS", before_db["stages"][0][3])] + [
        (1, "IN_PROGRESS", before_db["stages"][1][3])
    ]
    assert all(s[1] == "NOT_STARTED" and s[2] is None for s in states[2:])
    left = _tree(tmp_path)
    assert set(left) == set(_SHARED_FILES)  # 공유만 남음 · 그대로
    assert all(left[k] == v for k, v in _SHARED_FILES.items())
    bundle = tmp_path / SETS_DIR / "las"
    assert (bundle / "db.json").is_file() and (bundle / "initial_snapshot" / "db.json").is_file()
    assert (bundle / "files" / "B06_Section" / "sampling.json").is_file()
    assert sets.bundle_ready(tmp_path, "las", FP)

    route_map = asyncio.run(sets.load_set(conn, tmp_path, PID, "las"))
    assert set(route_map) == {10, 11}
    assert not set(route_map.values()) & {10, 11}  # 새 id
    assert _no_ids(db) == before_db  # 노선 · 자식 · 단계 · 상태 같음
    after = _tree(tmp_path)
    moved = before_files.pop("B05_Profile/corridor/corridor_0011.json")
    new_corridor = f"B05_Profile/corridor/corridor_{route_map[11]:04d}.json"
    assert after.pop(new_corridor) == moved  # 코리도 = 새 id 이름
    assert after == before_files
    assert not bundle.exists()  # 꺼낸 묶음은 지움


def test_지문_어긋나면_묵은_벌(tmp_path: Path) -> None:
    db = _seed_db()
    _seed_files(tmp_path)
    asyncio.run(sets.park_active_set(_Conn(db), tmp_path, PID, FP))
    assert sets.bundle_ready(tmp_path, "las", FP)
    assert sets.bundle_ready(tmp_path, "las", {**FP, "model_id": 99})  # 복제 · 갈음 id 는 안 봄
    assert not sets.bundle_ready(tmp_path, "las", {**FP, "model_digest": "bb"})
    assert not sets.bundle_ready(tmp_path, "las", {**FP, "source_filter": "ptd"})
    assert not sets.bundle_ready(tmp_path, "las", None)
    assert not sets.bundle_ready(tmp_path, "sheet", {**FP, "terrain": "sheet"})
    sets.note_unbuilt(tmp_path, "sheet", "수치지형도 모델 없음")
    assert sets.read_meta(tmp_path, "sheet") == {
        "terrain": "sheet",
        "built": False,
        "reason": "수치지형도 모델 없음",
    }
    assert not sets.bundle_ready(tmp_path, "sheet", {**FP, "terrain": "sheet"})


def _swap_world(monkeypatch, tmp_path: Path, *, sheet_fp: dict, chain):
    """las 활성 · sheet 묶음(지문 = sheet_fp) — 칩으로 sheet."""
    import B03_FileInput.B03_FileInput_Service_Chain as chain_mod
    import config.config_db as config_db

    db = _seed_db()
    conn = _Conn(db)
    _seed_files(tmp_path)
    sheet = {**FP, "terrain": "sheet", "source_filter": "sheet_laplace", "model_id": 6}
    # sheet 벌을 미리 떠 둠: 노선 하나(길이 777)
    asyncio.run(sets.park_active_set(conn, tmp_path, PID, FP))
    db.execute(
        "INSERT INTO routes (project_id, surface_model_id, status, total_length_m) VALUES"
        " (?, 6, 'CONFIRMED', 777)",
        (PID,),
    )
    db.commit()
    (tmp_path / "B06_Section").mkdir(parents=True, exist_ok=True)
    (tmp_path / "B06_Section" / "sheet.json").write_text("sheet", encoding="utf-8")
    asyncio.run(sets.park_active_set(conn, tmp_path, PID, sheet_fp))
    asyncio.run(sets.load_set(conn, tmp_path, PID, "las"))
    state = {"design": "las"}

    async def params(_c, _pid):
        return {
            "source_filter": "sheet_laplace" if state["design"] == "sheet" else "classification"
        }

    async def fingerprint(_c, _pid, _root, terrain):
        return sheet if terrain == "sheet" else FP

    async def switch(_c, _pid, terrain):
        state["design"] = terrain
        db.execute(
            "UPDATE surface_models SET status = CASE id WHEN 6 THEN 'CONFIRMED' ELSE 'COMPLETE' END"
        )
        return 6

    monkeypatch.setattr(config_db, "get_db_pool", lambda: _Pool(conn))
    monkeypatch.setattr(sets, "get_surface_confirmation_params", params)
    monkeypatch.setattr(sets, "set_fingerprint", fingerprint)
    monkeypatch.setattr(sets, "switch_terrain", switch)
    monkeypatch.setattr(chain_mod, "run_auto_design_chain", chain)
    return db, state


def test_맞바꿈_묶음_있으면_꺼냄(monkeypatch, tmp_path: Path) -> None:
    called = []

    async def chain(*a, **k):
        called.append(a)

    sheet_fp = {**FP, "terrain": "sheet", "source_filter": "sheet_laplace", "model_id": 6}
    db, _ = _swap_world(monkeypatch, tmp_path, sheet_fp=sheet_fp, chain=chain)
    result = asyncio.run(sets.swap_design_set(PID, tmp_path, "sheet"))  # type: ignore[arg-type]
    assert result["changed"] and not result["built"] and not called
    assert _rows(db, "SELECT total_length_m FROM routes") == [(777.0,)]
    assert (tmp_path / "B06_Section" / "sheet.json").read_text(encoding="utf-8") == "sheet"
    assert sets.bundle_ready(tmp_path, "las", FP)  # 떠난 벌 살림
    assert not (tmp_path / SETS_DIR / "sheet").exists()


def test_맞바꿈_묵은_벌은_버리고_지음(monkeypatch, tmp_path: Path) -> None:
    called = []
    holder: dict = {}

    async def chain(pid, model_id, inner=False):
        called.append((model_id, inner))
        holder["db"].execute(
            "INSERT INTO routes (project_id, surface_model_id, status, total_length_m) VALUES"
            " (?, 6, 'CONFIRMED', 555)",
            (PID,),
        )
        return {"route_id": 1}

    stale = {**FP, "terrain": "sheet", "source_filter": "sheet_laplace", "model_digest": "old"}
    db, _ = _swap_world(monkeypatch, tmp_path, sheet_fp=stale, chain=chain)
    holder["db"] = db
    result = asyncio.run(sets.swap_design_set(PID, tmp_path, "sheet"))  # type: ignore[arg-type]
    assert result["built"] and called == [(6, True)]
    assert _rows(db, "SELECT total_length_m FROM routes") == [(555.0,)]  # 묵은 777 버림
    assert not (tmp_path / "B06_Section" / "sheet.json").exists()
    assert sets.bundle_ready(tmp_path, "las", FP)


def test_맞바꿈_실패면_통째_되돌림(monkeypatch, tmp_path: Path) -> None:
    async def chain(*_a, **_k):
        return None  # 지음 실패

    stale = {**FP, "terrain": "sheet", "source_filter": "sheet_laplace", "model_digest": "old"}
    db, _ = _swap_world(monkeypatch, tmp_path, sheet_fp=stale, chain=chain)
    before_db, before_files = _no_ids(db), _tree(tmp_path)
    route_ids = _rows(db, "SELECT id FROM routes ORDER BY id")
    sheet_meta = sets.read_meta(tmp_path, "sheet")
    models = _rows(db, "SELECT id, status FROM surface_models ORDER BY id")
    with pytest.raises(RuntimeError):
        asyncio.run(sets.swap_design_set(PID, tmp_path, "sheet"))  # type: ignore[arg-type]
    assert _no_ids(db) == before_db
    assert _rows(db, "SELECT id FROM routes ORDER BY id") == route_ids
    assert _rows(db, "SELECT id, status FROM surface_models ORDER BY id") == models
    assert _tree(tmp_path) == before_files
    assert sets.read_meta(tmp_path, "sheet") == sheet_meta  # 묶음도 그대로
    assert not (tmp_path / SETS_DIR / "las").exists()


# ── 765-3 전처리 체인 = 두 벌 ─────────────────────────────────────────────


def _chain_world(monkeypatch, tmp_path: Path, *, other, built_other=True):
    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_storage as storage
    import config.config_db as config_db

    log: list = []
    expected = tmp_path / "B05_Profile" / "route" / "expected_route.csv"

    async def auto(pid, model_id=None, inner=False):
        log.append(("auto", model_id, inner))
        if not expected.is_file():  # 체인 2.5 — 없을 때만 그 지형으로 자른 예상노선
            expected.parent.mkdir(parents=True, exist_ok=True)
            expected.write_text(str(model_id), encoding="utf-8")
        first = sum(1 for item in log if item[0] == "auto") == 1
        return {"route_id": 1} if (built_other or not first) else None

    async def switch(_c, _pid, terrain):
        log.append(("switch", terrain))
        return 6 if terrain == "sheet" else 5

    async def park(_c, _root, _pid, fingerprint):
        log.append(("park", fingerprint["terrain"]))

    async def clear(_c, _root, _pid):
        log.append("clear")

    async def fingerprint(_c, _pid, _root, terrain):
        return {**FP, "terrain": terrain}

    async def params(_c, _pid):
        return {"source_filter": "classification"}

    monkeypatch.setattr(config_db, "get_db_pool", lambda: _Pool(_Conn(sqlite3.connect(":memory:"))))
    monkeypatch.setattr(repo, "get_project_storage_relative_path", lambda *_a: _done("p"))
    monkeypatch.setattr(storage, "resolve_stored_project_path", lambda _p: str(tmp_path))
    import common_util.common_util_surface_confirmation as confirmation

    monkeypatch.setattr(confirmation, "get_surface_confirmation_params", params)
    monkeypatch.setattr(sets, "other_terrain", lambda _p, _r: other)
    monkeypatch.setattr(sets, "switch_terrain", switch)
    monkeypatch.setattr(sets, "park_active_set", park)
    monkeypatch.setattr(sets, "clear_active_set", clear)
    monkeypatch.setattr(sets, "set_fingerprint", fingerprint)

    async def seed(_pid, _root, surface):
        log.append(("seed", surface["source_filter"]))
        expected.parent.mkdir(parents=True, exist_ok=True)
        expected.write_text("design", encoding="utf-8")

    monkeypatch.setattr(chain, "run_auto_design_chain", auto)
    monkeypatch.setattr(chain, "_seed_expected_route", seed)
    monkeypatch.setattr(router, "schedule_precompute", lambda pid: log.append("precompute"))
    return chain, log, expected


async def _done(value):
    return value


def test_전처리_체인_다른_지형_먼저_떠_두고_설계_지형(monkeypatch, tmp_path: Path) -> None:
    chain, log, expected = _chain_world(monkeypatch, tmp_path, other="sheet")
    (tmp_path / SETS_DIR / "las").mkdir(parents=True)  # 옛 묶음 = 버림
    summary = asyncio.run(chain.run_preprocess_chain(PID, 5))  # type: ignore[arg-type]
    assert summary == {"route_id": 1}
    assert log == [
        ("seed", "classification"),  # 예상노선 = 설계 지형으로 먼저
        ("switch", "sheet"),
        ("auto", 6, True),
        ("park", "sheet"),
        ("switch", "las"),
        ("auto", 5, True),
        "precompute",
    ]
    assert expected.read_text(encoding="utf-8") == "design"  # 두 벌 같은 평면 노선
    assert not (tmp_path / SETS_DIR / "las").exists()
    assert not (tmp_path / "initial_design.lock").exists()


def test_전처리_체인_다른_지형_없으면_한_벌(monkeypatch, tmp_path: Path) -> None:
    chain, log, _ = _chain_world(monkeypatch, tmp_path, other=None)
    asyncio.run(chain.run_preprocess_chain(PID, 5))  # type: ignore[arg-type]
    assert log == [("auto", 5, True), "precompute"]
    meta = sets.read_meta(tmp_path, "sheet")
    assert meta["built"] is False and "sheet" in meta["reason"]


def test_전처리_체인_다른_벌_실패해도_설계_지형(monkeypatch, tmp_path: Path) -> None:
    chain, log, _ = _chain_world(monkeypatch, tmp_path, other="sheet", built_other=False)
    summary = asyncio.run(chain.run_preprocess_chain(PID, 5))  # type: ignore[arg-type]
    assert summary == {"route_id": 1}
    assert log == [
        ("seed", "classification"),
        ("switch", "sheet"),
        ("auto", 6, True),
        "clear",
        ("switch", "las"),
        ("auto", 5, True),
        "precompute",
    ]
    assert sets.read_meta(tmp_path, "sheet")["built"] is False


def test_비우기에_묶음_포함(tmp_path: Path) -> None:
    from common_util.common_util_project_reset import purge_output_folders

    (tmp_path / SETS_DIR / "sheet" / "files").mkdir(parents=True)
    (tmp_path / "B04_PreProcess" / "models").mkdir(parents=True)
    removed = purge_output_folders(tmp_path, keep_preprocess=True)
    assert SETS_DIR in removed and not (tmp_path / SETS_DIR).exists()
    assert (tmp_path / "B04_PreProcess" / "models").is_dir()


# ── 765-6 초기값 사본 = 캐시 뺌 · 복제가 묶음 안 지표면 id 고침 ────────────


def test_초기값_사본은_배수_캐시를_빼고_복원은_캐시를_안_지움(tmp_path: Path) -> None:
    from common_util import common_util_initial_snapshot as snap

    _seed_files(tmp_path)
    drainage = tmp_path / "B04_PreProcess" / "drainage"
    (drainage / "rainfall_stations").mkdir()
    (drainage / "rainfall_stations" / "a.json").write_text("st", encoding="utf-8")
    target = tmp_path / "initial_snapshot" / "B04_PreProcess__drainage"
    snap._save_tree(drainage, target, snap.DRAINAGE_SHARED_PREFIXES)
    saved = sorted(p.name for p in target.iterdir())
    assert saved == ["01_primary_region.geojson", "edits", "manifest.json"]  # 캐시 없음

    # 작업본이 바뀜 · 캐시는 새로 지어짐 · 옛 사본(캐시 듦) 꼴도 섞음
    (drainage / "01_primary_region.geojson").write_text("edited", encoding="utf-8")
    (drainage / "04_detailed_basins.geojson").write_text("new", encoding="utf-8")
    (drainage / "descent_ab_cd.npz").write_text("d2", encoding="utf-8")
    (target / "00_sheet_metric.pkl").write_text("old-cache", encoding="utf-8")
    snap._restore_tree(target, drainage, snap.DRAINAGE_SHARED_PREFIXES)
    live = {p.name: p for p in drainage.iterdir()}
    assert live["01_primary_region.geojson"].read_text(encoding="utf-8") == "{}"
    assert "04_detailed_basins.geojson" not in live  # 초기값에 없던 노선 몫 = 지움
    assert live["descent_ab_cd.npz"].read_text(encoding="utf-8") == "d2"  # 캐시 그대로
    assert live["00_sheet_metric.pkl"].read_text(encoding="utf-8") == "s"  # 옛 사본 캐시 안 덮음
    assert (live["rainfall_stations"] / "a.json").is_file()
    assert (live["edits"] / "pipe_points.json").is_file()


def test_복제가_묶음_안_지표면_id_고침(tmp_path: Path) -> None:
    from common_util.common_util_project_copy import _fix_design_sets

    db = _seed_db()
    _seed_files(tmp_path)
    asyncio.run(sets.park_active_set(_Conn(db), tmp_path, PID, FP))
    bundle = tmp_path / SETS_DIR / "las"
    corridor = bundle / "files" / "B05_Profile" / "corridor" / "corridor_0011.json"
    report = _fix_design_sets(tmp_path, {5: 905})
    assert sorted(report["las"]) == ["db.json", "meta.json", "sampling"]
    dump = json.loads((bundle / "db.json").read_text(encoding="utf-8"))
    assert {r["surface_model_id"] for r in dump["routes"]} == {905}
    assert [r["id"] for r in dump["routes"]] == [10, 11]  # 노선 id 는 꺼낼 때 새로
    assert sets.read_meta(tmp_path, "las")["model_id"] == 905
    sampling = bundle / "files" / "B06_Section" / "sampling.json"
    assert json.loads(sampling.read_text(encoding="utf-8")) == {"surface_model_id": 905}
    assert corridor.is_file()  # 코리도는 안 건드림
    assert sets.bundle_ready(tmp_path, "las", FP)  # 지문은 id 를 안 봄


def test_초기값_사본은_비용면_예상노선도_빼고_복원이_안_건드림(tmp_path: Path) -> None:
    from common_util import common_util_initial_snapshot as snap

    _seed_files(tmp_path)
    asyncio.run(_save(snap, tmp_path))
    stored = tmp_path / "initial_snapshot" / "B05_Profile__route"
    assert not any(p.name.startswith(("cost_surface_", "expected_route")) for p in stored.iterdir())
    route = tmp_path / "B05_Profile" / "route"
    (route / "cost_surface_x_implicit.npz").write_text("c2", encoding="utf-8")
    (route / "plus_stations.json").write_text("[]", encoding="utf-8")
    (route / "expected_route.csv").write_text("other-terrain", encoding="utf-8")
    snap.restore_snapshot_files(tmp_path)
    assert (route / "cost_surface_x_implicit.npz").read_text(encoding="utf-8") == "c2"
    # 예상노선 = 공유 — 다른 벌 [초기화]가 덮지 않음
    assert (route / "expected_route.csv").read_text(encoding="utf-8") == "other-terrain"
    assert not (route / "plus_stations.json").exists()  # 초기값에 없던 설계 파일 = 지움
    assert (route / "structures.json").read_text(encoding="utf-8") == '{"items": [1]}'


async def _save(snap, root: Path) -> None:
    (root / "initial_snapshot").rename(root / "_old_snapshot")  # 시드 사본 비켜 둠
    db = _seed_db()
    await snap.save_initial_snapshot(_Conn(db), root, 11)


def test_벌_예정노선_그_지형으로_잘라_한_번_적음(monkeypatch, tmp_path: Path) -> None:
    from B03_FileInput import B03_FileInput_Service_Chain as chain
    from common_util import common_util_route_geometry as geometry

    calls: list[str] = []

    def load(_root, surface, _range, *, from_source=False):
        calls.append(f"{surface['source_filter']}:{from_source}")
        n = 5 if surface["source_filter"] == "sheet_laplace" else 3  # 수치지형도 = 끝까지
        return SimpleNamespace(vertices=[SimpleNamespace(x=float(i), y=0.0) for i in range(n)])

    monkeypatch.setattr(geometry, "load_design_route", load)
    geometry.write_route_csv(geometry.expected_route_csv_path(tmp_path), [{"x": 0, "y": 0}] * 2)
    sheet = {"source_filter": "sheet_laplace", "method": "dtm", "smooth": True}
    points = chain._terrain_route_points(tmp_path, sheet, None)
    assert len(points) == 5 and calls == ["sheet_laplace:True"]  # 공유 예상노선을 안 씀
    own = geometry.terrain_route_csv_path(tmp_path)
    assert len(geometry.read_planned_route_csv(own).vertices) == 5
    assert geometry.route_source_path(tmp_path) == own
    las = {"source_filter": "classification", "method": "implicit", "smooth": False}
    assert len(chain._terrain_route_points(tmp_path, las, None)) == 5  # 있으면 그것(다시 안 자름)
    assert calls == ["sheet_laplace:True"]


def test_초기_폴리라인_바탕은_벌_예정노선(tmp_path: Path) -> None:
    import os

    from B05_Profile import B05_Profile_Router_Replan as replan
    from common_util import common_util_route_geometry as geometry

    own = [{"x": float(x), "y": 0.0} for x in range(0, 401, 10)]
    geometry.write_route_csv(geometry.terrain_route_csv_path(tmp_path), own)
    assert replan._ensure_planned_initial(tmp_path, 30.0) is not None
    initial = geometry.planned_route_initial_path(tmp_path)
    vertices = geometry.read_planned_route_csv(initial).vertices
    assert (vertices[0].x, vertices[-1].x) == (0.0, 400.0)
    # 공유 예상노선(다른 지형으로 자름)이 나중에 적혀도 벌 초기 폴리라인은 그대로
    expected = geometry.expected_route_csv_path(tmp_path)
    geometry.write_route_csv(expected, own[:10])
    later = initial.stat().st_mtime + 5
    os.utime(expected, (later, later))
    assert replan._ensure_planned_initial(tmp_path, 30.0) is None