Aislo/resources/tester/test_143_1_b08_linked_snapshots.py

# -*- coding: utf-8 -*-
"""B08 구조물 집계표 — 조합 사본 없으면 없음(PLAN 143-1).

견본(관 지점 넷 + 세월교 둘 + 개거 하나):
  100  관 Φ800 L16 · 사본 있음 · 유출 흙막이 잇기 사본 없음 → 관 칸만 · 유출 흙막이 칸 없음
  200  관 Φ800 L18 · 지정 있으나 사본 폴더가 다른 항목 → 없음
  300  관 Φ800 L20 · 지정 있으나 사본 폴더 없음 → 없음
  400  관 Φ800 L22 · 지정 없음 → 없음
  500  세월교 Φ1000×2련 · 본체 잇기(gford) 사본 있음 → 관 몫 · 월류 폭 셈
  600  세월교 · 잇기 없음 → 관 몫 · pipes 에도 없음
  700  개거 · 지정 없음 → 「사본 없는 시설」 에 셈
"""

from __future__ import annotations

import json
from pathlib import Path

import pytest

from B08_Quantity import B08_Quantity_Router_Summary as summary
from M02_MasterTemplete import M02_Item_Snapshot as snapshots
from M02_MasterTemplete.StructOptions import M02_SO_Definition as definition

ROOT = Path(__file__).resolve().parents[2]


def _item(key: str) -> dict:
    return {"layer": "system", "owner": None, "key": key, "name": key}


def _soil() -> dict:
    return {
        "outlet_type": "기슭막이",
        "outlet_revet_form": "돌쌓기(메)",
        "outlet_revet_height_m": 2,
        "outlet_revet_before_m": 5,
        "outlet_revet_after_m": 5,
    }


POINTS = [
    {
        "chainage_m": 100.0,
        "options": {"pipe_diameter_mm": 800, **_soil()},
        "summary_item": _item("SS000001"),
        "group_items": {"g9lq9uk": _item("SS000002")},
    },
    {"chainage_m": 200.0, "options": {"pipe_diameter_mm": 800}, "summary_item": _item("SS000001")},
    {"chainage_m": 300.0, "options": {"pipe_diameter_mm": 800}, "summary_item": _item("SS000001")},
    {"chainage_m": 400.0, "options": {"pipe_diameter_mm": 800}},
    {
        "chainage_m": 500.0,
        "facility": "ford_bridge",
        "options": {"pipe_diameter_mm": 1000, "pipe_count": 2, "ford_width_m": 12},
        "group_items": {"gford": _item("SS000003")},
    },
    {
        "chainage_m": 600.0,
        "facility": "ford_bridge",
        "options": {"pipe_diameter_mm": 1000, "pipe_count": 1, "ford_width_m": 8},
    },
]
STRUCTURES = [
    {"structure_id": "s-ditch", "type_id": "open_ditch", "placement": "point", "chainage_m": 700},
]
DESIGNS = [
    {"chainage_m": 100.0, "design": {"pipe_length_m": 16}},
    {"chainage_m": 200.0, "design": {"pipe_length_m": 18}},
    {"chainage_m": 300.0, "design": {"pipe_length_m": 20}},
    {"chainage_m": 400.0, "design": {"pipe_length_m": 22}},
    {"chainage_m": 500.0, "design": {"ford_pipe_length_m": 9}},
    {"chainage_m": 600.0, "design": {"ford_pipe_length_m": 7}},
]


def _snapshot(root: Path, owner: str, item: dict, parts: tuple[str, ...] = ()) -> None:
    folder = snapshots.structures_dir(root) / snapshots.folder_name(owner)
    folder.mkdir(parents=True)
    origin = {**item, "경로": ["횡단배수", item["key"], "이름"]}
    (folder / snapshots.MANIFEST_NAME).write_text(
        json.dumps({"원본": origin}, ensure_ascii=False), encoding="utf-8"
    )
    for name in parts:
        (folder / name).write_text("{}", encoding="utf-8")


@pytest.fixture
def root(tmp_path: Path) -> Path:
    found = tmp_path / "proj"
    edits = found / "B04_PreProcess/drainage/edits"
    edits.mkdir(parents=True)
    points = [{**p, "source": "user"} for p in POINTS]
    (edits / "pipe_points.json").write_text(
        json.dumps({"points": points}, ensure_ascii=False), encoding="utf-8"
    )
    route = found / "B05_Profile/route"
    route.mkdir(parents=True)
    (route / "structures.json").write_text(
        json.dumps({"revision": 1, "structures": STRUCTURES}, ensure_ascii=False),
        encoding="utf-8",
    )
    _snapshot(found, "pipe@100.0", _item("SS000001"), ("도면.json", "일위대가.json"))
    _snapshot(found, "pipe@200.0", _item("SS000009"))  # 다른 항목 사본
    _snapshot(found, "pipe@500.0__gford", _item("SS000003"))
    return found


