# ruff: noqa: E501 — 견본 상자 · 선 덩어리는 한눈에 보이게 한 줄로 둠
"""단가산출 다이어그램 시험 길(`M02_MasterTemplete/DataFlow/M02_DF_Router.py`) 시험 — PLAN 35-7.

· Model CfLogic(boxes · links · userElements) 저장 → 열기 → 목록 → 지우기 · 견본(`_sample`)은 읽기만
· 저장 때 입력 값을 마스터에서 다시 받고 함수 상자 시트를 한 통합문서로 Node 셈 →
  받는 칸에 값만 되돌림
  (입력 → 함수 · 함수 → 함수 직렬 · 표 블록 · 변수 · 출력 상자 비용 종류 · 원가)
· 입력 값 — 인력 · 재료(키 · 검색어 묶음) · 유가전력 · 환율 · 요율 · 기계 값 칸 · 표 블록 ·
  한 줄 · 사용자 요소
· 사용자 요소 만들기 · 지우기 · 권한 없음 403 · 마스터 사본 파일은 안 바뀜
견본 표 · 다이어그램은 시험 안에 둠(임시 폴더 사본에 견본 표를 더함 — 진짜 값에 안 기댐).
"""

from __future__ import annotations

import copy
import json
import shutil
import subprocess
from decimal import Decimal
from pathlib import Path

import pytest
from fastapi import Depends, FastAPI
from fastapi.testclient import TestClient

from common_util.common_util_auth import require_system_admin, verify_session
from M01_MasterData import M01_MasterData_Router as m01_router
from M01_MasterData import M01_MasterData_Store as store
from M01_MasterData import M01_MasterData_Store_Flow as sflow
from M02_MasterTemplete.DataFlow import M02_DF_Router as cf
from M02_MasterTemplete.DataFlow import M02_DF_Values as dv

ROOT = Path(__file__).resolve().parents[2]
REAL = store.FOLDER
OWN = "로직_자체.json"
SAMPLE = "QF999995"
TABLE = {
    "키": SAMPLE,
    "원문번호": "",
    "구분": "",
    "상세구분": "",
    "이름": "견본 높이 표",
    "기준": "1㎡",
    "출처": "자체",
    "조건": {"돌": "고르기", "높이": "범위"},
    "범위규칙": "초과·이하",
    "값칸": {"석공": "인", "증가율": "%"},
    "줄": [
        {"돌": "깬돌", "높이": [None, 3], "석공": 0.5, "증가율": [80, 100]},
        {"돌": "견치돌", "높이": 4, "석공": 0.7},
    ],
    "주": [],
}
HEAD = ["돌", "높이·아래", "높이·위", "석공", "증가율·아래", "증가율·위"]
NO_NODE = shutil.which("node") is None


def _first(name: str, pred) -> dict:
    rows = json.loads((store.FOLDER / name).read_text(encoding="utf-8"))["줄"]
    return next(r for r in rows if pred(r))


def _labor() -> tuple[str, Decimal]:
    row = _first(
        "인력.json",
        lambda r: (
            r.get("상태") == "공표" and not r.get("준용") and isinstance(r.get("값"), (int, float))
        ),
    )
    return row["키"], Decimal(str(row["값"]))


def _box(id_: str, kind: str, name: str, **part) -> dict:
    return {"id": id_, "kind": kind, "name": name, "color": "", "unit": "-", "x": 0, "y": 0, **part}


def _func(sheet: dict, inputs: list, output: dict, variables: list | None = None) -> dict:
    return {"sheet": sheet, "inputs": inputs, "variables": variables or [], "output": output}


def _link(i: str, a: str, pa: str, b: str, pb: str) -> dict:
    return {"id": i, "from": {"box": a, "name": pa}, "to": {"box": b, "name": pb}}


