diff --git a/M01_MasterData/M01_MasterData_Router.py b/M01_MasterData/M01_MasterData_Router.py index 31501f7f8..de50a83e3 100644 --- a/M01_MasterData/M01_MasterData_Router.py +++ b/M01_MasterData/M01_MasterData_Router.py @@ -14,6 +14,7 @@ from pydantic import BaseModel from M01_MasterData import M01_MasterData_Store as store from M01_MasterData import M01_MasterData_Store_Combo as combo_store +from M01_MasterData import M01_MasterData_Store_Flow as flow from M01_MasterData import M01_MasterData_Store_Make as make router = APIRouter(prefix="/api/m01", tags=["M01 MasterData"]) @@ -221,6 +222,10 @@ def get_materials( @router.post("/calc") def post_calc(body: CalcBody) -> dict: + """줄에 「흐름」 이 있으면 흐름 셈(입력 = 설계 때 정할 것) · 없으면 옛 셈(정함 4).""" + got = _call(flow.calc, body.key, body.inputs, body.row) + if got is not None: + return got with store.cm.mf.mp.choosing(body.고름): return _call(store.calc, body.key, body.inputs, body.row, body.file) @@ -234,7 +239,14 @@ def post_text(body: CalcBody) -> dict: @router.post("/save") def post_save(body: SaveBody) -> dict: - return {"files": _call(store.save, [f.model_dump() for f in body.files])} + """「흐름」 줄은 검사 · 서버 다시 셈 · 새 줄(키 없음)은 자체 GX — {files, added, warnings}.""" + return _call(flow.save, [f.model_dump() for f in body.files]) + + +@router.get("/logic/flow") +def get_logic_flow(key: str, file: str = "") -> dict: + """흐름 받기 — {flow, converted, notes} · 흐름이 있으면 그대로 · 없으면 변환(파일 안 바뀜).""" + return _call(flow.get_flow, file, key) @router.post("/logic/new") diff --git a/M01_MasterData/M01_MasterData_Store_Flow.py b/M01_MasterData/M01_MasterData_Store_Flow.py new file mode 100644 index 000000000..eb2d329f7 --- /dev/null +++ b/M01_MasterData/M01_MasterData_Store_Flow.py @@ -0,0 +1,269 @@ +"""M01 로직 「흐름」 칸(PLAN 36-4 · 정함 1 · 2 · 4 · 10 · 11 · 12). + +정함 = `docs/raw/M01_쉬운단가/흐름도_적용_정함.md`. + +· 흐름 = DataFlow Model `CfLogic` 꼴 그대로 · 로직 줄 칸 「흐름」 · 있으면 흐름이 정본 +· 쓰는 때 = `/save` 줄에 실어 온 것뿐(자동 변환 결과는 안 씀) — 꼴 검사(상자 · 선 · 돌고 도는 선) → + 서버 셈(`M02_DF_Compute.run`)으로 다시 셈 → 받는 칸에 값만 되돌려 넣음 · 빈 받는 자리는 경고 +· 새 줄(키 없음) + 흐름 = 자체 파일(`로직_자체.json`)에 더함 → 키 = 대장 `GX` 다음 · 구분 자체 +· 계산 = 흐름이 있으면 흐름 셈 · 없으면 옛 셈(`Store.calc`) · 입력 이름 = 「설계 때 정할 것」 +· 다른 로직 입력 = 마스터 키도 — 흐름 로직은 흐름 셈 · 옛 로직은 옛 셈(노무비 · 재료비 · 경비 · 계) +· 흐름 받기 = 줄의 흐름 그대로 · 없으면 sub7 변환(`M02_DF_Convert.convert`) · 아직 없으면 빈 흐름 +권한 = `/api/m01` 전체가 시스템 관리자(`main.py`) — 공용 로직 흐름도 관리자가 고침. +""" + +from __future__ import annotations + +import importlib +from decimal import Decimal, InvalidOperation +from typing import Any + +from M01_MasterData import M01_MasterData_Store as store +from M01_MasterData import M01_MasterData_Store_Make as make +from M02_MasterTemplete.DataFlow import M02_DF_Compute as dc + +FLOW = "흐름" +KINDS = ("input", "function", "output") +CONVERT = "M02_MasterTemplete.DataFlow.M02_DF_Convert" +NOT_READY = "변환 준비 중" + + +# ── 검사(Model `link` · `ruleWhy` 와 같은 뜻) ────────────────────────────── +def _outs(box: dict) -> set: + """내보내는 알약 — Model `portsOut`.""" + if box.get("kind") == "function": + return {((box.get("func") or {}).get("output") or {}).get("name")} + inp = box.get("input") or {} + if box.get("kind") != "input": + return set() + if inp.get("mode") == "logic": + return {p.get("name") for p in (inp.get("logic") or {}).get("outputs") or []} + out = {dc._port(e) for e in inp.get("elements") or []} if inp.get("mode") == "group" else set() + if inp.get("mode") == "group" and inp.get("table"): + out.add(dc.TABLE_PORT) + return out | {v.get("name") for v in inp.get("variables") or []} + + +def _ins(box: dict) -> set: + """받는 알약 — Model `portsIn`.""" + if box.get("kind") == "function": + return {p.get("port") for p in (box.get("func") or {}).get("inputs") or []} + if box.get("kind") == "output": + return {dc.OUTPUT_PORT} + inp = box.get("input") or {} + return {p.get("name") for p in (inp.get("logic") or {}).get("designs") or []} + + +def check(flow: Any) -> list[str]: + """흐름 꼴 검사 — 틀리면 422 · 돌려받는 것 = 경고(빈 받는 자리).""" + if not isinstance(flow, dict) or not isinstance(flow.get("boxes"), list): + raise store.StoreError(422, {"errors": ["흐름 꼴이 아님 — 상자(boxes) 목록이 없음"]}) + if not isinstance(flow.get("links"), list): + raise store.StoreError(422, {"errors": ["흐름 꼴이 아님 — 선(links) 목록이 없음"]}) + errors, boxes = [], {} + for b in flow["boxes"]: + if not isinstance(b, dict) or not b.get("id"): + errors.append("번호(id) 없는 상자가 있음") + continue + name = b.get("name") or b["id"] + if b["id"] in boxes: + errors.append(f"상자 번호가 겹침 「{b['id']}」") + boxes[b["id"]] = b + if b.get("kind") not in KINDS: + errors.append(f"「{name}」 모르는 상자 종류 「{b.get('kind')}」") + elif b["kind"] == "function" and not ((b.get("func") or {}).get("output") or {}).get( + "name" + ): + errors.append(f"「{name}」 함수 상자에 결과 알약이 없음") + taken = set() + for ln in flow["links"]: + src, to = (ln or {}).get("from") or {}, (ln or {}).get("to") or {} + a, b = boxes.get(src.get("box")), boxes.get(to.get("box")) + if ( + a is None + or b is None + or src.get("name") not in _outs(a) + or to.get("name") not in _ins(b) + ): + errors.append(f"선 「{(ln or {}).get('id')}」 이 없는 알약을 이음") + continue + if (to["box"], to["name"]) in taken: + errors.append(f"「{b.get('name')} · {to['name']}」 에 선이 둘 이상") + taken.add((to["box"], to["name"])) + for p in dc._sources(ln.get("rule")): + if p.get("name") not in _outs(boxes.get(p.get("box")) or {}): + errors.append(f"선 「{ln.get('id')}」 규칙의 조건 알약이 없음") + if not errors: + errors += [ + f"「{b.get('name')}」 상자가 서로를 기다림(돌고 도는 선)" for b in dc._waves(flow)[1] + ] + if errors: + raise store.StoreError(422, {"errors": errors}) + return [ + f"「{b.get('name')} · {port}」 받는 자리에 들어오는 선이 없음" + for b in boxes.values() + if b.get("kind") != "input" + for port in sorted(_ins(b) - {n for bx, n in taken if bx == b["id"]}, key=str) + ] + + +# ── 다른 로직 · 계산 ─────────────────────────────────────────────────── +def _old(key: str, designs: dict) -> dict: + """옛 로직 값 — 옛 셈 · 내보내는 알약 = 노무비 · 재료비 · 경비 · 계(돈 아닌 로직은 결과).""" + got = store.calc(key, {k: v for k, v in designs.items() if v is not None}) + if not got.get("ok"): + return {"outputs": {}, "errors": [got.get("reason") or "옛 셈 오류"]} + if "result" in got: + return {"outputs": {"결과": got["result"]}, "errors": []} + return {"outputs": dict(got["sums"]), "errors": []} + + +def load(key: str) -> dict: + """다른 로직 입력이 부르는 마스터 로직 — 흐름이면 흐름 · 옛 로직이면 옛 셈 함수(`calc`).""" + row = store.logic(key)["logic"] + if isinstance(row.get(FLOW), dict): + return {**row[FLOW], "key": key} + return {"key": key, "calc": lambda designs: _old(key, designs)} + + +def old_designs(key: str) -> list[dict]: + """옛 로직의 받는 알약 = 로직 입력 — Model `CfDesign` 꼴.""" + out = [] + for one in store.logic(key)["logic"].get("입력") or []: + choices = one.get("고르기") + variable = {"name": one.get("이름"), "type": "choice" if choices else "number", + "unit": one.get("단위") or "-", "example": None} # fmt: skip + if choices: + variable["choices"] = choices + out.append({"box": "", "port": one.get("이름"), "kind": "변수", + "unit": one.get("단위") or "-", "variable": variable}) # fmt: skip + return out + + +def _value(v: Any) -> Any: + """화면이 보낸 값 — 수 글은 수로(「3」 → 3) · 그 밖은 그대로.""" + if isinstance(v, str) and v.strip(): + try: + return Decimal(v.strip()) + except InvalidOperation: + return v + return store._dec(v) + + +def run(flow: dict, key: str, inputs: dict) -> dict: + """흐름 셈 — 입력 = 「설계 때 정할 것」 이름 → 값.""" + given = {k: _value(v) for k, v in (inputs or {}).items()} + flow = {**flow, "key": key} + return dc.run(flow, dc._design_values(flow, given), load, flow.get("userElements") or [], [key]) + + +def calc(key: str, inputs: dict, row: dict | None = None) -> dict | None: + """계산 갈래 — 줄에 흐름이 없으면 None(옛 셈 그대로) · 있으면 흐름 셈을 `/calc` 꼴로.""" + if row is not None: + flow, key = row.get(FLOW), str(row.get("키") or key) + else: + try: + flow = store.logic(key)["logic"].get(FLOW) + except store.StoreError: + return None + if not isinstance(flow, dict): + return None + got = run(flow, key, inputs) + names = {b["id"]: b for b in flow.get("boxes") or []} + outputs = [ + {"box": i, "name": names[i].get("name"), "cost": (names[i].get("output") or {}).get("cost"), + "unit": (names[i].get("output") or {}).get("unit") or names[i].get("unit"), "value": v} + for i, v in got["outputs"].items() if i in names + ] # fmt: skip + cost = got["cost"] + out = {"ok": not got["errors"], "flow": True, "lines": [], "middle": {}, + "sums": {k: cost[k] for k in (*dc.COSTS, "계")}, "values": cost["값들"], + "outputs": outputs, "errors": got["errors"], "breaks": got["breaks"]} # fmt: skip + if got["errors"]: + out["reason"] = " · ".join(got["errors"]) + return out + + +# ── 흐름 받기 ────────────────────────────────────────────────────────── +def _row(file: str, key: str) -> dict: + if not file: + return store.logic(key)["logic"] + data = store.read(file)[0] + hit = next((r for r in data.get("줄") or [] if str(r.get("키")) == key), None) + if hit is None: + raise store.StoreError(404, f"{file} · 없는 키 「{key}」") + return hit + + +def get_flow(file: str, key: str) -> dict: + """{flow, converted, notes} — 흐름이 있으면 그대로 · 없으면 변환(파일은 안 바뀜).""" + row = _row(file, key) + if isinstance(row.get(FLOW), dict): + return {"flow": {**row[FLOW], "key": key}, "converted": False, "notes": []} + try: + convert = importlib.import_module(CONVERT).convert + except ModuleNotFoundError as e: + if e.name != CONVERT: + raise + empty = {"key": key, "name": row.get("이름") or "", "resultUnit": row.get("결과단위") or "", + "boxes": [], "links": [], "userElements": []} # fmt: skip + return {"flow": empty, "converted": True, "notes": [NOT_READY]} + got = convert(row, {"key": key, "file": file, "master": store.loaded()[1]}) + return {"flow": {**got["flow"], "key": key}, "converted": True, "notes": got.get("notes") or []} + + +# ── 저장 ────────────────────────────────────────────────────────────── +def _redo(row: dict, key: str) -> list[str]: + """흐름 검사 → 서버 셈으로 다시 셈 → 받는 칸 값만 되돌려 넣음 · 돌려받는 것 = 경고.""" + warns = check(row[FLOW]) + got = run(row[FLOW], key, {}) + filled = got["filled"] + filled.pop("key", None) # 키는 줄의 「키」 하나 — 흐름 안에 두지 않음 + row[FLOW] = filled + return warns + got["errors"] + + +def _bare(row: dict) -> dict: + """새 흐름 로직 — 옛 식 칸은 빈 것으로(파일 틀 검사) · 출처 없으면 「자체」.""" + old = {"입력": [], "중간": [], "호표": [], "끝수": None} + return {**old, **row, "출처": row.get("출처") or "자체"} + + +def save(batch: list[dict]) -> dict: + """`/save` — 흐름 줄은 검사 · 다시 셈 · 새 줄은 자체 GX · 그 밖은 `Store.save` 그대로.""" + warns, fresh = [], False + for part in batch: + for ch in part.get("changes") or []: + row = ch.get("row") + if not isinstance(row, dict) or FLOW not in row: + continue + new = ch.get("op") == "add" or (ch.get("op") == "edit" and not ch.get("key")) + if new: + if part.get("file") != make.OWN: + raise store.StoreError(400, f"새 흐름 로직은 「{make.OWN}」 에만 더함") + row = make._clean(_bare(row), str(row.get("소유") or make.OWNERS[0])) + ch.update(op="add", key=None) + ch["row"] = row + name = row.get("이름") or ch.get("key") or "새 로직" + warns += [f"「{name}」 {w}" for w in _redo(row, str(ch.get("key") or ""))] + if part.get("file") == make.OWN and not make._own_path().is_file(): + make._make_file() + fresh = True + part["version"] = part.get("version") or store.read(make.OWN)[1] + before = {p["file"]: _keys(p["file"]) for p in batch} + try: + files = store.save(batch) + except store.StoreError: + if fresh: + make._own_path().unlink(missing_ok=True) + raise + added = [{"file": f, "key": k} for f, old in before.items() for k in _keys(f) if k not in old] + return {"files": files, "added": added, "warnings": warns} + + +def _keys(file: str) -> list[str]: + try: + data = store.read(file)[0] + except store.StoreError: + return [] + return [str(r.get("키")) for r in data.get(store.items_key(data)) or []] diff --git a/M02_MasterTemplete/DataFlow/M02_DF_Compute.py b/M02_MasterTemplete/DataFlow/M02_DF_Compute.py index 331a2a151..e3d8cf96b 100644 --- a/M02_MasterTemplete/DataFlow/M02_DF_Compute.py +++ b/M02_MasterTemplete/DataFlow/M02_DF_Compute.py @@ -23,8 +23,10 @@ from decimal import ROUND_DOWN, ROUND_HALF_UP, ROUND_UP, Decimal from typing import Any, Callable from common_util import common_util_spreadsheet as spreadsheet +from M01_MasterData import ( + M01_MasterData_Router as m01, +) # _num — M01 Router 가 이 파일을 부름(서로 부름) from M01_MasterData import M01_MasterData_Store as store -from M01_MasterData.M01_MasterData_Router import _num from M02_MasterTemplete.DataFlow import M02_DF_Values as dv _CELL = re.compile(r"^([A-Z]+)(\d+)$") @@ -188,6 +190,12 @@ def _logic_ports(lg: dict, box: dict, given: dict, ports: dict, errors: list, ct if link.get("to", {}).get("box") == box["id"]: src = link.get("from") or {} designs[link["to"].get("name")] = ports.get(src.get("box"), {}).get(src.get("name")) + if callable( + other.get("calc") + ): # 옛 마스터 로직(흐름 없음) — 옛 셈 · 노무비 · 재료비 · 경비 · 계 + result = other["calc"](designs) + errors += [f"「{box.get('name')}」 {x}" for x in result["errors"]] + return result["outputs"] stack = [*ctx["stack"], key] result = run(other, _design_values(other, designs), ctx["load"], ctx["local"], stack) errors += [f"「{box.get('name')}」 {x}" for x in result["errors"]] @@ -559,7 +567,9 @@ def run(lg: dict, values: dict, load: Loader, local: list[dict], stack: list | N outputs[b["id"]] = arrive(b).get(OUTPUT_PORT) if funcs: # 이 물결의 함수 시트를 한 번에(시트마다 통합문서 하나 — Eval 과 같음) books = [ - _num({"종류": "통합문서", "판": 1, "서식": [{}] * 64, "시트": [sheets[f["id"]]]}) + m01._num( + {"종류": "통합문서", "판": 1, "서식": [{}] * 64, "시트": [sheets[f["id"]]]} + ) for f in funcs ] try: diff --git a/M02_MasterTemplete/DataFlow/M02_DF_Router.py b/M02_MasterTemplete/DataFlow/M02_DF_Router.py index c7375090c..3892588d9 100644 --- a/M02_MasterTemplete/DataFlow/M02_DF_Router.py +++ b/M02_MasterTemplete/DataFlow/M02_DF_Router.py @@ -36,6 +36,7 @@ from fastapi import APIRouter, HTTPException from pydantic import BaseModel from M01_MasterData import M01_MasterData_Store as store +from M01_MasterData import M01_MasterData_Store_Flow as sflow from M01_MasterData.M01_MasterData_Router import _num from M02_MasterTemplete.DataFlow import M02_DF_Compute as dc from M02_MasterTemplete.DataFlow import M02_DF_Values as dv @@ -184,7 +185,10 @@ def user_edit(op: str, element: dict | None, uid: str | None) -> dict: # ── 저장 · 계산(`M02_DF_Compute`) ───────────────────────────────────── def _load(key: str) -> dict: - """다른 로직 입력이 부르는 로직 — 내 것(`_data`) · 견본(`_sample`).""" + """다른 로직 입력이 부르는 로직 — 내 것(`_data`) · 견본(`_sample`) · 마스터 키(GX · GF …)는 + M01 계산 길(흐름이면 흐름 · 옛 로직이면 옛 셈 · 정함 11).""" + if not _KEY.match(key or "") and _sample(key) is None: + return sflow.load(key) logic, _ = _read(key) return {**logic, "key": key} @@ -216,6 +220,9 @@ def save(logic: dict) -> dict: def logic_value(key: str, designs: dict) -> dict: """다른 로직 값 — 저장 로직 · 견본을 받은 설계 값(없으면 예시값)으로 셈 · 출력 이름 → 값.""" other = _load(key) + if callable(other.get("calc")): # 옛 마스터 로직 — 노무비 · 재료비 · 경비 · 계 + got = other["calc"](designs) + return {"key": key, "outputs": got["outputs"], "cost": {}, "errors": got["errors"]} got = dc.run(other, dc._design_values(other, designs), _load, _local(other), [key]) names = {b["id"]: b.get("name") for b in other.get("boxes") or []} outputs = {names[i]: v for i, v in got["outputs"].items() if i in names} @@ -224,7 +231,10 @@ def logic_value(key: str, designs: dict) -> dict: def design_inputs(key: str) -> dict: """설계 때 정할 것 — 변수 · 재료 자리 · 다른 로직 받는 알약(선이 없을 때) · CfDesign 꼴.""" - return {"key": key, "designs": dc.design_inputs(_load(key))} + other = _load(key) + if callable(other.get("calc")): + return {"key": key, "designs": sflow.old_designs(key)} + return {"key": key, "designs": dc.design_inputs(other)} # ── 길 ──────────────────────────────────────────────────────────────── diff --git a/resources/tester/test_m01_logic_flow_save.py b/resources/tester/test_m01_logic_flow_save.py new file mode 100644 index 000000000..312664317 --- /dev/null +++ b/resources/tester/test_m01_logic_flow_save.py @@ -0,0 +1,240 @@ +# ruff: noqa: E501 +"""흐름도 적용(PLAN 36-4 서버 · 정함 1 · 2 · 4 · 10 · 11 · 12) — 로직 줄 「흐름」 칸. + +못박는 것(마스터 사본 폴더에서만 · 정본은 안 바뀜) + - `/save` 줄에 실은 흐름 = 검사 → 서버 다시 셈(받는 칸에 값) → 파일에 씀 · 다시 읽으면 그대로 + - 흐름 꼴이 틀리면(없는 알약 · 돌고 도는 선) 422 · 빈 받는 자리는 경고 + - `/calc` 갈래 — 흐름 있으면 흐름 셈(입력 = 설계 때 정할 것 · 노무비 · 재료비 · 경비 · 계) · 없으면 옛 셈 + - 새 줄(키 없음) + 흐름 → 자체 파일 GX 다음 키 · 구분 자체 · 본떠 만들기는 흐름도 함께 + - `GET /logic/flow` — 흐름 있으면 그대로 · 없으면 변환(준비 전이면 빈 흐름 + 「변환 준비 중」) + - DataFlow 다른 로직 값 · 설계 입력이 마스터 키도 받음(흐름 · 옛 로직) · 서로 부르면 막음 + - 시스템 관리자가 아니면 403 +""" + +from __future__ import annotations + +import copy +import hashlib +import importlib.util +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 M01_MasterData import M01_MasterData_Store_Flow as sflow +from M02_MasterTemplete.DataFlow import M02_DF_Router as df + +REAL = store.FOLDER +KEY = "GF000219" # 13-4-1 메쌓기 — 흐름을 얹어 볼 정본 로직 +OLD = "GF000160" # 12-4 합판거푸집 — 흐름 없는 옛 로직 +OWN = "로직_자체.json" + + +def _flow(name: str = "넓이 품") -> dict: + """길이 × 폭 × 1000 → 노무비 — 입력 상자(변수 둘) · 함수 상자 · 출력 상자.""" + sheet = {"id": "s-fn", "이름": "넓이", "칸": {"A2": {"값": "길이"}, "A3": {"값": "폭"}, + "A4": {"값": "금액"}, "B4": {"식": "B2*B3*1000"}}} # fmt: skip + return { + "key": None, "name": name, "resultUnit": "원/㎡", "userElements": [], + "boxes": [ + {"id": "in", "kind": "input", "name": "치수", "color": "", "unit": "-", "x": 0, "y": 0, + "input": {"mode": "variable", "elements": [], "variables": [ + {"name": "길이", "type": "number", "unit": "m", "example": 2}, + {"name": "폭", "type": "number", "unit": "m", "example": 3}]}}, + {"id": "fn", "kind": "function", "name": "넓이 금액", "color": "", "unit": "원", "x": 0, "y": 0, + "func": {"sheet": sheet, "inputs": [{"port": "길이", "cell": "B2"}, {"port": "폭", "cell": "B3"}], + "variables": [], "output": {"cell": "B4", "name": "금액", "unit": "원"}}}, + {"id": "out", "kind": "output", "name": "노무비", "color": "", "unit": "원/㎡", "x": 0, "y": 0, + "output": {"cost": "노무비", "unit": "원/㎡"}}, + ], + "links": [ + {"id": "l1", "from": {"box": "in", "name": "길이"}, "to": {"box": "fn", "name": "길이"}}, + {"id": "l2", "from": {"box": "in", "name": "폭"}, "to": {"box": "fn", "name": "폭"}}, + {"id": "l3", "from": {"box": "fn", "name": "금액"}, "to": {"box": "out", "name": "값"}}, + ], + } # fmt: skip + + +def _hashes(folder: Path) -> dict[str, str]: + return {p.name: hashlib.sha1(p.read_bytes()).hexdigest() for p in folder.glob("*.json")} + + +@pytest.fixture +def folder(tmp_path: Path, monkeypatch) -> Path: + master = tmp_path / "master" + master.mkdir() + for path in REAL.glob("*.json"): + shutil.copy(path, master / path.name) + monkeypatch.setattr(store, "FOLDER", master) + monkeypatch.setattr(df, "DATA", tmp_path / "_data") + return master + + +@pytest.fixture +def client(folder: Path) -> TestClient: + app = FastAPI() + app.include_router(m01_router.router) + app.include_router(df.router) + return TestClient(app) + + +def _get(client: TestClient, url: str, want: int = 200, **params) -> dict: + res = client.get(url, params=params) + assert res.status_code == want, res.text + return res.json() + + +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 _save_flow(client: TestClient, key: str, flow: dict, want: int = 200) -> dict: + got = _get(client, "/api/m01/logic", key=key) + row = {**got["logic"], "흐름": flow} + part = {"file": got["file"], "version": got["version"], + "changes": [{"op": "edit", "key": key, "row": row}]} # fmt: skip + return _post(client, "/api/m01/save", {"files": [part]}, want) + + +def test_흐름_저장_다시_셈_다시_읽기(client: TestClient, folder: Path) -> None: + before = _hashes(REAL) + done = _save_flow(client, KEY, _flow()) + assert done["warnings"] == [] and done["added"] == [] + saved = _get(client, "/api/m01/logic", key=KEY)["logic"]["흐름"] + cells = saved["boxes"][1]["func"]["sheet"]["칸"] + assert ( + Decimal(str(cells["B2"]["값"])) == 2 and Decimal(str(cells["B3"]["값"])) == 3 + ) # 서버가 넣은 값 + assert "key" not in saved + got = _get(client, "/api/m01/logic/flow", key=KEY) + assert got["converted"] is False and got["notes"] == [] and got["flow"]["key"] == KEY + assert [b["id"] for b in got["flow"]["boxes"]] == ["in", "fn", "out"] + assert _hashes(REAL) == before # 사본 밖(정본)은 안 바뀜 + + +def test_흐름_꼴_검사_422_와_빈_자리_경고(client: TestClient) -> None: + bad = _flow() + bad["links"][0]["from"]["name"] = "없는알약" + assert "없는 알약" in str(_save_flow(client, KEY, bad, 422)) + loop = _flow() + loop["boxes"].append({**copy.deepcopy(loop["boxes"][1]), "id": "fn2", "name": "되돌이"}) + loop["links"] += [ + {"id": "l4", "from": {"box": "fn", "name": "금액"}, "to": {"box": "fn2", "name": "길이"}}, + {"id": "l5", "from": {"box": "fn2", "name": "금액"}, "to": {"box": "fn", "name": "폭"}}, + ] + loop["links"] = [x for x in loop["links"] if x["id"] != "l2"] + assert "돌고 도는" in str(_save_flow(client, KEY, loop, 422)) + empty = _flow() + empty["links"] = [x for x in empty["links"] if x["id"] != "l2"] + warns = _save_flow(client, KEY, empty)["warnings"] + assert any("넓이 금액 · 폭" in w for w in warns), warns + + +def test_계산_갈래_흐름이면_흐름_셈_아니면_옛_셈(client: TestClient) -> None: + _save_flow(client, KEY, _flow()) + got = _post(client, "/api/m01/calc", {"key": KEY, "inputs": {"길이": 4, "폭": "5"}}) + assert got["flow"] is True and got["ok"] is True + assert {k: Decimal(str(v)) for k, v in got["sums"].items()} == { + "노무비": 20000, "재료비": 0, "경비": 0, "계": 20000} # fmt: skip + assert ( + got["outputs"][0]["name"] == "노무비" and Decimal(str(got["outputs"][0]["value"])) == 20000 + ) + plain = _post(client, "/api/m01/calc", {"key": KEY, "inputs": {}}) # 예시값 2 × 3 + assert Decimal(str(plain["sums"]["계"])) == 6000 + row = {**_get(client, "/api/m01/logic", key=KEY)["logic"], "흐름": _flow()} # 저장 전 줄 + row["흐름"]["boxes"][1]["func"]["sheet"]["칸"]["B4"] = {"식": "B2*B3*10"} + draft = _post(client, "/api/m01/calc", {"key": KEY, "inputs": {}, "row": row}) + assert Decimal(str(draft["sums"]["계"])) == 60 + old = _post(client, "/api/m01/calc", {"key": OLD, "inputs": {}}) + assert "flow" not in old # 흐름 없는 로직은 옛 셈 그대로 + + +def test_신규_흐름_로직은_자체_GX_복사는_흐름도(client: TestClient, folder: Path) -> None: + version = store.read(OWN)[1] if (folder / OWN).is_file() else "" + row = {"키": "", "이름": "시험 흐름 로직", "결과단위": "원/㎡", "출처": "", "구분": "산림품셈", + "흐름": _flow("시험 흐름 로직")} # fmt: skip + part = {"file": OWN, "version": version, "changes": [{"op": "add", "row": row}]} + done = _post(client, "/api/m01/save", {"files": [part]}) + assert len(done["added"]) == 1 and done["added"][0]["file"] == OWN + key = done["added"][0]["key"] + assert key.startswith("GX") + saved = _get(client, "/api/m01/logic", key=key)["logic"] + assert saved["구분"] == "자체" and saved["흐름"]["name"] == "시험 흐름 로직" + assert ( + Decimal(str(_post(client, "/api/m01/calc", {"key": key, "inputs": {}})["sums"]["계"])) + == 6000 + ) + made = _post(client, "/api/m01/logic/copy", {"key": key, "이름": "복사한 흐름"}) + assert made["key"].startswith("GX") and made["key"] != key + copied = _get(client, "/api/m01/logic/flow", key=made["key"]) + assert copied["converted"] is False and len(copied["flow"]["boxes"]) == 3 + bad = {"file": OWN, "version": store.read(OWN)[1], "changes": [{"op": "add", "row": row}]} + bad["changes"][0]["row"] = {**row, "흐름": {"boxes": "아님"}} + _post(client, "/api/m01/save", {"files": [bad]}, 422) + + +def test_흐름_없는_로직_받기는_변환(client: TestClient) -> None: + got = _get(client, "/api/m01/logic/flow", key=OLD) + assert got["converted"] is True and got["flow"]["key"] == OLD + if importlib.util.find_spec(sflow.CONVERT) is None: # sub7 변환이 오기 전 + assert got["notes"] == [sflow.NOT_READY] and got["flow"]["boxes"] == [] + _get(client, "/api/m01/logic/flow", 404, key="GX999999") + + +def test_다른_로직_값_마스터_키_흐름_옛_순환(client: TestClient) -> None: + _save_flow(client, KEY, _flow()) + got = _post(client, "/api/m02/df/logic-value", {"key": KEY, "designs": {"길이": 10}}) + assert Decimal(str(got["outputs"]["노무비"])) == 30000 and got["errors"] == [] + designs = _get(client, "/api/m02/df/design-inputs", key=KEY)["designs"] + assert [d["port"] for d in designs] == ["길이", "폭"] + given = {k: str(v) for k, v in store.cm.시험입력(store.logic(OLD)["logic"]).items()} + old = _post(client, "/api/m02/df/logic-value", {"key": OLD, "designs": given}) + want = _post(client, "/api/m01/calc", {"key": OLD, "inputs": given}) + assert old["outputs"] == want["sums"] and set(old["outputs"]) == { + "노무비", + "재료비", + "경비", + "계", + } + names = [d["port"] for d in _get(client, "/api/m02/df/design-inputs", key=OLD)["designs"]] + assert names == [x["이름"] for x in store.logic(OLD)["logic"].get("입력") or []] + # 흐름 안 다른 로직 입력이 옛 로직(OLD) 계 를 받아 노무비로 · 자기 자신을 부르면 막힘 + flow = _flow() + pills = [{"name": k, "unit": "-"} for k in given] + ref = {"key": OLD, "outputs": [{"name": "계", "unit": "원"}], "designs": pills} + flow["boxes"].append({"id": "lg", "kind": "input", "name": "거푸집", "color": "", "unit": "-", + "x": 0, "y": 0, "input": {"mode": "logic", "elements": [], "variables": [], + "logic": ref}}) # fmt: skip + flow["boxes"].append({**copy.deepcopy(flow["boxes"][2]), "id": "out2", "name": "경비"}) + flow["boxes"][-1]["output"]["cost"] = "경비" + flow["links"].append( + {"id": "l9", "from": {"box": "lg", "name": "계"}, "to": {"box": "out2", "name": "값"}} + ) + _save_flow(client, KEY, flow) + mixed = _post(client, "/api/m01/calc", {"key": KEY, "inputs": given}) + assert Decimal(str(mixed["sums"]["경비"])) == Decimal(str(want["sums"]["계"])) + ref["key"] = KEY + self_call = _save_flow(client, KEY, flow)["warnings"] + assert any("서로를 부름" in w for w in self_call), self_call + + +def test_시스템_관리자가_아니면_403(folder: Path) -> None: + app = FastAPI() + app.include_router( + m01_router.router, dependencies=[Depends(verify_session), Depends(require_system_admin)] + ) + app.dependency_overrides[verify_session] = lambda: {"role": "USER", "company_id": 1} + client = TestClient(app) + before = _hashes(folder) + assert client.get("/api/m01/logic/flow", params={"key": KEY}).status_code == 403 + assert client.post("/api/m01/save", json={"files": []}).status_code == 403 + assert client.post("/api/m01/calc", json={"key": KEY, "inputs": {}}).status_code == 403 + assert _hashes(folder) == before