# 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 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 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") 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"][0]["example"] = ( "견치돌" # 「높이 4」 줄 = 아래 = 위 → 같음(정함) ) lg["boxes"][0]["input"]["variables"][1]["example"] = 4 assert _d( _post(client, "/api/m02/df/save", {"logic": 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 = _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 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))