def _run(root: Path) -> dict:
    names = definition.read()["종류"]["pipe"]["이름"]
    return summary.pipe_summary(root, DESIGNS, names, 6.0)


def _cell(out: dict, at: float, cid: str):
    row = next((r for r in out["rows"] if abs(r["chainage_m"] - at) < 0.01), None)
    return None if row is None else row["values"].get(cid)


def test_연결된_사본은_지정_교집합_같은_항목(root):
    linked = snapshots.linked_snapshots(root)
    assert set(linked) == {"pipe@100.0", "pipe@500.0__gford"}
    one = linked["pipe@100.0"]
    assert one["항목"]["key"] == "SS000001" and one["폴더"] == "pipe@100.0"
    assert one["부분"] == ["도면", "일위대가"]  # 있는 파일만
    assert one["경로"] == ["횡단배수", "SS000001", "이름"]


def test_사본_없는_칸은_안_셈(root):
    out = _run(root)
    assert _cell(out, 100, "pipe:len|800") == 16
    assert _cell(out, 100, "pipe:cp|800") == 2
    # 유출 흙막이 잇기는 사본 없음 → 그 칸만 없음(열도 안 섬)
    assert not any(c["id"].startswith("pipe:revet") for c in out["columns"])
    # 다른 항목 사본 · 사본 없음 · 지정 없음 → 줄 없음
    assert [round(r["chainage_m"]) for r in out["rows"]] == [100, 500]
    assert out["total"]["pipe:len|800"] == 16


def test_세월교_관_몫은_본체_잇기_사본으로(root):
    out = _run(root)
    assert _cell(out, 500, "pipe:len|1000") == 18  # 9 × 2련
    assert _cell(out, 500, "pipe:ford|1000|2|width") == 12
    assert _cell(out, 600, "pipe:len|1000") is None
    assert [(round(p["chainage_m"]), p["column"]) for p in out["pipes"]] == [
        (100, "pipe:cp|800"),
        (500, "pipe:cp|1000"),
        (500, "pipe:cp|1000"),
    ]
    width = next(c for c in out["columns"] if c["id"] == "pipe:ford|1000|2|width")
    assert width["combos"] == ["system||SS000003"]


def test_사본_없는_시설_알림_응답에_사본(root):
    out = _run(root)
    # 100(유출 흙막이) · 200 · 300 · 400 · 500(기본 날개벽 잇기) · 600 · 개거 700 = 7곳
    assert not any("wing" in c["id"] for c in out["columns"])
    assert "조합 사본 없는 시설 7곳 — B05 · B06 좌측 패널에서 조합 연결 · [저장]" in out["notes"]
    assert not any("관 연장 없음" in n for n in out["notes"])
    assert set(out["linked"]) == {"pipe@100.0", "pipe@500.0__gford"}


def test_화면_조합_이름_구조물도는_사본에서():
    ui = (ROOT / "B08_Quantity/B08_Quantity_UI_Summary.ts").read_text(encoding="utf-8")
    assert "loadPickerRows" not in ui  # 층 목록 GET 안 씀
    assert "linkedOnly(" in ui
    drawing = (ROOT / "B08_Quantity/B08_Quantity_UI_Drawing.ts").read_text(encoding="utf-8")
    assert "readTemplate" not in drawing  # 층 폴백 없음
    assert "B08_Quantity_Sheets_NoCopy" in drawing


def test_화면_사본_있는_조합만_사본_id():
    """`linkedOnly` — 본체 · 그룹마다 사본 있는 것만 · 하나도 없으면 시설째 뺌 · `combosAt` 사본 id."""
    import re
    import subprocess

    imports = re.compile(r"^import[^;]*;\n", re.MULTILINE)
    sources = [
        ROOT / "common_util" / "common_util_summary_item.ts",
        ROOT / "B08_Quantity" / "B08_Quantity_Summary_Data.ts",
    ]
    things = [
        {"structure_id": "s1", "chainage_m": 40, "summary_item": _item("SS1"),
         "group_items": {"g1": _item("SS3"), "g2": _item("SS4")}},
        {"structure_id": "s2", "chainage_m": 80, "summary_item": _item("SS2")},
    ]  # fmt: skip
    linked = {"s1__g1": {}, "s1": {}}
    body = (
        f"\nconst kept = linkedOnly({json.dumps(things)}, {json.dumps(linked)});"
        "\nconsole.log(JSON.stringify([kept.map((t) => [t.structure_id, Object.keys(t.group_items)]),"
        " combosAt(kept, 40).map((c) => c.copyId)]));"
    )
    script = "\n".join(imports.sub("", p.read_text(encoding="utf-8")) for p in sources) + body
    result = subprocess.run(
        ["node", "--experimental-strip-types", "--no-warnings",
         "--input-type=module-typescript", "-e", script],
        capture_output=True, text=True, encoding="utf-8", timeout=60,
    )  # fmt: skip
    assert result.returncode == 0, result.stderr
    assert json.loads(result.stdout) == [[["s1", ["g1"]]], ["s1", "s1__g1"]]