Aislo/resources/tester/test_m01_logic_flow_engine.py

# ruff: noqa: E501
"""흐름도 적용(PLAN 36-4 덧) — 옛 엔진 한 곳 갈래 · 인력 · 기계 · 유가전력 조건 찾기.

못박는 것(마스터 사본 폴더)
  - `mf.run` 을 바로 부르는 곳(조합 · 프로젝트 단가 · 읽는 식 · 옛 식의 로직(…))도 흐름 로직이면 흐름 셈
  - `/logic/auto` — 흐름 로직은 「설계 때 정할 것」 예시값
  - 조건 찾기 요소가 인력 · 기계 · 유가전력에서도 이름 · 키(원문번호) 같음으로 찾음
"""

from __future__ import annotations

import shutil
from decimal import Decimal
from pathlib import Path

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

from M01_MasterData import M01_MasterData_Router as m01_router
from M01_MasterData import M01_MasterData_Store as store
from M02_MasterTemplete.DataFlow import M02_DF_Compute as dc
from M02_MasterTemplete.DataFlow import M02_DF_Values as dv
from resources.tester.test_m01_logic_flow_save import KEY, _flow, _get, _save_flow

REAL = store.FOLDER


@pytest.fixture
def client(tmp_path: Path, monkeypatch) -> TestClient:
    for path in REAL.glob("*.json"):
        shutil.copy(path, tmp_path / path.name)
    monkeypatch.setattr(store, "FOLDER", tmp_path)
    app = FastAPI()
    app.include_router(m01_router.router)
    return TestClient(app)


def test_옛_엔진도_흐름_로직은_흐름_셈(client: TestClient) -> None:
    _save_flow(client, KEY, _flow())
    whole = store.cm.mf.Master(store.cm.load(folder=store.FOLDER))
    got = store.cm.mf.run(whole, KEY, {"길이": Decimal(4), "폭": Decimal(5)})
    assert got["줄"] == [] and got["노무비"] == 20000 and got["계"] == 20000
    line = {
        "종류": "로직",
        "요소": f"로직({KEY}, 길이=4, 폭=5)",
        "이름": "흐름 로직",
        "단위": "㎡",
        "수량": "2",
    }
    old = store.calc("", {}, dv._row(line, ()), None)  # 옛 식이 흐름 로직을 부름
    assert old["ok"] and old["sums"]["노무비"] == 40000, old
    auto = _get(client, "/api/m01/logic/auto", key=KEY)
    assert auto == {"값": {"길이": 2, "폭": 3}}


def _labor_flow(field: str | None, name: str) -> dict:
    flow = _flow()
    el = {"ref": "", "name": "노임", "spec": "", "unit": "원/인", "mode": "rule",
          "rule": {"field": field, "op": "같음", "from": {"box": "who", "name": "직종"}}}  # fmt: skip
    flow["boxes"] += [
        {"id": "who", "kind": "input", "name": "직종", "color": "", "unit": "-", "x": 0, "y": 0,
         "input": {"mode": "variable", "elements": [], "variables": [
             {"name": "직종", "type": "choice", "unit": "-", "example": name, "choices": [name]}]}},
        {"id": "lab", "kind": "input", "name": "노임", "color": "", "unit": "-", "x": 0, "y": 0,
         "input": {"mode": "group", "group": "인력", "elements": [el], "variables": []}},
    ]  # fmt: skip
    flow["boxes"][2]["output"]["cost"] = "값"
    flow["links"][2] = {
        "id": "l3",
        "from": {"box": "lab", "name": "노임"},
        "to": {"box": "out", "name": "값"},
    }
    return flow


def test_인력_기계_유가전력_조건_찾기(client: TestClient) -> None:
    rows = dv._group_rows("인력")
    one = next(r for r in rows if sum(x.get("이름") == r.get("이름") for x in rows) == 1)
    want = dv.one_value({"group": "인력", "ref": one["키"]})
    for field, value in (("이름", one["이름"]), ("키", one["원문번호"]), (None, one["키"])):
        got = dc.run(_labor_flow(field, value), {}, lambda k: {}, [], ["t"])
        assert got["errors"] == [] and got["cost"]["값들"]["노무비"] == want["value"], (
            field,
            got["errors"],
        )
    miss = dc.run(_labor_flow("이름", "없는직종zz"), {}, lambda k: {}, [], ["t"])
    assert any("맞는 인력 없음" in e for e in miss["errors"])
    machine = dv._group_rows("기계")[0]
    got = dv.group_find("기계", {"field": "키"}, machine["원문번호"])
    assert got["ref"] == machine["키"]
    oils = dv._group_rows("유가전력")
    if oils:
        got = dv.group_find("유가전력", {"field": "키"}, oils[0]["키"])
        assert got["ref"] == oils[0]["키"]


def test_표_값_수_식_글은_수로_재료_키_찾기(client: TestClient) -> None:
    assert dv._arith("(0.05)") == Decimal("0.05") and dv._arith("1/1.30") == Decimal(1) / Decimal(
        "1.30"
    )
    assert (
        dv._arith("토사") == "토사" and dv._arith("1/0") == "1/0"
    )  # 수 식 아니거나 못 세면 그대로
    files = store.loaded()[0]
    hit = next(
        ((t["키"], n) for d in files.values() if d.get("그룹") in store.cm.mf.TABLE_GROUPS
         for t in d.get("표") or [] for r in t.get("줄") or [] for n in t.get("값칸") or {}
         if isinstance(r.get(n), str) and store.cm.mf._ARITH.fullmatch(r[n])),
        None,
    )  # fmt: skip
    if hit:  # 마스터에 식 글 값이 있으면 표 알약에서 수로 나옴
        got = dv._table(hit[0])
        col = got["rows"][0].index(hit[1])
        assert not any(
            isinstance(r[col], str) and store.cm.mf._ARITH.fullmatch(r[col])
            for r in got["rows"][1:]
        )
    item = next(r for r in dv._group_rows("재료") if not str(r.get("키")).startswith("MO"))
    got = dv.slot_price({"mode": "rule", "field": "키", "op": "같음", "scope": {}}, item["키"])
    assert got.get("ref") == item["키"], got