feat(m02-df): 단가산출 흐름도 0걸음 — CostFlow 를 DataFlow 로 옮겨 이름 바꿈

- 파일 M02_CF_* → M02_DF_* · css m02-cf/m01-cf → m02-df · 서버 길 /api/m02/cf → /api/m02/df · 키 CF- → DF- · 사용자 요소 UE- 그대로
- _sample · .gitignore(_data git 밖) 옮김 · DataFlow 는 CostFlow 를 import 안 함
- 시험 test_m02_df_model · router · sample · inputpanel_build(옛 cf 시험은 그대로)
- 서버 등록은 sub1 몫 — M02_MasterTemplete.DataFlow.M02_DF_Router 의 router

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01VBToTCGk5dqDKeXCEP3vhk
This commit is contained in:
2026-10-01 19:37:51 +09:00
co-authored by Claude Opus 5.5
parent 12a60baeeb
commit fc7affb019
28 changed files with 8538 additions and 0 deletions
@@ -0,0 +1,254 @@
"""단가산출 다이어그램 입력 상자 판 — 고르는 규칙(`M02_DF_InputPanel_Build.ts`).
화면 TS 를 그 자리에서 트랜스파일해 node 로 돌림 — 순수 함수만 부름. 견본 줄 · 상자는 이 시험 안에 둠.
"""
import json
import shutil
import subprocess
from pathlib import Path
import pytest
ROOT = Path(__file__).resolve().parents[2]
SOURCE = ROOT / "M02_MasterTemplete" / "DataFlow" / "M02_DF_InputPanel_Build.ts"
RUNNER = r"""
const fs = require("fs");
const path = require("path");
const [root, file] = process.argv.slice(1);
const ts = require(path.join(root, "config", "node_modules", "typescript"));
const js = ts.transpileModule(fs.readFileSync(file, "utf8"), {
compilerOptions: { module: ts.ModuleKind.CommonJS, target: ts.ScriptTarget.ES2020 },
}).outputText;
const box = { exports: {} };
new Function("exports", "module", "require", js)(box.exports, box, () => ({}));
const m = box.exports;
const jobs = JSON.parse(fs.readFileSync(0, "utf8"));
const out = jobs.map((j) => {
if (j.op === "fn") return m[j.name](...j.args);
let lg = j.lg;
for (const [fn, ...args] of j.steps || []) lg = m[fn](lg, ...args);
return lg;
});
process.stdout.write(JSON.stringify(out));
"""
def run(jobs: list[dict]) -> list:
if shutil.which("node") is None:
pytest.skip("node 가 없음")
done = subprocess.run(
["node", "-e", RUNNER, str(ROOT), str(SOURCE)],
input=json.dumps(jobs, ensure_ascii=False),
capture_output=True,
text=True,
encoding="utf-8",
check=True,
)
return json.loads(done.stdout)
def logic() -> dict:
plain = {"color": "#fff", "unit": "-", "x": 0, "y": 0}
return {
"key": None,
"name": "견본",
"resultUnit": "원/㎡",
"userElements": [],
"boxes": [
{
"id": "b1",
"kind": "input",
"name": "입력",
**plain,
"input": {"mode": "variable", "elements": [], "variables": []},
},
{"id": "b2", "kind": "function", "name": "함수", **plain},
],
"links": [],
}
def call(name: str, *args):
return run([{"op": "fn", "name": name, "args": list(args)}])[0]
def edit(steps: list, lg: dict | None = None) -> dict:
return run([{"op": "lg", "lg": lg or logic(), "steps": steps}])[0]
LABOR = {"키": "LB000033", "이름": "석공", "규격": "", "단위": "원/일", "값": 268908}
LABOR2 = {"키": "LB000002", "이름": "보통인부", "규격": "", "단위": "원/일", "값": 172068}
MAT = {
"키": "MT1",
"이름": "각형흄관",
"규격": "D=300mm",
"단위": "M",
"물가자료": 30000,
"물가정보": 29600,
}
MACHINE = {
"키": "EQ000001",
"이름": "불도저",
"규격": "7ton",
"단위": {"가격": "천원", "손료계수": "10^-7/시간"},
"값": {"가격": 76547, "손료계수": 1811},
}
RATE_ONE = {"키": "RT1", "이름": "산재보험료", "값": {"rate_percent": 3.56}}
RATE_BRACKET = {"키": "RT2", "이름": "고용보험료", "값": {"brackets": []}}
def test_요소는_이름과_규격으로_알약_이름이_됨():
labor = call("elementOf", "인력", LABOR)
assert labor == {"ref": "LB000033", "name": "석공", "spec": "", "unit": "원/일"}
got = call("elementOf", "재료", MAT)
assert got["name"] == "각형흄관 D=300mm" and got["unit"] == "M"
def test_기계는_값_칸을_골라_요소가_됨():
assert call("machineFields", MACHINE) == ["가격", "손료계수"]
got = call("elementOf", "기계", MACHINE, "손료계수")
assert got["ref"] == "EQ000001#손료계수" and got["unit"] == "10^-7/시간"
assert got["name"].endswith("· 손료계수")
assert call("queryOf", "기계", "EQ000001#손료계수") == {
"group": "기계",
"ref": "EQ000001",
"field": "손료계수",
}
def test_값_읽기_재료는_낮은_값_요율은_비율():
assert call("valueOf", "재료", MAT) == 29600
assert call("valueOf", "요율", RATE_ONE) == 3.56
assert call("valueOf", "요율", RATE_BRACKET) is None
assert call("elementOf", "요율", RATE_ONE)["unit"] == "%"
assert call("elementOf", "요율", RATE_BRACKET)["unit"] == "-"
assert call("valueOf", "기계", MACHINE, "가격") == 76547
def test_사용자_요소는_그_그룹의_ref_로_물음():
assert call("queryOf", "인력", "UE-0001") == {"group": "인력", "ref": "UE-0001"}
assert call("queryOf", "재료", "MT1") == {"group": "재료", "ref": "MT1"}
row = {"키": "UE-0001", "이름": "특수공", "규격": "", "단위": "원/일", "값": 300000}
assert call("elementOf", "인력", row)["unit"] == "원/일"
assert call("elementOf", "요율", {**row, "단위": "%"})["unit"] == "%"
def test_상자_단위는_같으면_그_단위_여럿이면_하이픈():
one = edit([["setKind", "b1", "인력"], ["addElement", "b1", call("elementOf", "인력", LABOR)]])
assert one["boxes"][0]["unit"] == "원/일"
two = edit([["addElement", "b1", call("elementOf", "인력", LABOR2)]], one)
assert two["boxes"][0]["unit"] == "원/일" and len(two["boxes"][0]["input"]["elements"]) == 2
oil = {"ref": "X", "name": "경유", "spec": "", "unit": "원/L"}
assert edit([["addElement", "b1", oil]], two)["boxes"][0]["unit"] == "-"
def test_같은_요소_이름은_한_번만():
el = call("elementOf", "인력", LABOR)
lg = edit(
[
["setKind", "b1", "인력"],
["addElement", "b1", el],
["addElement", "b1", el],
["addElement", "b1", {**el, "ref": "다른"}],
]
)
assert len(lg["boxes"][0]["input"]["elements"]) == 1
def test_요소를_빼면_그_알약에서_나간_선도_지움():
el1, el2 = call("elementOf", "인력", LABOR), call("elementOf", "인력", LABOR2)
base = edit([["setKind", "b1", "인력"], ["addElement", "b1", el1], ["addElement", "b1", el2]])
base["links"] = [
{"id": "l1", "from": {"box": "b1", "name": "석공"}, "to": {"box": "b2", "name": "a"}},
{"id": "l2", "from": {"box": "b1", "name": "보통인부"}, "to": {"box": "b2", "name": "b"}},
]
out = edit([["removeElement", "b1", "LB000033"]], base)
assert [x["id"] for x in out["links"]] == ["l2"]
assert [e["name"] for e in out["boxes"][0]["input"]["elements"]] == ["보통인부"]
def test_갈래를_바꾸면_고른_것을_비움_표는_요소와_함께_못_둠():
el = call("elementOf", "인력", LABOR)
lg = edit([["setKind", "b1", "인력"], ["addElement", "b1", el], ["setKind", "b1", "재료"]])
assert lg["boxes"][0]["input"]["group"] == "재료"
assert lg["boxes"][0]["input"]["elements"] == []
t = edit([["setKind", "b1", "소요량"], ["setTable", "b1", {"key": "QF000421"}, "메쌓기(인력)"]])
assert t["boxes"][0]["input"]["table"] == {"key": "QF000421"}
assert t["boxes"][0]["name"] == "메쌓기(인력)" and t["boxes"][0]["unit"] == "-"
row = edit([["setTable", "b1", {"key": "QF000421", "row": 3}]], t)
assert row["boxes"][0]["input"]["table"] == {"key": "QF000421", "row": 3}
assert call("pillNames", row["boxes"][0]) == ["표"]
def test_변수_넣기_고치기_삭제_알약_이름():
v = {"name": "면적", "type": "number", "unit": "㎡", "example": 100, "range": [1, 1000]}
lg = edit([["setKind", "b1", "변수"], ["putVariable", "b1", v]])
box = lg["boxes"][0]
assert box["input"]["mode"] == "variable" and box["unit"] == "㎡"
assert call("pillNames", box) == ["면적"]
renamed = edit([["putVariable", "b1", {**v, "name": "연장", "unit": "m"}, "면적"]], lg)
assert call("pillNames", renamed["boxes"][0]) == ["연장"] and renamed["boxes"][0]["unit"] == "m"
gone = edit([["removeVariable", "b1", "연장"]], renamed)
assert gone["boxes"][0]["input"]["variables"] == [] and gone["boxes"][0]["unit"] == "-"
def test_변수_검사는_쉬운_말():
ok = {"name": "면적", "type": "number", "unit": "㎡", "example": 100, "range": [1, 1000]}
assert call("checkVariable", ok, []) is None
assert call("checkVariable", {**ok, "name": " "}, []) == "이름을 적어 주세요"
assert call("checkVariable", ok, [ok]) == "같은 이름의 변수가 이미 있음"
assert call("checkVariable", {**ok, "range": [9, 1]}, []) == "범위는 아래가 위보다 작아야 함"
assert call("checkVariable", {**ok, "example": 5000}, []) == "예시값이 범위 밖"
many = {**ok, "type": "choice", "choices": ["가"]}
assert call("checkVariable", many, []) == "선택 목록은 둘 이상"
bad = {**ok, "type": "choice", "choices": ["가", "나"], "example": "다"}
assert call("checkVariable", bad, []) == "예시값이 선택 목록에 없음"
def test_환율_유가전력_알약_이름에_날짜가_안_들어감():
fx = {"키": "FX1", "이름": "USD", "규격": "2026-08-14", "단위": "원/USD", "값": 1414.9}
assert call("elementOf", "환율", fx)["name"] == "USD"
oil = {
"키": "MO1",
"이름": "휘발유",
"규격": "2026-08-14",
"상세구분": "서울",
"단위": "원/L",
"값": 1900,
}
assert call("elementOf", "유가전력", oil)["name"] == "휘발유 서울"
def test_단위가_안_맞게_저장된_상자는_열_때_맞춤():
el1, el2 = call("elementOf", "인력", LABOR), call("elementOf", "인력", LABOR2)
lg = logic()
lg["boxes"][0]["input"] = {
"mode": "group",
"group": "인력",
"elements": [el1, el2],
"variables": [],
}
assert lg["boxes"][0]["unit"] == "-"
fixed = run([{"op": "fn", "name": "reconcileUnit", "args": [lg, "b1"]}])[0]
assert fixed["boxes"][0]["unit"] == "원/일"
again = run([{"op": "fn", "name": "reconcileUnit", "args": [fixed, "b1"]}])[0]
assert again == fixed
oil = {"ref": "X", "name": "경유", "spec": "", "unit": "원/L"}
lg["boxes"][0]["input"]["elements"].append(oil)
lg["boxes"][0]["unit"] = "원/일"
assert (
run([{"op": "fn", "name": "reconcileUnit", "args": [lg, "b1"]}])[0]["boxes"][0]["unit"]
== "-"
)
one = {
"mode": "variable",
"elements": [],
"variables": [{"name": "면적", "type": "number", "unit": "㎡", "example": 1}],
}
lg["boxes"][0]["input"] = one
assert (
run([{"op": "fn", "name": "reconcileUnit", "args": [lg, "b1"]}])[0]["boxes"][0]["unit"]
== "㎡"
)
+348
View File
@@ -0,0 +1,348 @@
"""M02 단가산출 다이어그램 Model · Eval 시험(35-7) — Node 로 `M02_DF_Model.ts` · `M02_DF_Eval.ts` 와 진짜 스프레드시트 엔진을 돌림.
· 견본 둘은 이 시험 안에서 만듦 — 마스터 · `_data` 에 기대지 않음 · 저장 안 함
· 입력(인력 석공 노임) + 변수(면적) → 함수(품 × 면적 × 노임) → 출력 노무비 · 직렬 · 병렬 · 순환 막음 · 선 규칙 · 출력 하나 · 원가 합 · 검사
· TypeScript 를 그 자리에서 CommonJS 로 옮겨 부름(`helper_spreadsheet_engine.cjs` 와 같은 길)
"""
from __future__ import annotations
import json
import re
import shutil
import subprocess
from pathlib import Path
import pytest
ROOT = Path(__file__).resolve().parents[2]
CF = ROOT / "M02_MasterTemplete" / "DataFlow"
pytestmark = pytest.mark.skipif(shutil.which("node") is None, reason="node 가 없음")
LOADER = r"""
const fs = require("fs");
const path = require("path");
const ROOT = process.argv[1];
const ts = require(path.join(ROOT, "config", "node_modules", "typescript"));
const cache = new Map();
function loadTs(file) {
const full = path.resolve(file);
if (cache.has(full)) return cache.get(full).exports;
const js = ts.transpileModule(fs.readFileSync(full, "utf8"), {
compilerOptions: { module: ts.ModuleKind.CommonJS, target: ts.ScriptTarget.ES2022 },
}).outputText;
const mod = { exports: {} };
cache.set(full, mod);
const req = (r) => {
if (r === "@ui/ui_template_locale") return { currentLanguageIndex: 0 };
if (r.endsWith(".css")) return {};
if (r.startsWith("@ui/")) return loadTs(path.join(ROOT, "ui_template", `${r.slice(4)}.ts`));
if (r.startsWith(".")) return loadTs(path.join(path.dirname(full), r.endsWith(".ts") ? r : `${r}.ts`));
return require(r);
};
new Function("exports", "module", "require", js)(mod.exports, mod, req);
return mod.exports;
}
const M = loadTs(path.join(ROOT, "M02_MasterTemplete", "DataFlow", "M02_DF_Model.ts"));
const E = loadTs(path.join(ROOT, "M02_MasterTemplete", "DataFlow", "M02_DF_Eval.ts"));
const out = (x) => console.log(JSON.stringify(x));
const add = (lg, kind, x = 0, y = 0) => M.addBox(lg, kind, x, y);
const join = (lg, a, ap, b, bp) => {
const r = M.link(lg, { box: a.id, name: ap }, { box: b.id, name: bp });
if (r.why) throw new Error(r.why);
return r.lg;
};
/** 함수 상자 시트 · 받는 칸 · 변수 · 출력 칸 한 번에 */
const func = (lg, box, cells, inputs, output, variables = []) =>
M.updateBox(lg, box.id, { func: { ...box.func, sheet: { ...box.func.sheet, 칸: cells }, inputs, variables,
output: { ...box.func.output, ...output } } });
"""
def _run(body: str) -> object:
result = subprocess.run(
["node", "-e", LOADER + body, str(ROOT)],
capture_output=True, text=True, encoding="utf-8", timeout=120, cwd=ROOT,
) # fmt: skip
assert result.returncode == 0, result.stderr
return json.loads(result.stdout.strip().splitlines()[-1])
# 견본 하나 — 입력(인력 석공) + 입력(변수 면적) → 함수(품 × 면적 × 노임) → 출력 노무비
ONE = r"""
let lg = M.newLogic("견본 하나", "원/㎡");
let r = add(lg, "input"); lg = r.lg; let man = r.box;
lg = M.updateBox(lg, man.id, { input: { mode: "group", group: "인력",
elements: [{ ref: "LB000033", name: "석공", spec: "", unit: "인" }], variables: [] } });
r = add(lg, "input"); lg = r.lg; let area = r.box;
lg = M.updateBox(lg, area.id, { input: { mode: "variable", elements: [],
variables: [{ name: "면적", type: "number", unit: "㎡", example: 10 }] } });
r = add(lg, "function"); lg = r.lg; let f = r.box;
lg = func(lg, f, { A2: { 값: "노임" }, A3: { 값: "면적" }, A4: { 값: "품" }, A5: { 값: "노무비" }, B5: { 식: "B4*B3*B2" } },
[{ port: "노임", cell: "B2" }, { port: "면적", cell: "B3" }], { cell: "B5", name: "노무비", unit: "원" },
[{ name: "품", type: "number", unit: "인/㎡", example: 0.2, cell: "B4" }]);
r = add(lg, "output"); lg = r.lg; let o = r.box;
lg = join(lg, man, "석공", f, "노임");
lg = join(lg, area, "면적", f, "면적");
lg = join(lg, f, "노무비", o, "값");
const V = { [man.id]: { 석공: 268908 } };
"""
# 견본 둘 — 직렬 · 병렬: f1 = 면적 × 2 · f2 = 노임 × 0.1 · f3 = f1 + f2(병렬 모음) · f4 = f3 × 1.1(직렬)
TWO = r"""
let lg = M.newLogic("견본 둘");
let r = add(lg, "input"); lg = r.lg; const inp = r.box;
lg = M.updateBox(lg, inp.id, { input: { mode: "group", group: "인력",
elements: [{ ref: "LB000002", name: "보통인부", spec: "", unit: "인" }],
variables: [{ name: "면적", type: "number", unit: "㎡", example: 5 }] } });
const fn = (cells, inputs) => { r = add(lg, "function"); lg = r.lg; const b = r.box;
lg = func(lg, b, cells, inputs, { cell: "B9", name: "결과" }); return b; };
const f1 = fn({ B9: { 식: "B1*2" } }, [{ port: "가", cell: "B1" }]);
const f2 = fn({ B9: { 식: "B1*0.1" } }, [{ port: "가", cell: "B1" }]);
const f3 = fn({ B9: { 식: "B1+B2" } }, [{ port: "가", cell: "B1" }, { port: "나", cell: "B2" }]);
const f4 = fn({ B9: { 식: "ROUND(B1*1.1,0)" } }, [{ port: "가", cell: "B1" }]);
lg = join(lg, inp, "면적", f1, "가");
lg = join(lg, inp, "보통인부", f2, "가");
lg = join(lg, f1, "결과", f3, "가");
lg = join(lg, f2, "결과", f3, "나");
lg = join(lg, f3, "결과", f4, "가");
r = add(lg, "output"); lg = r.lg; const oExp = r.box; lg = M.updateBox(lg, oExp.id, { output: { cost: "경비", unit: "원" } });
r = add(lg, "output"); lg = r.lg; const oVal = r.box; lg = M.updateBox(lg, oVal.id, { output: { cost: "값", unit: "㎡" } });
r = add(lg, "output"); lg = r.lg; const oMat = r.box; lg = M.updateBox(lg, oMat.id, { output: { cost: "재료비", unit: "원" } });
lg = join(lg, f4, "결과", oExp, "값");
lg = join(lg, f1, "결과", oVal, "값");
lg = join(lg, inp, "면적", oMat, "값");
const V = { [inp.id]: { 보통인부: 172068 } };
"""
def test_견본하나_입력_변수_함수_출력_노무비() -> None:
got = _run(
ONE
+ "out([E.evalLogic(lg, V), E.checkLogic(lg, V), lg.boxes.map(b => [b.id, b.kind, b.name, b.color, b.unit])]);"
)
ev, check, boxes = got
assert ev["ports"]["b3"] == {"노무비": 537816} # 0.2 × 10 × 268,908
assert ev["outputs"] == {"b4": 537816}
assert ev["cost"] == {"노무비": 537816, "재료비": 0, "경비": 0, "값들": {}, "계": 537816}
assert ev["errors"] == [] and check == []
assert boxes == [
["b1", "input", "입력", "#2f80ed", "-"],
["b2", "input", "입력 2", "#2f80ed", "-"],
["b3", "function", "함수", "#7b4fd6", "-"],
["b4", "output", "출력", "#27ae60", "-"],
]
def test_변수값_바꾸면_값이_바뀜() -> None:
got = _run(
ONE
+ "out(E.evalLogic(lg, { ...V, [area.id]: { 면적: 20 }, [f.id]: { 품: 0.1 } }).cost.계);"
)
assert got == 0.1 * 20 * 268908
def test_견본둘_직렬_병렬_원가_합() -> None:
got = _run(
TWO
+ "const ev = E.evalLogic(lg, V);\nout([ev.ports, ev.cost, ev.errors, E.checkLogic(lg, V)]);"
)
ports, cost, errors, check = got
f1, f2, f3, f4 = (ports[k]["결과"] for k in ("b2", "b3", "b4", "b5"))
assert (f1, f2, f3, f4) == (10, 17206.8, 17216.8, 18938)
assert cost == {"노무비": 0, "재료비": 5, "경비": 18938, "값들": {"출력 2": 10}, "계": 18943}
assert errors == [] and check == []
def test_선_규칙() -> None:
got = _run(
TWO
+ "const t = (a, ap, b, bp) => M.link(lg, { box: a.id, name: ap }, { box: b.id, name: bp }).why;\n"
"out([t(f4, '결과', f3, '나'), t(f3, '결과', f1, '가'), t(oExp, '값', f1, '가'), t(f1, '결과', inp, '면적'),"
" t(f3, '결과', f3, '가'), t(f2, '결과', f3, '가'), t(f1, '없음', f4, '가'), t(inp, '면적', oExp, '값'),"
" M.link(lg, { box: f1.id, name: '결과' }, { box: f3.id, name: '가' }).lg === lg]);"
)
assert got == [
"이으면 돌고 도는 선이 됨",
"이으면 돌고 도는 선이 됨",
"출력 상자에서는 선이 나가지 않음",
"입력 상자로는 선이 들어가지 않음",
"같은 상자끼리는 못 이음",
"이미 선이 들어온 자리",
"알약이 없음",
"이미 선이 들어온 자리",
True,
]
def test_알약_출력은_하나() -> None:
got = _run(
TWO
+ "lg = M.updateBox(lg, inp.id, { input: { ...lg.boxes[0].input, table: { key: 'QF000421' } } });\n"
"out([M.portsOut(lg.boxes[0]), M.portsIn(lg.boxes[0]), M.portsOut(lg.boxes[3]), M.portsIn(lg.boxes[3]),"
" M.portsOut(lg.boxes.find(b => b.id === oExp.id)), M.portsIn(lg.boxes.find(b => b.id === oExp.id))]);"
)
in_out, in_in, f_out, f_in, o_out, o_in = got
assert in_out == [
{"name": "보통인부", "unit": "인"},
{"name": "표", "unit": "-"},
{"name": "면적", "unit": "㎡"},
]
assert in_in == [] and f_out == [{"name": "결과", "unit": "-"}]
assert f_in == [{"name": "가", "unit": "-"}, {"name": "나", "unit": "-"}]
assert o_out == [] and o_in == [{"name": "값", "unit": "원"}]
def test_선_끊기_상자_지우기_검사() -> None:
got = _run(
TWO + "const cut = lg.links.find(l => l.to.box === f3.id && l.to.name === '나').id;\n"
"const a = M.unlink(lg, cut);\n"
"const b = M.removeBox(lg, f3.id);\n"
"const c = M.removeBox(M.removeBox(M.removeBox(lg, oExp.id), oVal.id), oMat.id);\n"
"out([E.evalLogic(a, V).ports[f3.id], E.checkLogic(a, V), b.links.length, lg.links.length,"
" E.checkLogic(b, V), E.checkLogic(c, V), E.checkLogic(lg, {})]);"
)
f3_after, check_a, n_b, n_all, check_b, check_c, check_novalue = got
assert f3_after == {"결과": 10}
assert check_a == ["「함수 3」 의 받는 알약 「나」 에 선이 없음"]
assert (n_b, n_all) == (5, 8)
assert check_b == [
"「함수 4」 의 받는 알약 「가」 에 선이 없음",
]
assert check_c == ["출력 상자가 없음"]
assert check_novalue == ["「입력」 의 「보통인부」 값이 아직 없음"]
def test_오류칸_순환칸_숫자아닌_출력() -> None:
got = _run(
ONE
+ "lg = M.updateBox(lg, f.id, { func: { ...lg.boxes[2].func, sheet: { ...lg.boxes[2].func.sheet,"
" 칸: { ...lg.boxes[2].func.sheet.칸, B5: { 식: 'B4/0' } } } } });\n"
"const a = E.evalLogic(lg, V);\n"
"lg = M.updateBox(lg, f.id, { func: { ...lg.boxes[2].func, sheet: { ...lg.boxes[2].func.sheet,"
" 칸: { ...lg.boxes[2].func.sheet.칸, B5: { 식: 'B6' }, B6: { 식: 'B5' } } } } });\n"
"const b = E.evalLogic(lg, V);\n"
"out([a.errors, a.cost.계, b.errors]);"
)
a_err, total, b_err = got
assert a_err == ["「함수」 B5 칸 계산이 막힘 — #DIV/0!", "「출력」 에 숫자가 아닌 값이 들어옴"]
assert total == 0
assert "「함수」 B5 칸 계산이 막힘 — #CYCLE! 돌고 도는 참조" in b_err
def test_자리_색_단위() -> None:
got = _run(
ONE
+ "let g = M.moveBox(lg, f.id, 120, 40); g = M.setColor(g, f.id, '#ff0000'); g = M.setUnit(g, f.id, '원/㎡');\n"
"g = M.setUnit(g, o.id, ' ');\n"
"const fb = g.boxes.find(b => b.id === f.id), ob = g.boxes.find(b => b.id === o.id);\n"
"out([fb.x, fb.y, fb.color, fb.unit, fb.func.output.unit, ob.unit, ob.output.unit, lg.boxes[2].x]);"
)
assert got == [120, 40, "#ff0000", "원/㎡", "원/㎡", "-", "-", 0]
def test_사용자_요소_값() -> None:
got = _run(
ONE
+ "lg = { ...lg, userElements: [{ id: 'U1', group: '인력', name: '석공', spec: '', unit: '인', value: 300000 }] };\n"
"lg = M.updateBox(lg, man.id, { input: { ...lg.boxes[0].input, elements: [{ ref: 'U1', name: '석공', spec: '', unit: '인' }] } });\n"
"out(E.evalLogic(lg, {}).cost.노무비);"
)
assert got == 0.2 * 10 * 300000
def test_화면_글에_금지_낱말_없음() -> None:
for name in ("M02_DF_Model.ts", "M02_DF_Eval.ts"):
src = (CF / name).read_text(encoding="utf-8")
block = src[src.index("const TEXT = {") : src.index("} as const;")]
korean = [ko for _, ko in re.findall(r'(\w+):\s*\[\s*"([^"]*)"', block)]
assert korean
for text in korean:
for bad in ("키", "식", "호표", "덧줄", "비목", "정본", "자체"):
assert bad not in text, (name, bad, text)
def test_알약_이름_규칙() -> None:
got = _run(
"let lg = M.newLogic(); let r = add(lg, 'input'); lg = r.lg;\n"
"lg = M.updateBox(lg, r.box.id, { input: { mode: 'group', group: '재료',"
" elements: [{ ref: 'MT1', name: '시멘트', spec: '40kg', unit: '포' }], table: { key: 'QF000421', row: 2 },"
" variables: [{ name: '길이', type: 'number', unit: 'm', example: 3 }] } });\n"
"out(M.portsOut(lg.boxes[0]).map(p => p.name));"
)
assert got == ["시멘트", "표", "길이"]
def test_새_함수_상자에_놓으면_받는_알약이_생김() -> None:
got = _run(
ONE + "r = add(lg, 'function'); lg = r.lg; const g = r.box;\n"
"let a = M.link(lg, { box: man.id, name: '석공' }, { box: g.id, name: '' });\n"
"let b = M.link(a.lg, { box: area.id, name: '면적' }, { box: g.id, name: '' });\n"
"let c = M.link(b.lg, { box: f.id, name: '노무비' }, { box: g.id, name: '석공' });\n"
"let d = M.link(c.lg, { box: f.id, name: '노무비' }, { box: g.id, name: '' });\n"
"const gb = (x) => x.boxes.find(y => y.id === g.id);\n"
"const inputs = gb(d.lg).func.inputs;\n"
"const check = E.checkLogic(d.lg, V).filter(x => x.includes('칸에 놓지 않음'));\n"
"const placed = M.updateBox(d.lg, g.id, { func: { ...gb(d.lg).func, inputs: inputs.map(i => i.port === '면적' ? { ...i, cell: 'B2' } : i) } });\n"
"const cutAll = placed.links.filter(l => l.to.box === g.id).reduce((x, l) => M.unlink(x, l.id), placed);\n"
"const removed = M.removeBox(d.lg, man.id);\n"
"out([a.why, b.why, c.why, d.why, inputs, M.portsIn(gb(d.lg)).map(p => p.name), d.lg.links.slice(-2).map(l => l.to),"
" check, gb(cutAll).func.inputs, gb(removed).func.inputs.map(i => i.port),"
" M.link(lg, { box: man.id, name: '석공' }, { box: o.id, name: '없음' }).why]);"
)
a, b, c, d, inputs, ports, last_to, check, cut_all, removed, to_output = got
assert (a, b, c, d) == (None, None, "이미 선이 들어온 자리", None)
assert inputs == [
{"port": "석공", "cell": ""},
{"port": "면적", "cell": ""},
{"port": "노무비", "cell": ""},
]
assert ports == ["석공", "면적", "노무비"]
assert last_to == [{"box": "b5", "name": "면적"}, {"box": "b5", "name": "노무비"}]
assert check == [
"「함수 2」 의 받는 알약 「석공」 을 칸에 놓지 않음",
"「함수 2」 의 받는 알약 「면적」 을 칸에 놓지 않음",
"「함수 2」 의 받는 알약 「노무비」 을 칸에 놓지 않음",
]
assert cut_all == [{"port": "면적", "cell": "B2"}] # 칸에 놓인 것만 남음
assert removed == ["면적", "노무비"] # 석공 상자를 지우면 칸 없는 「석공」 받는 알약도 지움
assert to_output == "알약이 없음" # 출력 상자는 받는 알약을 새로 만들지 않음
def test_같은_이름_받는_알약은_번호() -> None:
got = _run(
ONE + "r = add(lg, 'function'); lg = r.lg; const g = r.box;\n"
"let x = M.link(lg, { box: man.id, name: '석공' }, { box: g.id, name: '' }).lg;\n"
"x = M.link(x, { box: f.id, name: '노무비' }, { box: g.id, name: '' }).lg;\n"
"r = add(x, 'input'); x = r.lg; const m2 = r.box;\n"
"x = M.updateBox(x, m2.id, { input: { mode: 'group', group: '인력', elements: [{ ref: 'LB1', name: '석공', spec: '', unit: '인' }], variables: [] } });\n"
"x = M.link(x, { box: m2.id, name: '석공' }, { box: g.id, name: '' }).lg;\n"
"x = M.link(x, { box: area.id, name: '면적' }, { box: g.id, name: '할증' }).lg;\n"
"out(x.boxes.find(y => y.id === g.id).func.inputs.map(i => i.port));"
)
assert got == ["석공", "노무비", "석공 (2)", "할증"]
def test_함수_출력_기본_이름은_겹치지_않음_받는_알약은_번호() -> None:
got = _run(
"let lg = M.newLogic(); let r = add(lg, 'function'); lg = r.lg; const f1 = r.box;\n"
"r = add(lg, 'function'); lg = r.lg; const f2 = r.box;\n"
"r = add(lg, 'function'); lg = r.lg; const f3 = r.box;\n"
"lg = M.removeBox(lg, f2.id); r = add(lg, 'function'); lg = r.lg; const f4 = r.box;\n"
"lg = M.link(lg, { box: f1.id, name: '결과 1' }, { box: f3.id, name: '' }).lg;\n"
"lg = M.link(lg, { box: f4.id, name: '결과 2' }, { box: f3.id, name: '' }).lg;\n"
"r = add(lg, 'function'); lg = r.lg; const f5 = r.box;\n"
"lg = M.updateBox(lg, f5.id, { func: { ...f5.func, output: { ...f5.func.output, name: '결과 1' } } });\n"
"lg = M.link(lg, { box: f5.id, name: '결과 1' }, { box: f3.id, name: '' }).lg;\n"
"out([[f1, f3, f4, f5].map(b => [b.func.output.name, b.func.sheet.칸.A1.값]),"
" lg.boxes.find(b => b.id === f3.id).func.inputs.map(i => i.port)]);"
)
names, ports = got
assert names == [
["결과 1", "결과 1"],
["결과 3", "결과 3"],
["결과 2", "결과 2"],
["결과 4", "결과 4"],
]
assert ports == ["결과 1", "결과 2", "결과 1 (2)"]
+353
View File
@@ -0,0 +1,353 @@
"""단가산출 다이어그램 시험 길(`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
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 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/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"]})
+275
View File
@@ -0,0 +1,275 @@
"""M02 단가산출 다이어그램 견본(35-7) 시험 — `M02_MasterTemplete/DataFlow/_sample/*.json` · `M02_DF_Sample.ts`.
· 견본 셋 모양 — 상자(입력 · 함수 · 출력) · 알약 이름 규칙 · 선(받는 자리마다 하나 · 출력에서 안 나감 · 순환 없음)
· 견본 ① 원가 = 지금 `/calc`(GF000219 · 같은 예시값 · 끝수 원 미만) — 함수 상자 시트를 우리 스프레드시트 엔진
(Node 번들)으로 차례대로 셈 · 입력 값(노임 · 품셈 표)은 마스터에서 읽기만
· ③ 값 = `/calc`(GF000133) · ② = 노무비 × 율 / 100
· `sampleList` · `sampleLoad`(사본 · key 없음) — Node 로 바로 돌림 · 파일 안 씀
"""
from __future__ import annotations
import json
import re
import shutil
import subprocess
from decimal import Decimal
from pathlib import Path
import pytest
ROOT = Path(__file__).resolve().parents[2]
FOLDER = ROOT / "M02_MasterTemplete" / "DataFlow"
SAMPLE_TS = FOLDER / "M02_DF_Sample.ts"
SAMPLES = {p.stem: json.loads(p.read_text(encoding="utf-8")) for p in sorted((FOLDER / "_sample").glob("*.json"))} # fmt: skip
HOOKS = r"""
import { readFileSync } from "node:fs";
import { fileURLToPath } from "node:url";
import { stripTypeScriptTypes } from "node:module";
export async function load(url, ctx, next) {
if (!url.endsWith(".ts")) return next(url, ctx);
let src = readFileSync(fileURLToPath(url), "utf8");
src = src.replace(/import\.meta\.glob(?:<[\s\S]*?>)?\([^)]*\)/g, "globalThis.__samples");
src = src.replace(/import\s*\{([^}]*)\}\s*from\s*"([^"]+)";/g, (m, names) => {
const list = names.split(",").map((x) => x.trim()).filter((x) => x && !x.startsWith("type "));
return list.length ? `const { ${list.join(", ")} } = globalThis.__stub;` : "";
});
return { format: "module", source: stripTypeScriptTypes(src, { mode: "transform" }), shortCircuit: true };
}
"""
def _node(body: str) -> object:
files = {f"./_sample/{k}.json": v for k, v in SAMPLES.items()}
script = (
"import { register } from 'node:module';\n"
f"register('data:text/javascript,' + encodeURIComponent({json.dumps(HOOKS)}));\n"
"globalThis.__stub = { currentLanguageIndex: 0 };\n"
f"globalThis.__samples = {json.dumps(files, ensure_ascii=False)};\n"
f"const S = await import({json.dumps(SAMPLE_TS.as_uri())});\n"
"const out = (x) => process.stdout.write(JSON.stringify(x));\n" + body
)
result = subprocess.run(
["node", "--no-warnings", "--input-type=module", "-e", script],
capture_output=True, text=True, encoding="utf-8", timeout=120, cwd=ROOT,
) # fmt: skip
assert result.returncode == 0, result.stderr[-2000:]
return json.loads(result.stdout)
# ── 모양 ─────────────────────────────────────────────────────────────────
def _ports_out(box: dict) -> list[str]:
if box["kind"] == "function":
return [box["func"]["output"]["name"]]
if box["kind"] == "output":
return []
inp = box["input"]
return (
[e["name"] for e in inp.get("elements", [])]
+ (["표"] if inp.get("table") else [])
+ [v["name"] for v in inp.get("variables", [])]
)
def _ports_in(box: dict) -> list[str]:
if box["kind"] == "function":
return [i["port"] for i in box["func"]["inputs"]]
return ["값"] if box["kind"] == "output" else []
def _order(lg: dict) -> list[dict]:
"""함수 상자 차례(앞 값을 받는 것이 뒤) — 순환이면 실패"""
boxes = {b["id"]: b for b in lg["boxes"]}
need = {b["id"]: {link["from"]["box"] for link in lg["links"] if link["to"]["box"] == b["id"]} for b in lg["boxes"]} # fmt: skip
done: list[str] = []
while len(done) < len(boxes):
ready = [i for i in boxes if i not in done and need[i] <= set(done)]
assert ready, "선이 돌고 돎"
done += ready
return [boxes[i] for i in done]
# Model 기본 색(`M02_DF_Model.ts` BOX_COLOR) — 견본 색은 이것과 같아야 함
BOX_COLOR = dict(
re.findall(
r'(\w+): "(#[0-9a-fA-F]{6})"',
re.search(
r"BOX_COLOR[^{]*\{([^}]*)\}", (FOLDER / "M02_DF_Model.ts").read_text(encoding="utf-8")
).group(1),
)
)
@pytest.mark.parametrize("name", ["masonry", "rate", "value"])
def test_견본_모양(name: str) -> None:
lg = SAMPLES[name]
assert lg["key"] and lg["name"] and lg["resultUnit"] and lg["userElements"] == []
boxes = {b["id"]: b for b in lg["boxes"]}
assert len(boxes) == len(lg["boxes"])
sheets = [b["func"]["sheet"]["id"] for b in lg["boxes"] if b["kind"] == "function"]
assert len(set(sheets)) == len(sheets)
for b in lg["boxes"]:
assert b["kind"] in ("input", "function", "output")
assert (
b["color"] == BOX_COLOR[b["kind"]]
and b["unit"]
and isinstance(b["x"], int)
and isinstance(b["y"], int)
)
if b["kind"] == "function":
fn = b["func"]
cells = fn["sheet"]["칸"]
assert "식" in cells[fn["output"]["cell"]] # 출력 하나 = 수식 칸
assert len({i["cell"] for i in fn["inputs"]}) == len(fn["inputs"])
if b["kind"] == "output":
assert b["output"]["cost"] in ("노무비", "재료비", "경비", "값")
units = {e["unit"] for e in (b.get("input") or {}).get("elements") or []}
if len(units) == 1: # 요소 단위가 모두 같으면 상자 단위도 그것(공통 단위 규칙)
assert b["unit"] == next(iter(units)), b["id"]
for link in lg["links"]:
a, z = boxes[link["from"]["box"]], boxes[link["to"]["box"]]
assert link["from"]["name"] in _ports_out(a), link
assert link["to"]["name"] in _ports_in(z), link
# 받는 자리마다 선 하나 · 출력에서 나가는 선 없음 · 왼 → 오
for b in lg["boxes"]:
for port in _ports_in(b):
got = [x for x in lg["links"] if x["to"]["box"] == b["id"] and x["to"]["name"] == port]
assert len(got) == 1, (b["id"], port)
for link in lg["links"]:
assert boxes[link["from"]["box"]]["kind"] != "output"
assert boxes[link["from"]["box"]]["x"] < boxes[link["to"]["box"]]["x"]
_order(lg)
def test_견본_목록_열기() -> None:
got = _node(
"out({ list: S.sampleList(), one: S.sampleLoad('masonry'), none: S.sampleLoad('없음') });"
)
assert [x["id"] for x in got["list"]] == ["masonry", "rate", "value"]
assert all(x["note"] for x in got["list"])
assert got["one"]["key"] is None and got["one"]["boxes"] == SAMPLES["masonry"]["boxes"]
assert got["none"] is None
# ── 값 ───────────────────────────────────────────────────────────────────
@pytest.fixture(scope="module")
def master():
from M01_MasterData import M01_MasterData_Store as store
cm = store.cm
files = cm.load(folder=store.FOLDER)
return cm, files
def _table_rows(table: dict) -> list[list]:
"""품셈 표 → 블록(머리 줄 + 줄) — 서버 「표」 와 같은 규칙(범위 = 「이름·아래」 「이름·위」 · 숫자 하나면 둘 다)"""
conds, lines = table.get("조건") or {}, table.get("줄") or []
names = [*conds, *(table.get("값칸") or {})]
split = {
n for n in names if conds.get(n) == "범위" or any(isinstance(r.get(n), list) for r in lines)
}
head = [h for n in names for h in ((f"{n}·아래", f"{n}·위") if n in split else (n,))]
rows = []
for r in lines:
line: list = []
for n in names:
v = r.get(n)
if n in split:
line += list(v) if isinstance(v, list) else [v, v]
else:
line.append(v)
rows.append(line)
return [head, *rows]
def _plain(v):
return float(v) if isinstance(v, Decimal) else v
def _run(lg: dict, master) -> dict:
"""상자 차례대로 — 입력 값(예시값 · 노임 · 표) → 함수 상자 시트에 넣고 엔진으로 셈 → 출력 상자 값"""
from common_util.common_util_spreadsheet import recalc_workbooks
cm, files = master
m = cm.mf.Master(files)
tables = {str(t["키"]): t for d in files.values() if d.get("그룹") in cm.mf.TABLE_GROUPS for t in d.get("표") or []} # fmt: skip
ports: dict[str, dict] = {}
for b in _order(lg):
into = {x["to"]["name"]: ports[x["from"]["box"]][x["from"]["name"]] for x in lg["links"] if x["to"]["box"] == b["id"]} # fmt: skip
if b["kind"] == "input":
inp = b["input"]
got = {v["name"]: v["example"] for v in inp.get("variables", [])}
got |= {
e["name"]: _plain(cm.mf.element(m, e["ref"], {})[0])
for e in inp.get("elements", [])
}
if inp.get("table"):
got["표"] = _table_rows(tables[inp["table"]["key"]])
ports[b["id"]] = got
elif b["kind"] == "function":
fn = b["func"]
sheet = json.loads(json.dumps(fn["sheet"]))
for i in fn["inputs"]:
value = into[i["port"]]
col, row = ord(i["cell"][0]) - 65, int(i["cell"][1:])
block = value if isinstance(value, list) else [[value]]
for r, line in enumerate(block):
for c, x in enumerate(line):
if x is not None:
sheet["칸"][f"{chr(65 + col + c)}{row + r}"] = {"값": _plain(x)}
book = {"종류": "통합문서", "판": 1, "열": "cf", "서식": [{}], "시트": [sheet]}
out = recalc_workbooks([book])[0][sheet["id"]][fn["output"]["cell"]]
assert "수" in out, out
ports[b["id"]] = {fn["output"]["name"]: float(out["수"])}
else:
ports[b["id"]] = {"값": into["값"]}
return {b["id"]: ports[b["id"]]["값"] for b in lg["boxes"] if b["kind"] == "output"}
def test_견본_표_그룹이_마스터와_같음(master) -> None:
"""표 입력 상자의 group = 그 표가 든 마스터 파일 그룹(소요량 · 계수)"""
cm, files = master
group = {str(t["키"]): d["그룹"] for d in files.values() if d.get("그룹") in cm.mf.TABLE_GROUPS for t in d.get("표") or []} # fmt: skip
for lg in SAMPLES.values():
for b in lg["boxes"]:
table = b.get("input", {}).get("table")
if table:
assert b["input"]["group"] == group[table["key"]], (b["id"], table["key"])
def test_견본1_원가_calc_와_같음(master) -> None:
from M01_MasterData.LogicBuilder import M01_LB_Calc as lb
cm, files = master
row = next(
r
for d in files.values()
if d.get("그룹") == "로직"
for r in d["줄"]
if r["키"] == "GF000219"
)
given = {v["name"]: v["example"] for v in SAMPLES["masonry"]["boxes"][0]["input"]["variables"]}
want = lb.calc(json.loads(json.dumps(row, default=_plain)), {**given, "초과증가율": 80}, {})
assert want["ok"], want
got = _run(SAMPLES["masonry"], master)["out-pay"]
print(f"\n견본 ① 노무비 {got} · /calc {want['sums']['계']}")
assert abs(got - float(want["sums"]["계"])) < 1
def test_견본3_값_calc_와_같음_견본2_비율(master) -> None:
from M01_MasterData.LogicBuilder import M01_LB_Calc as lb
cm, files = master
row = next(
r
for d in files.values()
if d.get("그룹") == "로직"
for r in d["줄"]
if r["키"] == "GF000133"
)
given = {v["name"]: v["example"] for v in SAMPLES["value"]["boxes"][0]["input"]["variables"]}
given["qt"] = given.pop("q") # 견본은 정함대로 「q」 · 기존 로직 입력 이름은 「qt」
want = lb.calc(json.loads(json.dumps(row, default=_plain)), given, {})
assert _run(SAMPLES["value"], master)["out-q"] == float(want["result"])
assert _run(SAMPLES["rate"], master)["out-mat"] == 100000 * 3 / 100