"""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_SUB = "건설품셈 공통" # 구분 = 원문 + 부문 LOGIC_NUM = "3-3-1" # 원문번호는 절 번호만 LOGIC_NAME = "암발파(미진동굴착 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 = "GC000091" # 공통 3-3-1 암발파(미진동굴착 TYPE-Ⅰ) @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={"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_NAME) assert one["원문번호"] == LOGIC_NUM assert one["count"] >= 1 and "줄" not in one and one["file"] == file 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: """파일을 가로지른 한 목록 — 구분 · 상세구분 · 찾기 · 쪽 나눔.""" every = _get(client, "/api/m01/tables", group="소요량", size=1) kind = _get(client, "/api/m01/tables", group="소요량", sub=LOGIC_SUB, size=500) assert 0 < kind["total"] < every["total"] assert {t["구분"] for t in kind["tables"]} == {LOGIC_SUB} assert len({t["file"] for t in kind["tables"]}) > 1 # 여러 장 파일이 한 목록에 one = _get( client, "/api/m01/tables", group="소요량", sub=LOGIC_SUB, detail="03장 토공사", size=500 ) assert 0 < one["total"] < kind["total"] assert {t["상세구분"] for t in one["tables"]} == {"03장 토공사"} page = _get(client, "/api/m01/tables", group="소요량", size=5, page=2) assert page["page"] == 2 and len(page["tables"]) == 5 and page["total"] == every["total"] def test_갈래_목록은_서버가_줌(client: TestClient) -> None: """소요량 · 계수 · 로직 컨테이너의 구분 · 상세구분 — 장 차례대로.""" for group in ("소요량", "계수", "로직"): got = _get(client, "/api/m01/subs", group=group) names = [s["name"] for s in got["subs"]] assert got["slot"] == "구분" and names[0] == "산림품셈" assert LOGIC_SUB in names kind = next(s for s in got["subs"] if s["name"] == LOGIC_SUB) details = kind["details"] assert kind["book"] == "건설품셈" and all(d[:2].isdigit() and d[2] == "장" for d in details) assert details == sorted(details) # 장 차례대로 def test_로직_목록_하나_시험계산(client: TestClient) -> None: detail = "03장 토공사" listed = _get(client, "/api/m01/logics", sub=LOGIC_SUB, detail=detail, q="암발파")["logics"] assert LOGIC_KEY in [x["키"] for x in listed] assert all((x["구분"], x["상세구분"]) == (LOGIC_SUB, detail) for x in listed) assert not any(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", 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", 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", 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={"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 = { "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("GF000219"), "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", 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: assert client.get("/api/m01/pick", params={"kind": "procure"}).status_code == 404 market = _get(client, "/api/m01/pick", kind="price", q="H형강 관급", limit=50) # 조달청 값이 아직 안 이어진 줄도 섞여 있음 — 이어진 줄이 하나라도 있으면 됨 assert any(i["관급"] and i["값"] > 0 for i in market["items"]) 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: """구분 목록 API — 파일 머리에 등록된 갈래 그대로(화면이 목록을 안 품음).""" got = _get(client, "/api/m01/subs", file=LABOR) names = [s["name"] for s in got["subs"]] assert got["slot"] == "구분" and names[0] == "건설업" and names[-1] == "미확보" survey = next(s for s in got["subs"] if s["name"] == "측량") assert survey["book"] == "측량노임" and "기능계 도화" in survey["details"] machine = _get(client, "/api/m01/subs", file="기계.json") assert machine["slot"] == "세부분류" and [s["name"] for s in machine["subs"]] == [ "건설품셈", "산림품셈", ] assert _get(client, "/api/m01/subs", file="재료_자재품목.json")["subs"] == [] 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("GC000893") one = _get(client, "/api/m01/logic", key=key) power = one["prices"][key_of("MP:전력")] assert power["값"] > 0 and power["출처"] == key_of("MT:M0005250064") + ".물가자료" def test_하위_거름과_참조_이름(client: TestClient) -> None: every = _get(client, "/api/m01/rows", file=LABOR, size=500) survey = _get(client, "/api/m01/rows", file=LABOR, size=500, sub="측량") assert 0 < survey["total"] < every["total"] and {r["구분"] for r in survey["rows"]} == {"측량"} field = _get(client, "/api/m01/rows", file=LABOR, size=500, sub="측량", detail="기능계 도화") assert 0 < field["total"] < survey["total"] assert {r["상세구분"] for r in field["rows"]} == {"기능계 도화"} machine = _get(client, "/api/m01/rows", file="기계.json", size=500, sub="산림품셈") assert 0 < machine["total"] < 100 and {r["세부분류"] for r in machine["rows"]} == {"산림품셈"} linked = _get(client, "/api/m01/rows", file="재료_품셈재료.json", size=500) keyed = [r for r in linked["rows"] if isinstance(r["연결"], str)] assert keyed and all(linked["refs"][r["연결"]] for r in keyed) def test_자재품목_저장은_띄어쓰기를_안_바꿈(client: TestClient) -> None: market = "재료_자재품목.json" old = (store.FOLDER / market).read_text(encoding="utf-8").split("\n") rows = _get(client, "/api/m01/rows", file=market, size=1) row = rows["rows"][0] row["물가자료"] = 1 change = {"op": "edit", "key": row["키"], "row": row} body = {"files": [{"file": market, "version": rows["version"], "changes": [change]}]} assert client.post("/api/m01/save", json=body).status_code == 200 new = (store.FOLDER / market).read_text(encoding="utf-8").split("\n") diff = [(a, b) for a, b in zip(old, new) if a != b] assert len(new) == len(old) and len(diff) == 1 assert diff[0][1].replace('"물가자료": 1,', '"물가자료": X,').count("{ ") == diff[0][0].count( "{ " )