Files
Aislo/resources/tester/test_m02_df_router.py
T
eomsangdonandClaude Opus 5.5 868dc8e37f feat(m02-df): 흐름도 서버 — 재료 자리 셋 · 자재 찾기 · 다른 로직 값 · 설계 때 정할 것 · 표준 표 · Eval 과 같은 차례로 다시 셈
- M02_DF_Compute.py — 화면 Eval(evalLogic)과 같은 차례(Model depends) · 선 규칙은 Rule.ts applyRule 과 같은 뜻(값 기준 · 막히면 받는 상자 멈춤) · 끝수 = ROUND 계열 · 함수 시트는 같은 엔진 Node(서로 안 기대는 것은 한 번에) · 다른 로직은 받은 설계 값으로 셈 · 서로 부르면 막음 · 받는 칸에 값만 되돌림
- M02_DF_Values.py(Router 에서 나눔) — 재료 고정 · 선택(고른 품목 · 기본) · 조건 찾기(범위 + 규격/이름 포함/같음 · 0 · 여럿 후보) · 자재 찾기(띄어쓰기 AND · 이름 · 규격 · 키 · 원문번호 · 쪽) · 표준 표
- Router — GET /materials · /design-inputs · POST /logic-value · GET/PUT/DELETE /standards · main.py 등록
- Api — cfMaterials · cfLogicValue · cfDesignInputs · cfStandards(Save · Reset) · slot 물음
- 시험 — 재료 셋 · AND · 다른 로직 · 순환 · 선 규칙 넷 · 끝수 · 설계 때 정할 것 · 표준 표 · 권한

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01KmELpdXFDRZHhfbGuRhSmL
2026-10-01 19:58:10 +09:00

592 lines
31 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
# 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 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.DataFlow import M02_DF_Router as cf
from M02_MasterTemplete.DataFlow import M02_DF_Values as dv
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/df/save", {"logic": _logic(labor)})
assert saved["key"] == "DF-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/df/list").json()["logics"]
assert [(x["key"], x["sample"]) for x in listed] == [("DF-0001", False), ("견본 돌쌓기", True)]
sample = client.get("/api/m02/df/get", params={"key": "견본 돌쌓기"}).json()
assert sample["sample"] is True
assert _post(client, "/api/m02/df/save", {"logic": sample["logic"]})["key"] == "DF-0002"
_post(client, "/api/m02/df/delete", {"key": "견본 돌쌓기"}, want=403)
renamed = {**saved["logic"], "name": "고친 이름"}
assert _post(client, "/api/m02/df/save", {"logic": renamed})["key"] == "DF-0001"
got = client.get("/api/m02/df/get", params={"key": "DF-0001"}).json()["logic"]
assert (
got["name"] == "고친 이름" and got["boxes"][3]["func"]["sheet"]["칸"]["A1"]["값"] == price
)
_post(client, "/api/m02/df/delete", {"key": "DF-0002"})
assert client.get("/api/m02/df/get", params={"key": "DF-0002"}).status_code == 404
_post(client, "/api/m02/df/save", {"logic": _logic(labor, "DF-0009")}, want=404)
_post(client, "/api/m02/df/save", {"logic": {**_logic(labor), "name": " "}}, want=400)
_post(client, "/api/m02/df/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")) == ["DF-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/df/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/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["값"]["가격"]
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/df/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/df/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/df/list").status_code == 403
assert client.post("/api/m02/df/values", json={"items": []}).status_code == 403
assert client.get("/api/m02/df/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/df/list").json()["logics"]] == ["견본 돌쌓기"]
@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 = _post(client, "/api/m02/df/save", {"logic": body})
assert _d(saved["values"]["cost"]["경비"]) == 2 * _d(want)
_post(client, "/api/m02/df/delete", {"key": saved["key"]})
# ── 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 = _post(client, "/api/m02/df/save", {"logic": _area()})["key"]
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 = _post(client, "/api/m02/df/save", {"logic": 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 = {**b, "key": saved["key"]}
loop["boxes"][0]["input"]["logic"] = {**ref, "key": saved["key"]}
got = _post(client, "/api/m02/df/save", {"logic": 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 = _post(client, "/api/m02/df/save", {"logic": 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"
} # 받는 칸에 값만
designs = client.get("/api/m02/df/design-inputs", params={"key": got["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(
_post(client, "/api/m02/df/save", {"logic": lg})["values"]["ports"]["f"]["합"]
) == int(Decimal("0.5") + 33000 + 3)
lg["boxes"][0]["input"]["variables"][2]["example"] = "D999" # 맞는 재료 없음
lg["boxes"][0]["input"]["variables"][0]["example"] = "없는돌" # 표에 없는 조건 → 함수 멈춤
bad = _post(client, "/api/m02/df/save", {"logic": lg})["values"]
assert any("맞는 재료 없음" in e for e in bad["errors"]) and any(
"표에 없는 조건" in e for e in bad["errors"]
)
assert bad["ports"]["f"]["합"] is None and _d(bad["cost"]["재료비"]) == 0
def test_표준_표_읽기_쓰기_처음값(client: TestClient, folder: Path) -> None:
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 and res.json()["saved"] is True
assert client.get("/api/m02/df/standards").json()["money"]["줄"][4]["지위"] == 1
assert client.put("/api/m02/df/standards", json={"unit": {}, "money": {}}).status_code == 400
back = client.delete("/api/m02/df/standards").json()
assert (
back["saved"] is False and back["money"]["줄"][4]["지위"] == first["money"]["줄"][4]["지위"]
)
assert (cf.STANDARD / "금액단위표준.json").is_file() # 처음 값 파일은 그대로
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