Aislo/resources/tester/test_m02_cf_router.py

"""단가산출 다이어그램 시험 길(`M02_MasterTemplete/CostFlow/M02_CF_Router.py`) 시험 — PLAN 35-7.

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

from __future__ import annotations

import json
import shutil
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 M02_MasterTemplete.CostFlow import M02_CF_Router as cf

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")
    sample = tmp_path / "_sample"
    sample.mkdir()
    one = {"key": None, "name": "견본 돌쌓기", "resultUnit": "원/㎡", "boxes": [], "links": [],
           "userElements": []}  # fmt: skip
    (sample / "견본 돌쌓기.json").write_text(json.dumps(one, ensure_ascii=False), encoding="utf-8")
    monkeypatch.setattr(store, "FOLDER", master)
    monkeypatch.setattr(cf, "DATA", tmp_path / "_data")
    monkeypatch.setattr(cf, "SAMPLE", sample)
    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))


@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 = _post(client, "/api/m02/cf/save", {"logic": _logic(labor)})
    assert saved["key"] == "CF-0001"
    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"]

    # 열기 · 목록 · 견본 · 고치기 · 지우기
    listed = client.get("/api/m02/cf/list").json()["logics"]
    assert [(x["key"], x["sample"]) for x in listed] == [("CF-0001", False), ("견본 돌쌓기", True)]
    sample = client.get("/api/m02/cf/get", params={"key": "견본 돌쌓기"}).json()
    assert sample["sample"] is True
    assert _post(client, "/api/m02/cf/save", {"logic": sample["logic"]})["key"] == "CF-0002"
    _post(client, "/api/m02/cf/delete", {"key": "견본 돌쌓기"}, want=403)
    renamed = {**saved["logic"], "name": "고친 이름"}
    assert _post(client, "/api/m02/cf/save", {"logic": renamed})["key"] == "CF-0001"
    got = client.get("/api/m02/cf/get", params={"key": "CF-0001"}).json()["logic"]
    assert (
        got["name"] == "고친 이름" and got["boxes"][3]["func"]["sheet"]["칸"]["A1"]["값"] == price
    )
    _post(client, "/api/m02/cf/delete", {"key": "CF-0002"})
    assert client.get("/api/m02/cf/get", params={"key": "CF-0002"}).status_code == 404
    _post(client, "/api/m02/cf/save", {"logic": _logic(labor, "CF-0009")}, want=404)
    _post(client, "/api/m02/cf/save", {"logic": {**_logic(labor), "name": " "}}, want=400)
    _post(client, "/api/m02/cf/save", {"logic": {"name": "x"}}, want=400)

    assert {p.name: p.read_bytes() for p in folder.glob("*.json")} == before  # 마스터 사본 그대로
    assert sorted(p.name for p in cf.DATA.glob("*.json")) == ["CF-0001.json"]


@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 = _post(client, "/api/m02/cf/save", {"logic": 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/cf/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["값"]["가격"]
    assert got["t"]["rows"][0] == HEAD
    assert got["t1"]["rows"] == [HEAD, ["견치돌", 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")}
    url = "/api/m02/cf/user-elements"
    a = _post(client, url, {"op": "add", "element": {"name": "가", "group": "재료", "value": 1.5}})
    b = _post(client, url, {"op": "add", "element": {"name": "나", "unit": "회"}})
    assert [a["element"]["id"], b["element"]["id"]] == ["UE-0001", "UE-0002"]
    assert b["element"]["group"] == "재료" and b["element"]["value"] is None
    _post(client, url, {"op": "add", "element": {"name": " "}}, want=400)
    _post(client, url, {"op": "add", "element": {"name": "다", "value": "x"}}, want=400)
    _post(client, url, {"op": "add", "element": {"name": "라", "group": "모름"}}, want=400)
    _post(client, url, {"op": "delete", "id": "UE-0001"})
    _post(client, url, {"op": "delete", "id": "UE-0001"}, want=404)
    assert [e["id"] for e in client.get(url).json()["elements"]] == ["UE-0002"]
    got = _post(
        client, "/api/m02/cf/values", {"items": [{"id": "x", "group": "재료", "ref": "UE-0002"}]}
    )
    assert got["values"][0]["ok"] is False  # 값이 비어 있으면 쉬운 말
    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.get("/api/m02/cf/list").status_code == 403
    assert client.post("/api/m02/cf/values", json={"items": []}).status_code == 403
    assert client.get("/api/m02/cf/user-elements").status_code == 403
    app.dependency_overrides[verify_session] = lambda: {"role": "SYSTEM_ADMIN", "company_id": None}
    assert [x["key"] for x in client.get("/api/m02/cf/list").json()["logics"]] == ["견본 돌쌓기"]