"""M01 마스터 데이터 API — 계약 `resources/master_data/_화면_계약.md`. 임시 폴더에 master_data 첫 층 JSON 사본을 두고 읽기 · 시험 계산 · 저장 · 409 · 422. """ from __future__ import annotations import json import shutil from decimal import Decimal from functools import cache from pathlib import Path import pytest from fastapi import FastAPI from fastapi.testclient import TestClient from M01_MasterData import M01_MasterData_Router as router_module from M01_MasterData import M01_MasterData_Store as store REAL = store.FOLDER LABOR = "인력.json" LOGIC_BOOK = "건설품셈" LOGIC_NUM = "공통 3-3-1 암발파(미진동굴착 TYPE-Ⅰ)" # 화약취급공 1016 · 보통인부 1002 LOGIC_INPUTS = {"지역": "전국평균", "보정작업": "해당 없음"} @cache def key_of(ref: str) -> str: """「ID:원문번호」 → 키(실제 폴더).""" return _whole().get(ref)["키"] @cache def _whole(): return store.cm.master(REAL) LOGIC_KEY = key_of("GC:" + LOGIC_NUM) @pytest.fixture def client(tmp_path: Path, monkeypatch) -> TestClient: for path in REAL.glob("*.json"): if not path.name.startswith("_") or path.name == store.mk.BOOK.name: shutil.copy(path, tmp_path / path.name) monkeypatch.setattr(store, "FOLDER", tmp_path) app = FastAPI() app.include_router(router_module.router) return TestClient(app) def _get(client: TestClient, url: str, **params) -> dict: res = client.get(url, params=params) assert res.status_code == 200, res.text return res.json() def _calc(client: TestClient, **inputs) -> dict: res = client.post( "/api/m01/calc", json={"book": LOGIC_BOOK, "key": LOGIC_KEY, "inputs": {**LOGIC_INPUTS, **inputs}}, ) assert res.status_code == 200, res.text return res.json() def _labor_row(client: TestClient, number: str) -> tuple[dict, str]: got = _get(client, "/api/m01/rows", file=LABOR, q="", size=500) return next(r for r in got["rows"] if r["원문번호"] == number), got["version"] def test_그룹_파일_줄_쪽나눔_찾기(client: TestClient) -> None: groups = {g["group"]: g for g in _get(client, "/api/m01/groups")["groups"]} assert list(groups) == ["인력", "재료", "기계", "소요량", "계수", "환율", "요율", "로직"] files = _get(client, "/api/m01/groups/인력/files")["files"] assert LABOR in [f["file"] for f in files] assert groups["인력"]["files"] == len(files) page2 = _get(client, "/api/m01/rows", file=LABOR, page=2, size=10) assert page2["page"] == 2 and len(page2["rows"]) == 10 and page2["total"] > 20 hit = _get(client, "/api/m01/rows", file=LABOR, q="보통인부") assert [r["원문번호"] for r in hit["rows"]] == ["1002"] assert _get(client, "/api/m01/rows", file=LABOR, q=key_of("LB:1002"))["total"] == 1 assert client.get("/api/m01/rows", params={"file": "없음.json"}).status_code == 404 assert client.get("/api/m01/rows", params={"file": "../main.py"}).status_code == 404 def test_표형_요소_표_목록과_표_하나(client: TestClient) -> None: file = "소요량_건설품셈_03장_토공사.json" listed = _get(client, "/api/m01/tables", file=file, q="암발파") one = next(t for t in listed["tables"] if t["원문번호"] == LOGIC_NUM) assert one["count"] >= 1 and "줄" not in one table = _get(client, "/api/m01/table", file=file, key=one["키"])["table"] assert len(table["줄"]) == one["count"] and table["값칸"] == one["값칸"] assert client.get("/api/m01/tables", params={"file": LABOR}).status_code == 400 def test_로직_목록_하나_시험계산(client: TestClient) -> None: chapter = "공통 03장 토공사" listed = _get(client, "/api/m01/logics", book=LOGIC_BOOK, chapter=chapter, q="암발파")["logics"] assert LOGIC_KEY in [x["키"] for x in listed] assert all(x["chapter"] == chapter and not x["blocked"] for x in listed) files = _get(client, "/api/m01/groups/로직/files")["files"] assert files[0]["file"].startswith("로직_산림품셈_") and files[-1]["chapter"].startswith( "유지관리" ) assert _get(client, "/api/m01/logics", blocked=1)["logics"] == [] one = _get(client, "/api/m01/logic", book=LOGIC_BOOK, key=LOGIC_KEY) assert one["file"] == "로직_건설품셈_03장_토공사.json" and one["blocked"] is False got = _calc(client) assert got["ok"] is True lines = {x["이름"]: x for x in got["lines"]} assert Decimal(str(lines["화약취급공"]["금액"])) == Decimal( str(lines["화약취급공"]["수량"]) ) * Decimal(str(lines["화약취급공"]["단가"])) assert got["sums"]["계"] == pytest.approx(sum(x["금액"] for x in got["lines"])) stop = _calc(client, 없는입력=1) assert stop["ok"] is False and "모르는 입력" in stop["reason"] def test_저장_고침은_파일에_쓰고_낡은_판본은_409(client: TestClient) -> None: row, version = _labor_row(client, "1016") before = _calc(client)["sums"]["계"] edited = {**row, "값": row["값"] + 1000} body = { "files": [ { "file": LABOR, "version": version, "changes": [{"op": "edit", "key": row["키"], "row": edited}], } ] } res = client.post("/api/m01/save", json=body) assert res.status_code == 200, res.text new_version = res.json()["files"][0]["version"] assert new_version != version assert _labor_row(client, "1016") == (edited, new_version) assert _calc(client)["sums"]["계"] == pytest.approx(before + 1000 * 0.04) # 화약취급공 0.040 인 data = json.loads((store.FOLDER / LABOR).read_text(encoding="utf-8")) assert ( data["그룹"] == "인력" and len(data["줄"]) == _get(client, "/api/m01/rows", file=LABOR)["total"] ) again = client.post("/api/m01/save", json=body) # 옛 판본 그대로 assert again.status_code == 409 and again.json()["detail"]["stale"] == [LABOR] def test_저장_검사에_걸리면_아무것도_안_씀(client: TestClient) -> None: row, version = _labor_row(client, "1016") raw = (store.FOLDER / LABOR).read_bytes() book = (store.FOLDER / store.mk.BOOK.name).read_bytes() gone = {"file": LABOR, "version": version, "changes": [{"op": "delete", "key": row["키"]}]} res = client.post("/api/m01/save", json={"files": [gone]}) assert res.status_code == 422 and any(row["키"] in x for x in res.json()["detail"]["errors"]) bad_form = {**_labor_row(client, "1002")[0], "키": "", "원문번호": "9999"} del bad_form["출처"] res = client.post( "/api/m01/save", json={"files": [{**gone, "changes": [{"op": "add", "row": bad_form}]}]} ) assert res.status_code == 422 and any("출처" in x for x in res.json()["detail"]["errors"]) dup = {**gone, "changes": [{"op": "add", "row": _labor_row(client, "1002")[0]}]} res = client.post("/api/m01/save", json={"files": [dup]}) assert res.status_code == 422 and any( "원문번호 겹침" in x for x in res.json()["detail"]["errors"] ) rekey = { **gone, "changes": [{"op": "edit", "key": row["키"], "row": {**row, "키": "LB999999"}}], } assert client.post("/api/m01/save", json={"files": [rekey]}).status_code == 400 assert (store.FOLDER / LABOR).read_bytes() == raw assert (store.FOLDER / store.mk.BOOK.name).read_bytes() == book def test_더함_지움_로직_파일도_같은_길(client: TestClient) -> None: one = _get(client, "/api/m01/logic", book=LOGIC_BOOK, key=LOGIC_KEY) copy = {**one["logic"], "키": "", "원문번호": LOGIC_NUM + " 사본"} add = {"file": one["file"], "version": one["version"], "changes": [{"op": "add", "row": copy}]} res = client.post("/api/m01/save", json={"files": [add]}) assert res.status_code == 200, res.text book = json.loads((store.FOLDER / store.mk.BOOK.name).read_text(encoding="utf-8")) new = f"GC{book['다음']['GC'] - 1:06d}" # 대장의 다음 번호를 받음 assert book["키"][new] == {"원문번호": LOGIC_NUM + " 사본", "파일": one["file"]} assert _get(client, "/api/m01/logic", key=new)["blocked"] is False drop = { "file": one["file"], "version": res.json()["files"][0]["version"], "changes": [{"op": "delete", "key": new}], } assert client.post("/api/m01/save", json={"files": [drop]}).status_code == 200 assert client.get("/api/m01/logic", params={"key": new}).status_code == 404 def test_단가_자동_요소_찾기_저장전_시험계산(client: TestClient) -> None: one = _get(client, "/api/m01/logic", book=LOGIC_BOOK, key=LOGIC_KEY) labor = one["prices"][key_of("LB:1016")] assert labor["이름"] == "화약취급공" and labor["값"] > 0 found = _get(client, "/api/m01/elements", group="인력", q="보통인부") assert key_of("LB:1002") in [x["ref"] for x in found["items"]] tables = _get(client, "/api/m01/elements", group="소요량", q="암발파", limit=5) assert tables["items"] and all(x["값칸"] for x in tables["items"]) raw = (store.FOLDER / one["file"]).read_bytes() before = _calc(client)["sums"]["계"] half = {**one["logic"], "호표": one["logic"]["호표"][:1]} # 보통인부 줄 뺌 res = client.post( "/api/m01/calc", json={"book": LOGIC_BOOK, "key": LOGIC_KEY, "inputs": LOGIC_INPUTS, "row": half}, ) lines = res.json()["lines"] assert [x["이름"] for x in lines] == ["화약취급공"] and res.json()["sums"]["계"] < before new = {**half, "키": "", "원문번호": "새 로직"} body = { "book": LOGIC_BOOK, "key": "", "inputs": LOGIC_INPUTS, "row": new, "file": one["file"], } assert client.post("/api/m01/calc", json=body).json()["ok"] is True assert (store.FOLDER / one["file"]).read_bytes() == raw # 시험 계산은 안 씀 def test_쓰기_모양은_읽은_값과_같음() -> None: for name in ( "소요량_건설품셈_13장_플랜트설비공사.json", "로직_건설품셈_02장_하천공사.json", LABOR, ): data = json.loads((REAL / name).read_text(encoding="utf-8"), parse_float=Decimal) text = store.dump(data) assert json.loads(text, parse_float=Decimal) == data assert "\r" not in text and text.endswith("}\n") def test_시험계산_글로_온_수도_받음(client: TestClient) -> None: def calc(**inputs): body = {"key": key_of("GF:13-4-1 메쌓기"), "inputs": inputs} return client.post("/api/m01/calc", json=body).json() base = {"돌": "견치돌", "쌓기": "골쌓기", "높이": 1, "초과증가율": 80} assert calc(뒷길이=35, **base)["sums"]["계"] == pytest.approx(203281.2) as_text = calc(**{**base, "뒷길이": "35", "높이": "1"}) assert as_text["ok"] is True and as_text["sums"]["계"] == pytest.approx(203281.2) assert "고르기 밖" in calc(뒷길이="36", **base)["reason"] def test_저장은_고친_줄만_바꿈(client: TestClient) -> None: old = (store.FOLDER / LABOR).read_text(encoding="utf-8").split("\n") row, version = _labor_row(client, "1016") change = {"op": "edit", "key": row["키"], "row": {**row, "값": row["값"] + 1}} body = {"files": [{"file": LABOR, "version": version, "changes": [change]}]} assert client.post("/api/m01/save", json=body).status_code == 200 new = (store.FOLDER / LABOR).read_text(encoding="utf-8").split("\n") assert len(new) == len(old) and sum(a != b for a, b in zip(old, new)) == 1 one = _get(client, "/api/m01/logic", book=LOGIC_BOOK, key=LOGIC_KEY) path = store.FOLDER / one["file"] old = path.read_text(encoding="utf-8") edited = {**one["logic"], "비고": "고침"} change = {"op": "edit", "key": LOGIC_KEY, "row": edited} body = {"files": [{"file": one["file"], "version": one["version"], "changes": [change]}]} assert client.post("/api/m01/save", json=body).status_code == 200 new = path.read_text(encoding="utf-8") at = old.index(f'"키": "{LOGIC_KEY}"') head = old[: old.rfind("\n", 0, at)] # 고친 요소 앞은 글자 그대로 assert new.startswith(head) and "\r" not in new tail = old[old.index('"키"', at + 1) :] # 다음 요소부터도 그대로 assert new.endswith(tail) def test_main_은_시스템관리자만() -> None: main = (Path(__file__).resolve().parents[2] / "main.py").read_text(encoding="utf-8") assert "app.include_router(m01_master_data_router, dependencies=system_admin_only)" in main def test_고르기_나라장터_시중물가_직종(client: TestClient) -> None: nara = _get(client, "/api/m01/pick", kind="procure", q="육각볼트 M6*20") assert nara["total"] >= 1 and nara["items"][0]["ref"].startswith("나라장터:") market = _get(client, "/api/m01/pick", kind="price", q="H형강 관급", limit=5) assert market["items"] and market["items"][0]["관급"] and market["items"][0]["값"] > 0 job = _get(client, "/api/m01/pick", kind="job", q="보통인부") assert job["items"][0]["값"] > 0 and job["items"][0]["ref"] == key_of("LB:1002") assert _get(client, "/api/m01/pick", kind="job", q="CAD설계사")["items"][0]["ref"] == key_of( "LB:1" ) assert all(i["값"] is not None for i in _get(client, "/api/m01/pick", kind="job")["items"]) assert _get(client, "/api/m01/pick", kind="job", q="기술사(건설)")["items"][0]["ref"] == key_of( "LB:기술사(건설)" ) assert ( not _get(client, "/api/m01/pick", kind="job", q="중급기술자(기계")["items"][0].get("조사") == "준용대상" ) def test_못_이은_줄만_거름(client: TestClient) -> None: every = _get(client, "/api/m01/rows", file="재료_품셈재료.json", size=500) cut = _get(client, "/api/m01/rows", file="재료_품셈재료.json", size=500, unlinked=1) assert 0 < cut["total"] < every["total"] and not any(r["연결"] for r in cut["rows"]) def test_품셈재료_단가는_연결의_낮은_값과_출처(client: TestClient) -> None: key = key_of("GC:기계설비 13-2-4 강판 전기아크용접 전기아크용접(V형)") one = _get(client, "/api/m01/logic", key=key) power = one["prices"][key_of("MP:전력")] assert power["값"] > 0 and power["출처"].startswith(key_of("MT:M0005250064") + ".")