"""단가산출 다이어그램 시험 길(`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"]] == ["견본 돌쌓기"] @pytest.mark.skipif(NO_NODE, reason="node 가 없음") def test_기계_키_칸_꼴과_시간당_경비(client: TestClient) -> None: """입력 판(sub3) 기계 ref 「EQ키#칸」 · 물음 {ref, field} · 「시간당 경비」 = 기계경비 로직.""" logic = min( (r for r in cf._machine_family() if "지역" in [i["이름"] for i in r["입력"]]), key=cf._machine_rank, ) code = str(cf._codes(logic)[0]) eq = _first("기계.json", lambda r: str(r.get("원문번호")) == code) family = [r for r in cf._machine_family() if code in [str(c) for c in cf._codes(r)]] chosen = min(family, key=cf._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/cf/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/cf/save", {"logic": body}) assert _d(saved["values"]["cost"]["경비"]) == 2 * _d(want) _post(client, "/api/m02/cf/delete", {"key": saved["key"]})