def _logic(labor: str, key: str | None = None) -> dict:
    """노임 × 2 × 면적 × (1 + 할증%) = 노무비 → +1 = 경비(함수 → 함수) · 표 블록 E1 → H3 = 품(값)
    · 브라우저가 보낸 받는 칸 값(999)은 엉터리."""
    s1 = {"id": "s1", "이름": "노무", "칸": {"A1": {"값": 999}, "B1": {"식": "A1*2*A2*(1+A3/100)"}}}
    group = {"mode": "group", "elements": [], "variables": []}
    return {
        "key": key,
        "name": "견본 다이어그램",
        "resultUnit": "원/㎡",
        "boxes": [
            _box("in1", "input", "인력", input={**group, "group": "인력",
                 "elements": [{"ref": labor, "name": "석공", "spec": "", "unit": "인"}]}),
            _box("in2", "input", "표",
                 input={**group, "group": "소요량", "table": {"key": SAMPLE}}),
            _box("in3", "input", "면적", input={"mode": "variable", "elements": [], "variables": [
                {"name": "면적", "type": "number", "unit": "㎡", "example": 3}]}),
            _box("f1", "function", "노무 셈", func=_func(
                s1, [{"port": "노임", "cell": "A1"}, {"port": "면적", "cell": "A2"}],
                {"cell": "B1", "name": "노무비", "unit": "원"},
                [{"name": "할증", "type": "number", "unit": "%", "example": 10, "cell": "A3"}])),
            _box("f2", "function", "더하기", func=_func(
                {"id": "s2", "이름": "더하기", "칸": {"B1": {"식": "A1+1"}}},
                [{"port": "앞", "cell": "A1"}], {"cell": "B1", "name": "합", "unit": "원"})),
            _box("f3", "function", "표 찾기", func=_func(
                {"id": "s3", "이름": "표 찾기", "칸": {"A1": {"식": "H3"}}},
                [{"port": "표", "cell": "E1"}], {"cell": "A1", "name": "품", "unit": "인"})),
            _box("o1", "output", "노무비", output={"cost": "노무비", "unit": "원"}),
            _box("o2", "output", "경비", output={"cost": "경비", "unit": "원"}),
            _box("o3", "output", "품", output={"cost": "값", "unit": "인"}),
        ],
        "links": [
            _link("l1", "in1", "석공", "f1", "노임"),
            _link("l2", "in3", "면적", "f1", "면적"),
            _link("l3", "f1", "노무비", "f2", "앞"),
            _link("l4", "in2", "표", "f3", "표"),
            _link("l5", "f1", "노무비", "o1", "값"),
            _link("l6", "f2", "합", "o2", "값"),
            _link("l7", "f3", "품", "o3", "값"),
        ],
        "userElements": [],
    }  # fmt: skip


@pytest.fixture
def folder(tmp_path: Path, monkeypatch) -> Path:
    master = tmp_path / "master"
    master.mkdir()
    for path in REAL.glob("*.json"):
        if path.name == OWN:
            continue
        if not path.name.startswith("_") or path.name == store.mk.BOOK.name:
            shutil.copy(path, master / path.name)
    host = sorted(master.glob("소요량_*.json"))[0]
    data = json.loads(host.read_text(encoding="utf-8"))
    data["표"].append(TABLE)
    host.write_text(json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8")
    monkeypatch.setattr(store, "FOLDER", master)
    return master


@pytest.fixture
def client(folder: Path) -> TestClient:
    app = FastAPI()
    app.include_router(m01_router.router)
    app.include_router(cf.router)
    return TestClient(app)


def _post(client: TestClient, url: str, body: dict, want: int = 200) -> dict:
    res = client.post(url, json=body)
    assert res.status_code == want, res.text
    return res.json()


def _d(v) -> Decimal:
    return Decimal(str(v))


def _save(logic: dict) -> dict:
    """서버 셈(`M02_DF_Compute.run`) — 받는 칸에 값을 되돌린 흐름 · 값(흐름 저장은 M01 로직 줄 「흐름」 칸)."""
    got = cf.dc.run(logic, {}, cf._load, logic.get("userElements") or [], [logic.get("key") or ""])
    return {"key": logic.get("key"), "logic": got.pop("filled"), "values": got}


def _store(logic: dict) -> str:
    """흐름을 마스터 사본의 새 자체 로직 줄 「흐름」 칸에 저장(M01 `/save` 와 같은 길) — 다른 로직 입력이 부름."""
    own = store.FOLDER / "로직_자체.json"
    version = store.read(own.name)[1] if own.is_file() else ""
    row = {
        "키": "",
        "이름": logic["name"],
        "결과단위": logic.get("resultUnit") or "원",
        "흐름": {**logic, "key": None},
    }
    done = sflow.save(
        [{"file": own.name, "version": version, "changes": [{"op": "add", "row": row}]}]
    )
    return done["added"][0]["key"]


@pytest.mark.skipif(NO_NODE, reason="node 가 없음")
def test_저장_서버가_다시_셈_값만_되돌림(client: TestClient, folder: Path) -> None:
    labor, price = _labor()
    before = {p.name: p.read_bytes() for p in folder.glob("*.json")}

    saved = _save(_logic(labor))
    v = saved["values"]
    labor_cost = price * 2 * 3 * Decimal("1.1")
    assert _d(v["ports"]["f1"]["노무비"]) == labor_cost
    assert _d(v["ports"]["f2"]["합"]) == labor_cost + 1  # 함수 → 함수(직렬)
    assert _d(v["ports"]["f3"]["품"]) == Decimal("0.7")  # 표 블록 E1 부터 → H3
    assert _d(v["cost"]["노무비"]) == labor_cost and _d(v["cost"]["경비"]) == labor_cost + 1
    assert _d(v["cost"]["계"]) == 2 * labor_cost + 1 and _d(v["cost"]["값들"]["품"]) == Decimal(
        "0.7"
    )
    assert v["errors"] == []

    boxes = {b["id"]: b for b in saved["logic"]["boxes"]}
    c1 = boxes["f1"]["func"]["sheet"]["칸"]
    assert c1["A1"]["값"] == price and c1["A2"]["값"] == 3 and c1["A3"]["값"] == 10  # 999 는 버림
    assert c1["B1"] == {"식": "A1*2*A2*(1+A3/100)"} and boxes["f1"]["func"]["sheet"]["id"] == "s1"
    assert _d(boxes["f2"]["func"]["sheet"]["칸"]["A1"]["값"]) == labor_cost  # 참조 아닌 값으로
    c3 = boxes["f3"]["func"]["sheet"]["칸"]
    assert [c3[f"{c}1"]["값"] for c in "EFGHIJ"] == HEAD and c3["F3"]["값"] == c3["G3"]["값"] == 4
    assert saved["logic"]["links"] == _logic(labor)["links"]

    assert {p.name: p.read_bytes() for p in folder.glob("*.json")} == before  # 마스터 사본 그대로


@pytest.mark.skipif(NO_NODE, reason="node 가 없음")
def test_빈_입력_값은_쉬운_말(client: TestClient) -> None:
    labor, _ = _labor()
    logic = _logic(labor)
    logic["boxes"][0]["input"]["elements"][0]["ref"] = "LB999999"
    v = _save(logic)["values"]
    assert v["errors"] and "석공" in v["errors"][0]


def test_입력_값_그룹마다(client: TestClient) -> None:
    labor, price = _labor()
    mat = _first(
        "재료_자재품목.json",
        lambda r: isinstance(r.get("물가자료"), (int, float)) and r.get("구분"),
    )
    word = f"{mat['이름']} {mat['규격'] or ''}".strip()
    cond = {"구분": mat["구분"], "상세구분": mat["상세구분"], "검색어": word}
    line = {"종류": "재료", "요소": cond, "이름": "재료", "단위": mat["단위"], "수량": "1",
            "비목": "재료비"}  # fmt: skip
    row = {"키": "", "원문번호": "", "이름": "견본", "결과단위": "원/㎡", "출처": "자체",
           "입력": [], "중간": [], "호표": [line], "덧줄": [], "끝수": None}  # fmt: skip
    want = _post(client, "/api/m01/calc", {"key": "", "inputs": {}, "row": row})["lines"][0]["단가"]
    oil = _first("재료_유가전력.json", lambda r: isinstance(r.get("값"), (int, float)))
    fx = _first("환율_한국은행환율.json", lambda r: isinstance(r.get("값"), (int, float)))
    rate = _first("요율_조달청제비율.json", lambda r: isinstance(r.get("값"), dict))
    eq = _first("기계.json", lambda r: isinstance((r.get("값") or {}).get("가격"), (int, float)))
    local = [{"id": "UE-0007", "group": "재료", "name": "현장 자재", "spec": "", "unit": "개",
              "value": 500}]  # fmt: skip
    items = [
        {"id": "l", "group": "인력", "ref": labor},
        {
            "id": "m",
            "group": "재료",
            "ref": json.dumps(cond, ensure_ascii=False),
            "unit": mat["단위"],
        },
        {"id": "o", "group": "유가전력", "ref": oil["키"]},
        {"id": "f", "group": "환율", "ref": fx["키"]},
        {"id": "r", "group": "요율", "ref": rate["키"]},
        {"id": "e", "group": "기계", "ref": eq["키"], "field": "가격"},
        {"id": "e2", "group": "기계", "ref": eq["키"]},
        {"id": "t", "group": "소요량", "table": {"key": SAMPLE}},
        {"id": "t1", "group": "소요량", "table": {"key": SAMPLE, "row": 1}},
        {"id": "u", "group": "재료", "ref": "UE-0007"},
        {"id": "w", "group": "인력", "ref": oil["키"]},
        {"id": "z", "group": "모름", "ref": ""},
    ]
    body = {"items": items, "userElements": local}
    got = {v["id"]: v for v in _post(client, "/api/m02/df/values", body)["values"]}
    assert _d(got["l"]["value"]) == price
    assert got["m"]["value"] == want  # 검색어 묶음 = 기존 /calc 재료 줄 단가
    assert got["o"]["value"] == oil["값"] and got["f"]["value"] == fx["값"]
    assert got["r"]["value"] == rate["값"]["rate_percent"] and got["r"]["unit"] == "%"
    assert got["e"]["value"] == eq["값"]["가격"]
    ruled = [*HEAD, dv.RANGE_RULE]  # 끝 칸 = 줄 경계 말(범위규칙 없는 표 = 초과·이하)
    assert got["t"]["rows"][0] == ruled
    assert got["t1"]["rows"] == [ruled, ["견치돌", 4, 4, 0.7, None, None, "초과·이하"]]
    assert got["u"]["value"] == 500 and got["u"]["unit"] == "개"
    for k in ("w", "z", "e2"):  # 기계 묶음은 칸을 골라야 함
        assert got[k]["ok"] is False and got[k]["reason"], got[k]


def test_사용자_요소는_흐름_안_것만(client: TestClient, folder: Path) -> None:
    before = {p.name: p.read_bytes() for p in folder.glob("*.json")}
    mine = [{"id": "UE-0002", "group": "재료", "name": "나", "spec": "", "unit": "회", "value": None},
            {"id": "UE-0003", "group": "재료", "name": "다", "spec": "", "unit": "개", "value": 2.5}]  # fmt: skip
    items = [
        {"id": "x", "group": "재료", "ref": "UE-0002"},
        {"id": "y", "group": "재료", "ref": "UE-0003"},
    ]
    got = _post(client, "/api/m02/df/values", {"items": items, "userElements": mine})["values"]
    assert got[0]["ok"] is False and _d(got[1]["value"]) == Decimal("2.5")  # 빈 값은 쉬운 말
    assert client.get("/api/m02/df/user-elements").status_code in (404, 405)  # 공용 목록 길 없음
    assert {p.name: p.read_bytes() for p in folder.glob("*.json")} == before  # 마스터에 안 씀


def test_시스템_관리자가_아니면_403(folder: Path) -> None:
    app = FastAPI()
    app.include_router(
        cf.router, dependencies=[Depends(verify_session), Depends(require_system_admin)]
    )
    app.dependency_overrides[verify_session] = lambda: {"role": "USER", "company_id": 1}
    client = TestClient(app)
    assert client.post("/api/m02/df/values", json={"items": []}).status_code == 403
    assert client.get("/api/m02/df/design-inputs", params={"key": "GF000219"}).status_code == 403
    app.dependency_overrides[verify_session] = lambda: {"role": "SYSTEM_ADMIN", "company_id": None}
    assert client.post("/api/m02/df/values", json={"items": []}).json() == {"values": []}


@pytest.mark.skipif(NO_NODE, reason="node 가 없음")
def test_기계_키_칸_꼴과_시간당_경비(client: TestClient) -> None:
    """입력 판(sub3) 기계 ref 「EQ키#칸」 · 물음 {ref, field} · 「시간당 경비」 = 기계경비 로직."""
    logic = min(
        (r for r in dv._machine_family() if "지역" in [i["이름"] for i in r["입력"]]),
        key=dv._machine_rank,
    )
    code = str(dv._codes(logic)[0])
    eq = _first("기계.json", lambda r: str(r.get("원문번호")) == code)
    family = [r for r in dv._machine_family() if code in [str(c) for c in dv._codes(r)]]
    chosen = min(family, key=dv._machine_rank)
    given = {
        i["이름"]: (code if i["이름"] == "기계" else (i.get("고르기") or [None])[0])
        for i in chosen["입력"]
    }
    want = _post(client, "/api/m01/calc", {"key": chosen["키"], "inputs": given})["sums"]["계"]

    items = [
        {"id": "p", "group": "기계", "ref": f"{eq['키']}#가격"},
        {"id": "p2", "group": "기계", "ref": eq["키"], "field": "가격"},
        {"id": "h", "group": "기계", "ref": f"{eq['키']}#시간당 경비"},
        {"id": "h2", "group": "기계", "ref": eq["키"], "field": "시간당 경비"},
        {"id": "x", "group": "기계", "ref": f"{eq['키']}#없는칸"},
    ]
    got = {v["id"]: v for v in _post(client, "/api/m02/df/values", {"items": items})["values"]}
    assert got["p"]["value"] == got["p2"]["value"] == eq["값"]["가격"]
    assert got["h"]["value"] == got["h2"]["value"] == want and got["h"]["unit"] == "원/시간"
    assert got["x"]["ok"] is False and "없는칸" not in got["x"].get("value", "")

    # 저장 — 입력 상자 요소 ref 그대로(「EQ키#칸」)
    box = _box("m1", "input", "기계", input={"mode": "group", "group": "기계", "variables": [],
               "elements": [{"ref": f"{eq['키']}#시간당 경비", "name": "기계 · 시간당 경비",
                             "spec": "", "unit": "원/시간"}]})  # fmt: skip
    f1 = _box("f1", "function", "곱", func=_func(
        {"id": "s1", "이름": "곱", "칸": {"B1": {"식": "A1*2"}}},
        [{"port": "단가", "cell": "A1"}],
        {"cell": "B1", "name": "경비", "unit": "원"}))  # fmt: skip
    o1 = _box("o1", "output", "경비", output={"cost": "경비", "unit": "원"})
    body = {"key": None, "name": "기계 견본", "resultUnit": "원/시간", "boxes": [box, f1, o1],
            "links": [_link("a", "m1", "기계 · 시간당 경비", "f1", "단가"),
                      _link("b", "f1", "경비", "o1", "값")], "userElements": []}  # fmt: skip
    saved = _save(body)
    assert _d(saved["values"]["cost"]["경비"]) == 2 * _d(want)


# ── 35-9 넓힘 — 재료 셋 · 자재 찾기 · 다른 로직 · 설계 때 정할 것 · 선 규칙 · 끝수 · 표준 표 ─────
PIPES = [  # 견본 자재(사본 자재품목에만 더함)
    {"이름": "견본관", "규격": "D300", "물가자료": 30000},
    {"이름": "견본관", "규격": "D450", "물가자료": 45000},
    {"이름": "견본관", "규격": "D450 B종", "물가자료": 47000},
]


def _add_pipes(folder: Path) -> list[str]:
    path = folder / "재료_자재품목.json"
    data = json.loads(path.read_text(encoding="utf-8"))
    cols = list(data["줄"][0])
    keys = []
    for i, p in enumerate(PIPES):
        row = dict.fromkeys(cols)
        row.update(
            {"키": f"MT99999{i}", "구분": "시험구분", "상세구분": "시험상세", "단위": "m", **p}
        )
        data["줄"].append(row)
        keys.append(row["키"])
    path.write_text(json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8")
    return keys


def _slot(**slot) -> dict:
    return {"id": "s", "group": "재료", "slot": slot}


def test_재료_세_방식(client: TestClient, folder: Path) -> None:
    keys = _add_pipes(folder)
    scope = {"구분": "시험구분", "상세구분": "시험상세", "찾는말": "견본관"}
    ask = [
        {**_slot(mode="fixed", ref=keys[0]), "id": "fixed"},
        {**_slot(mode="choice", default=keys[0]), "id": "choice"},
        {**_slot(mode="choice", default=keys[0]), "id": "picked", "pick": keys[1]},
        {**_slot(mode="choice"), "id": "none"},
        {**_slot(mode="rule", scope=scope, field="규격", op="포함"), "id": "one", "value": "D300"},
        {**_slot(mode="rule", scope=scope, field="규격", op="포함"), "id": "many", "value": "D450"},
        {**_slot(mode="rule", scope=scope, field="규격", op="같음"), "id": "same", "value": "D450"},
        {
            **_slot(mode="rule", scope=scope, field="이름", op="포함"),
            "id": "zero",
            "value": "없는관",
        },
    ]
    got = {v["id"]: v for v in _post(client, "/api/m02/df/values", {"items": ask})["values"]}
    assert got["fixed"]["value"] == got["choice"]["value"] == 30000
    assert got["picked"]["value"] == 45000 and got["none"]["ok"] is False
    assert got["one"]["value"] == 30000 and got["same"]["value"] == 45000
    assert got["many"]["ok"] is False and "2개" in got["many"]["reason"]
    assert [c["spec"] for c in got["many"]["candidates"]] == ["D450", "D450 B종"]
    assert got["zero"]["ok"] is False and got["zero"]["reason"].startswith("맞는 재료 없음")


def test_자재_찾기_띄어쓰기_AND_쪽(client: TestClient, folder: Path) -> None:
    keys = _add_pipes(folder)

    def find(**q) -> dict:
        res = client.get("/api/m02/df/materials", params=q)
        assert res.status_code == 200, res.text
        return res.json()

    assert find(q="견본관 D450")["total"] == 2
    assert [x["ref"] for x in find(q="견본관 b종")["items"]] == [keys[2]]
    assert find(q=f"{keys[1]}")["total"] == 1  # 키로도
    page = find(q="견본관", sub="시험구분", detail="시험상세", page=2, size=2)
    assert page["total"] == 3 and page["page"] == 2 and len(page["items"]) == 1
    one = page["items"][0]
    assert one == {"ref": keys[2], "name": "견본관", "spec": "D450 B종", "unit": "m", "price": 47000,
                   "구분": "시험구분", "상세구분": "시험상세"}  # fmt: skip


def _area(x: int = 100) -> dict:
    """면적(예시 2) × x = 결과(값) — 다른 로직이 부를 견본."""
    return {
        "key": None,
        "name": "면적 곱",
        "resultUnit": "-",
        "boxes": [
            _box("v", "input", "면적", input={"mode": "variable", "elements": [], "variables": [
                {"name": "면적", "type": "number", "unit": "㎡", "example": 2}]}),
            _box("f", "function", "곱", func=_func({"id": "s", "이름": "곱", "칸": {"B1": {"식": f"A1*{x}"}}},
                 [{"port": "면적", "cell": "A1"}], {"cell": "B1", "name": "곱", "unit": "-"})),
            _box("o", "output", "결과", output={"cost": "값", "unit": "-"}),
        ],
        "links": [_link("a", "v", "면적", "f", "면적"), _link("b", "f", "곱", "o", "값")],
        "userElements": [],
    }  # fmt: skip


@pytest.mark.skipif(NO_NODE, reason="node 가 없음")
def test_다른_로직_값_설계_값_순환(client: TestClient) -> None:
    a = _store(_area())
    assert _post(client, "/api/m02/df/logic-value", {"key": a})["outputs"] == {"결과": 200}
    assert _post(client, "/api/m02/df/logic-value", {"key": a, "designs": {"면적": 5}})[
        "outputs"
    ] == {"결과": 500}

    ref = {"key": a, "name": "면적 곱", "outputs": [{"name": "결과", "unit": "-"}],
           "designs": [{"name": "면적", "unit": "㎡"}]}  # fmt: skip
    b = {
        "key": None, "name": "다른 로직 잇기", "resultUnit": "원",
        "boxes": [
            _box("L", "input", "면적 곱", input={"mode": "logic", "elements": [], "variables": [], "logic": ref}),
            _box("o", "output", "경비", output={"cost": "경비", "unit": "원"}),
        ],
        "links": [_link("a", "L", "결과", "o", "값")],
        "userElements": [],
    }  # fmt: skip
    saved = _save({**b, "key": _store(b)})
    assert (
        saved["values"]["ports"]["L"] == {"결과": 200}
        and _d(saved["values"]["cost"]["경비"]) == 200
    )
    designs = client.get("/api/m02/df/design-inputs", params={"key": saved["key"]}).json()[
        "designs"
    ]
    assert [(d["box"], d["port"], d["kind"]) for d in designs] == [("L", "면적", "다른 로직")]

    # 순환 — 로직이 자기를 부름
    loop = copy.deepcopy({**b, "key": saved["key"]})
    loop["boxes"][0]["input"]["logic"] = {**ref, "key": saved["key"]}
    got = _save(loop)["values"]
    assert any("서로를 부름" in e for e in got["errors"])


@pytest.mark.skipif(NO_NODE, reason="node 가 없음")
def test_선_규칙_끝수_재료_조건_설계_때_정할_것(client: TestClient, folder: Path) -> None:
    """Eval 과 같은 뜻 — 규칙은 값 기준 · 막힌 선(통과 안 됨 · 표에 없는 조건)은 받는 상자를 멈춤."""
    keys = _add_pipes(folder)
    scope = {"구분": "시험구분", "상세구분": "시험상세", "찾는말": "견본관"}
    vars_ = [{"name": "돌", "type": "choice", "unit": "-", "example": "깬돌"},
             {"name": "높이", "type": "number", "unit": "m", "example": 2},
             {"name": "관경", "type": "text", "unit": "-", "example": "D300"}]  # fmt: skip
    pipe = {"ref": "", "name": "관", "slot": "관 자리", "spec": "", "unit": "m", "mode": "rule", "scope": scope,
            "rule": {"field": "규격", "op": "포함", "from": {"box": "v", "name": "관경"}}}  # fmt: skip
    pick = {"ref": "", "name": "관2", "slot": "골라 쓰는 관", "spec": "", "unit": "m", "mode": "choice",
            "default": keys[1]}  # fmt: skip
    sheet = {"id": "s", "이름": "셈", "칸": {"B1": {"식": "A1+A2+A3"}}}
    lg = {
        "key": None, "name": "규칙 견본", "resultUnit": "원",
        "boxes": [
            _box("v", "input", "값들", input={"mode": "variable", "elements": [], "variables": vars_}),
            _box("t", "input", "표", input={"mode": "group", "group": "소요량", "elements": [],
                 "variables": [], "table": {"key": SAMPLE}}),
            _box("m", "input", "재료", input={"mode": "group", "group": "재료", "elements": [pipe, pick],
                 "variables": []}),
            _box("f", "function", "셈", func=_func(sheet, [{"port": "품", "cell": "A1"},
                 {"port": "관", "cell": "A2"}, {"port": "높이", "cell": "A3"}],
                 {"cell": "B1", "name": "합", "unit": "원", "round": {"places": 0, "method": "버림"}})),
            _box("o", "output", "재료비", output={"cost": "재료비", "unit": "원",
                 "round": {"places": -1, "method": "버림"}}),
            _box("o2", "output", "골라 쓴 관", output={"cost": "값", "unit": "원"}),
            _box("o3", "output", "낮을 때만", output={"cost": "값", "unit": "m"}),
        ],
        "links": [
            {**_link("l1", "t", "표", "f", "품"), "rule": {"kind": "lookup", "col": "석공", "conds": [
                {"col": "돌", "from": {"box": "v", "name": "돌"}}, {"col": "높이", "from": {"box": "v", "name": "높이"}}]}},
            {**_link("l2", "m", "관 자리", "f", "관"), "rule": {"kind": "scale", "op": "+%", "value": 10}},
            {**_link("l3", "v", "높이", "f", "높이"), "rule": {"kind": "gate", "from": {"box": "v", "name": "높이"},
             "op": "≥", "value": 1}},
            _link("l4", "f", "합", "o", "값"),
            {**_link("l5", "m", "골라 쓰는 관", "o2", "값"), "rule": {"kind": "round", "places": -3, "method": "반올림"}},
            {**_link("l6", "v", "높이", "o3", "값"), "rule": {"kind": "gate", "from": {"box": "v", "name": "높이"},
             "op": "<", "value": 1}},
        ],
        "userElements": [],
    }  # fmt: skip
    got = _save(lg)
    v = got["values"]
    assert v["ports"]["m"] == {
        "관 자리": 30000,
        "골라 쓰는 관": 45000,
    }  # 조건 찾기 D300 · 선택 = 기본
    total = Decimal("0.5") + 30000 * Decimal("1.1") + 2  # 깬돌 · 높이 2 → 「~3」 줄 · +10% · 통과
    assert _d(v["ports"]["f"]["합"]) == int(total)  # 함수 출력 끝수 0 자리 버림
    assert _d(v["cost"]["재료비"]) == int(total) // 10 * 10  # 출력 끝수 −1 자리 버림
    assert _d(v["cost"]["값들"]["골라 쓴 관"]) == 45000 and v["cost"]["값들"]["낮을 때만"] is None
    assert [b["kind"] for b in v["breaks"]] == ["통과 안 됨"] and v["errors"] == []
    cells = {b["id"]: b for b in got["logic"]["boxes"]}["f"]["func"]["sheet"]["칸"]
    assert _d(cells["A1"]["값"]) == Decimal("0.5") and cells["B1"] == {
        "식": "A1+A2+A3"
    }  # 받는 칸에 값만

    key = _store(lg)  # 로직 줄 「흐름」 칸에 저장한 뒤 길로 읽음
    designs = client.get("/api/m02/df/design-inputs", params={"key": key}).json()["designs"]
    assert [(d["port"], d["kind"]) for d in designs] == [
        ("돌", "변수"), ("높이", "변수"), ("관경", "변수"), ("골라 쓰는 관", "재료 자리")]  # fmt: skip

    lg["boxes"][0]["input"]["variables"][1]["example"] = 3  # 「~3」 줄(위 끝 포함)
    assert _d(_save(lg)["values"]["ports"]["f"]["합"]) == int(Decimal("0.5") + 33000 + 3)
    lg["boxes"][0]["input"]["variables"][0]["example"] = (
        "견치돌"  # 「높이 4」 줄 = 아래 = 위 → 같음(정함)
    )
    lg["boxes"][0]["input"]["variables"][1]["example"] = 4
    assert _d(_save(lg)["values"]["ports"]["f"]["합"]) == int(Decimal("0.7") + 33000 + 4)
    lookup = {
        "kind": "lookup",
        "col": "석공",
        "conds": [{"col": "높이", "from": {"box": "v", "name": "높이"}}],
    }
    block = [["높이·아래", "높이·위", "석공"], [4, 4, 0.7], [None, 3, 0.5]]
    assert [cf.dc.apply_rule(lookup, block, {"높이": h})[0] for h in (4, 3.5, 3)] == [
        0.7,
        None,
        0.5,
    ]
    lg["boxes"][0]["input"]["variables"][2]["example"] = "D999"  # 맞는 재료 없음
    lg["boxes"][0]["input"]["variables"][0]["example"] = "없는돌"  # 표에 없는 조건 → 함수 멈춤
    bad = _save(lg)["values"]
    assert any("맞는 재료 없음" in e for e in bad["errors"]) and any(
        "표에 없는 조건" in e for e in bad["errors"]
    )
    # 함수 결과 오류 = 오류 값(화면 Eval 과 같음 · 받는 칸으로 번짐)
    assert bad["ports"]["f"]["합"] == "#N/A" and _d(bad["cost"]["재료비"]) == 0


def test_표준_표_읽기_쓰기_바로_저장(client: TestClient, tmp_path: Path, monkeypatch) -> None:
    """고친 표준 표 = `_standard/` 파일에 바로(본 프로그램 값) — 시험은 그 사본 폴더 · 정본 파일은 안 바뀜."""
    real = cf.STANDARD
    before = {p.name: p.read_bytes() for p in real.glob("*.json")}
    shutil.copytree(real, tmp_path / "_standard")
    monkeypatch.setattr(cf, "STANDARD", tmp_path / "_standard")
    first = client.get("/api/m02/df/standards").json()
    assert first["saved"] is False and first["unit"]["줄"] and first["money"]["줄"]
    edited = json.loads(json.dumps(first))
    edited["money"]["줄"][4]["지위"] = 1
    res = client.put(
        "/api/m02/df/standards", json={"unit": edited["unit"], "money": edited["money"]}
    )
    assert res.status_code == 200
    assert client.get("/api/m02/df/standards").json()["money"]["줄"][4]["지위"] == 1
    saved = json.loads((cf.STANDARD / "금액단위표준.json").read_text(encoding="utf-8"))
    assert saved["줄"][4]["지위"] == 1  # 파일에 바로
    assert client.put("/api/m02/df/standards", json={"unit": {}, "money": {}}).status_code == 400
    assert (
        client.delete("/api/m02/df/standards").json()["money"]["줄"][4]["지위"] == 1
    )  # 되돌릴 처음 값 없음
    assert {p.name: p.read_bytes() for p in real.glob("*.json")} == before


def test_새_길도_시스템_관리자만(folder: Path) -> None:
    app = FastAPI()
    app.include_router(
        cf.router, dependencies=[Depends(verify_session), Depends(require_system_admin)]
    )
    app.dependency_overrides[verify_session] = lambda: {"role": "USER", "company_id": 1}
    client = TestClient(app)
    for method, url in (
        ("get", "/api/m02/df/materials"),
        ("get", "/api/m02/df/standards"),
        ("delete", "/api/m02/df/standards"),
        ("get", "/api/m02/df/design-inputs?key=x"),
    ):
        assert getattr(client, method)(url).status_code == 403  # fmt: skip
    assert client.post("/api/m02/df/logic-value", json={"key": "x"}).status_code == 403


def test_자재_찾기_이름_속_띄어쓰기(client: TestClient, folder: Path) -> None:
    """품목 글 · 찾는 낱말 모두 띄어쓰기를 지우고 비교 — 「흄  관(B형)」 도 「흄관」 으로 걸림."""
    path = folder / "재료_자재품목.json"
    data = json.loads(path.read_text(encoding="utf-8"))
    row = dict.fromkeys(data["줄"][0])
    row.update(
        {
            "키": "MT999990",
            "구분": "시험구분",
            "이름": "견  본 관(B형)",
            "규격": "Φ 300mm",
            "물가자료": 1,
        }
    )
    data["줄"].append(row)
    path.write_text(json.dumps(data, ensure_ascii=False), encoding="utf-8")
    for q in ("견본관 300", "견 본관 Φ300", "본관(b형)"):
        got = client.get("/api/m02/df/materials", params={"q": q, "sub": "시험구분"}).json()
        assert [x["ref"] for x in got["items"]] == ["MT999990"], q


RULE_RUNNER = r"""
const fs = require("fs"), path = require("path");
const [root, jobs] = [process.argv[1], JSON.parse(process.argv[2])];
const ts = require(path.join(root, "config", "node_modules", "typescript"));
const cache = {};
function load(file) {
  if (cache[file]) return cache[file].exports;
  const m = { exports: {} };
  cache[file] = m;
  const js = ts.transpileModule(fs.readFileSync(file, "utf8"), {
    compilerOptions: { module: ts.ModuleKind.CommonJS, target: ts.ScriptTarget.ES2020 },
  }).outputText;
  const req = (spec) => {
    if (spec === "@ui/ui_template_locale") return { currentLanguageIndex: 0 };
    if (spec.endsWith(".css")) return {};
    const base = spec.startsWith("@ui/") ? path.join(root, "ui_template", spec.slice(4))
      : spec.startsWith("@config/") ? path.join(root, "config", spec.slice(8))
      : spec.startsWith(".") ? path.resolve(path.dirname(file), spec) : null;
    if (!base) return require(spec);
    const hit = [base + ".ts", base].find((c) => fs.existsSync(c) && fs.statSync(c).isFile());
    if (!hit) throw new Error("못 찾음 " + spec);
    return hit.endsWith(".ts") ? load(hit) : require(hit);
  };
  new Function("exports", "module", "require", js)(m.exports, m, req);
  return m.exports;
}
const rule = load(path.join(root, "M02_MasterTemplete", "DataFlow", "M02_DF_Rule.ts"));
console.log(JSON.stringify(jobs.map((j) => { const r = rule.applyRule(j.rule, j.value, j.conds); return [r.value, r.why]; })));
"""


@pytest.mark.skipif(NO_NODE, reason="node 가 없음")
def test_선_규칙_화면_Rule_ts_와_서버가_같은_값() -> None:
    """화면 `M02_DF_Rule.applyRule`(Node 로 그대로) = 서버 `apply_rule` — 값 · 까닭."""
    block = [["돌", "높이·아래", "높이·위", "석공"], ["깬돌", None, 3, 0.5], ["깬돌", 3, 7.5, 0.6],
             ["깬돌", 7.5, None, 0.8], ["견치돌", 4, 4, 0.7], ["견치돌", 4, 4, 0.9]]  # fmt: skip
    look = {"kind": "lookup", "col": "석공", "conds": [{"col": "돌", "from": {"box": "v", "name": "돌"}},
            {"col": "높이", "from": {"box": "v", "name": "높이"}}]}  # fmt: skip
    gate = {"kind": "gate", "from": {"box": "v", "name": "높이"}, "op": "≥", "value": 3}
    jobs = [{"rule": look, "value": block, "conds": {"돌": d, "높이": h}}
            for d, h in (("깬돌", 2), ("깬돌", 3), ("깬돌", 3.1), ("깬돌", 7.5), ("깬돌", 100), ("깬돌", -5),
                         ("견치돌", 4), ("견치돌", 5), ("없는돌", 1))]  # fmt: skip
    jobs += [{"rule": gate, "value": 10, "conds": {"높이": h}} for h in (2, 3, 4)]
    jobs += [{"rule": {"kind": "gate", "from": {"box": "v", "name": "s"}, "op": op, "value": "가"},
              "value": 1, "conds": {"s": s}} for op, s in (("=", "가"), ("≠", "가"), ("=", "나"))]  # fmt: skip
    jobs += [{"rule": {"kind": "scale", "op": op, "value": 7}, "value": 123.45, "conds": {}}
             for op in ("×", "÷", "+%", "-%")]  # fmt: skip
    jobs += [{"rule": {"kind": "round", "places": p, "method": m}, "value": v, "conds": {}}
             for p, m, v in ((0, "반올림", 2.5), (1, "올림", 1.21), (-1, "버림", 1239), (-3, "반올림", 45500),
                             (2, "버림", -1.239), (0, "반올림", -2.5))]  # fmt: skip
    result = subprocess.run(
        ["node", "-e", RULE_RUNNER, str(ROOT), json.dumps(jobs, ensure_ascii=False)],
        capture_output=True, text=True, encoding="utf-8", timeout=120,
    )  # fmt: skip
    assert result.returncode == 0, result.stderr
    screen = json.loads(result.stdout)
    server = [cf.dc.apply_rule(j["rule"], j["value"], j["conds"]) for j in jobs]
    for job, (sv, sw), (cv, cw) in zip(jobs, screen, server):
        same = (sv is None and cv is None) or (
            sv is not None
            and cv is not None
            and abs(Decimal(str(sv)) - Decimal(str(cv))) < Decimal("1e-9")
        )
        assert same and sw == cw, (job, (sv, sw), (cv, cw))