From e11827f8f42e43e73340e0938deabaf3181eefa3 Mon Sep 17 00:00:00 2001 From: umsangdon Date: Sat, 19 Sep 2026 18:15:20 +0900 Subject: [PATCH] =?UTF-8?q?feat(M01):=20=EB=A7=88=EC=8A=A4=ED=84=B0=20?= =?UTF-8?q?=EB=8D=B0=EC=9D=B4=ED=84=B0=20=EA=B4=80=EB=A6=AC=EC=9E=90=20API?= =?UTF-8?q?=20=EC=84=9C=EB=B2=84=20=EC=AA=BD=20=E2=80=94=20=EC=9D=BD?= =?UTF-8?q?=EA=B8=B0=20=C2=B7=20=EC=8B=9C=ED=97=98=20=EA=B3=84=EC=82=B0=20?= =?UTF-8?q?=C2=B7=20=EC=A0=80=EC=9E=A5?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - M01_MasterData 라우터 + 파일 저장소 (/api/m01 · 시스템관리자 전용) - 저장 = 판본 대조(409) → 적용 → 틀·로직 변수 검사(새 걸림 422) → 씀 - check_master.py 에 load(folder) · check_saved · calc 드러냄 - 계약 문서 장 이름 보정(공통3장) - 시험 test_m01_api.py 8개 (임시 폴더 사본) Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01PAdA5ThqmtVSsbzjusk1cJ --- M01_MasterData/M01_MasterData_Router.py | 101 ++++++ M01_MasterData/M01_MasterData_Store.py | 300 ++++++++++++++++++ main.py | 2 + resources/master_data/_화면_계약.md | 2 +- resources/master_data/scripts/check_master.py | 28 +- resources/tester/test_m01_api.py | 176 ++++++++++ 6 files changed, 602 insertions(+), 7 deletions(-) create mode 100644 M01_MasterData/M01_MasterData_Router.py create mode 100644 M01_MasterData/M01_MasterData_Store.py create mode 100644 resources/tester/test_m01_api.py diff --git a/M01_MasterData/M01_MasterData_Router.py b/M01_MasterData/M01_MasterData_Router.py new file mode 100644 index 00000000..17ffe0b3 --- /dev/null +++ b/M01_MasterData/M01_MasterData_Router.py @@ -0,0 +1,101 @@ +"""M01 마스터 데이터 관리자 API — 계약 `resources/master_data/_화면_계약.md`. + +⚠ 권한은 등록하는 쪽(`main.py`)이 `system_admin_only` 로 붙임. +""" + +from __future__ import annotations + +from decimal import Decimal +from typing import Any, Literal + +from fastapi import APIRouter, HTTPException +from pydantic import BaseModel + +from M01_MasterData import M01_MasterData_Store as store + +router = APIRouter(prefix="/api/m01", tags=["M01 MasterData"]) + + +class CalcBody(BaseModel): + book: str + key: str + inputs: dict[str, Any] = {} + + +class Change(BaseModel): + op: Literal["edit", "add", "delete"] + key: str | None = None + row: dict[str, Any] | None = None + + +class FileChanges(BaseModel): + file: str + version: str + changes: list[Change] + + +class SaveBody(BaseModel): + files: list[FileChanges] + + +def _num(v): + """Decimal → JSON 수(그냥 두면 글자로 나감).""" + if isinstance(v, Decimal): + return int(v) if v == v.to_integral_value() else float(v) + if isinstance(v, list): + return [_num(x) for x in v] + if isinstance(v, dict): + return {k: _num(x) for k, x in v.items()} + return v + + +def _call(fn, *args): + try: + return _num(fn(*args)) + except store.StoreError as e: + raise HTTPException(status_code=e.status, detail=e.detail) from e + + +@router.get("/groups") +def get_groups() -> dict: + return {"groups": _call(store.groups)} + + +@router.get("/groups/{group}/files") +def get_files(group: str) -> dict: + return {"files": _call(store.files_of, group)} + + +@router.get("/rows") +def get_rows(file: str, page: int = 1, size: int = 50, q: str = "") -> dict: + return _call(store.rows, file, page, size, q) + + +@router.get("/tables") +def get_tables(file: str, q: str = "") -> dict: + return _call(store.tables, file, q) + + +@router.get("/table") +def get_table(file: str, key: str) -> dict: + return _call(store.table, file, key) + + +@router.get("/logics") +def get_logics(book: str = "", chapter: str = "", q: str = "", blocked: int | None = None) -> dict: + return {"logics": _call(store.logics, book, chapter, q, blocked)} + + +@router.get("/logic") +def get_logic(book: str, key: str) -> dict: + return _call(store.logic, book, key) + + +@router.post("/calc") +def post_calc(body: CalcBody) -> dict: + return _call(store.calc, body.book, body.key, body.inputs) + + +@router.post("/save") +def post_save(body: SaveBody) -> dict: + return {"files": _call(store.save, [f.model_dump() for f in body.files])} diff --git a/M01_MasterData/M01_MasterData_Store.py b/M01_MasterData/M01_MasterData_Store.py new file mode 100644 index 00000000..6540b563 --- /dev/null +++ b/M01_MasterData/M01_MasterData_Store.py @@ -0,0 +1,300 @@ +"""M01 마스터 데이터 파일 저장소 — `resources/master_data/` 첫 층 JSON 읽기·검사·쓰기. + +계약: `resources/master_data/_화면_계약.md` · 파일 모양: `_틀.md` +검사·계산 엔진: `scripts/check_master.py`. +""" + +from __future__ import annotations + +import hashlib +import json +import os +import re +import sys +import threading +from decimal import Decimal +from pathlib import Path + +SCRIPTS = Path(__file__).resolve().parents[1] / "resources/master_data/scripts" +sys.path.insert(0, str(SCRIPTS)) + +import check_master as cm # noqa: E402 + +FOLDER: Path = cm.MASTER # 시험은 사본 폴더로 바꿈 +_CHAPTER = re.compile(r"_([^_]*\d+장)\.json$") # 「공통3장」 · 「13장」 +# ponytail: 저장은 한 번에 하나(프로세스 안 잠금) · 서버를 여럿 띄우면 파일 잠금으로 +_LOCK = threading.Lock() + + +class StoreError(Exception): + def __init__(self, status: int, detail): + super().__init__(detail) + self.status, self.detail = status, detail + + +# ── 읽기 ────────────────────────────────────────────────────────────── +def _path(file: str) -> Path: + path = FOLDER / file + if "/" in file or "\\" in file or file.startswith("_") or not path.is_file(): + raise StoreError(404, f"없는 파일 「{file}」") + return path + + +def version_of(raw: bytes) -> str: + return hashlib.sha256(raw).hexdigest()[:16] + + +def read(file: str) -> tuple[dict, str]: + raw = _path(file).read_bytes() + return json.loads(raw, parse_float=Decimal, parse_int=Decimal), version_of(raw) + + +def items_key(data: dict) -> str: + return "표" if data.get("그룹") in cm.mf.TABLE_GROUPS else "줄" + + +def chapter_of(file: str) -> str: + m = _CHAPTER.search(file) + return m.group(1) if m else "" + + +def _hit(row: dict, q: str) -> bool: + q = q.lower() + return not q or q in str(row.get("열쇠", "")).lower() or q in str(row.get("이름", "")).lower() + + +def groups() -> list[dict]: + out = {g: {"group": g, "files": 0, "rows": 0} for g in cm.mf.GROUPS} + for data in cm.load(folder=FOLDER).values(): + slot = out.get(data.get("그룹")) + if slot: + slot["files"] += 1 + slot["rows"] += len(data.get(items_key(data)) or []) + return list(out.values()) + + +def files_of(group: str) -> list[dict]: + if group not in cm.mf.GROUPS: + raise StoreError(404, f"없는 그룹 「{group}」") + out = [] + for path in sorted(FOLDER.glob(f"{group}_*.json")): + data, version = read(path.name) + out.append( + { + "file": path.name, + "book": data.get("원문"), + "chapter": chapter_of(path.name), + "edition": data.get("판"), + "rows": len(data.get(items_key(data)) or []), + "version": version, + } + ) + return out + + +def rows(file: str, page: int, size: int, q: str) -> dict: + data, version = read(file) + hits = [r for r in data.get(items_key(data)) or [] if _hit(r, q)] + page, size = max(page, 1), min(max(size, 1), 500) + return { + "file": file, + "version": version, + "total": len(hits), + "page": page, + "size": size, + "rows": hits[(page - 1) * size : page * size], + } + + +def tables(file: str, q: str) -> dict: + data, version = read(file) + if items_key(data) != "표": + raise StoreError(400, f"표형 파일 아님 「{file}」") + heads = ("열쇠", "이름", "기준", "출처", "조건", "값칸") + return { + "file": file, + "version": version, + "tables": [ + {**{k: t.get(k) for k in heads}, "count": len(t.get("줄") or [])} + for t in data["표"] + if _hit(t, q) + ], + } + + +def _one(data: dict, key: str) -> dict: + for row in data.get(items_key(data)) or []: + if str(row.get("열쇠")) == key: + return row + raise StoreError(404, f"없는 열쇠 「{key}」") + + +def table(file: str, key: str) -> dict: + data, version = read(file) + return {"file": file, "version": version, "table": _one(data, key)} + + +def _logic_files(files: dict[str, dict]): + for name, data in files.items(): + if data.get("그룹") == "로직": + yield name, data + + +def logics(book: str, chapter: str, q: str, blocked: int | None) -> list[dict]: + files = cm.load(folder=FOLDER) + whole = cm.mf.Master(files) + out = [] + for name, data in _logic_files(files): + if (book and data.get("원문") != book) or (chapter and chapter_of(name) != chapter): + continue + for row in data.get("줄") or []: + if not _hit(row, q): + continue + reasons = cm.mf.check_logic(whole, data.get("원문"), row)[0] + if blocked is not None and bool(reasons) != bool(blocked): + continue + out.append( + { + "file": name, + "book": data.get("원문"), + "chapter": chapter_of(name), + **{k: row.get(k) for k in ("열쇠", "이름", "결과단위", "출처")}, + "blocked": bool(reasons), + "reasons": reasons, + } + ) + return out + + +def logic(book: str, key: str) -> dict: + files = cm.load(folder=FOLDER) + for name, data in _logic_files(files): + if data.get("원문") != book: + continue + for row in data.get("줄") or []: + if str(row.get("열쇠")) == key: + reasons = cm.mf.check_logic(cm.mf.Master(files), book, row)[0] + _, version = read(name) + return { + "file": name, + "version": version, + "logic": row, + "blocked": bool(reasons), + "reasons": reasons, + } + raise StoreError(404, f"없는 로직 「{book}:{key}」") + + +# ── 시험 계산 ────────────────────────────────────────────────────────── +def calc(book: str, key: str, inputs: dict) -> dict: + given = {k: _dec(v) for k, v in inputs.items()} + try: + result = cm.calc(cm.load(folder=FOLDER), f"{book}:{key}", given) + except (cm.mf.FormulaError, ArithmeticError, KeyError, TypeError, ValueError) as e: + return {"ok": False, "reason": str(e) or type(e).__name__} + if "결과" in result: + return {"ok": True, "result": result["결과"], "middle": result["중간"]} + return { + "ok": True, + "lines": result["줄"], + "sums": {k: result[k] for k in cm.mf.BUNDLE}, + "middle": result["중간"], + } + + +# ── 저장 ────────────────────────────────────────────────────────────── +def _dec(v): + """화면이 보낸 수 → Decimal(파일을 읽은 모양과 같게).""" + if isinstance(v, bool) or v is None: + return v + if isinstance(v, (int, float)): + return Decimal(str(v)) + if isinstance(v, list): + return [_dec(x) for x in v] + if isinstance(v, dict): + return {k: _dec(x) for k, x in v.items()} + return v + + +def _apply(data: dict, changes: list[dict], file: str) -> None: + items = data.setdefault(items_key(data), []) + for ch in changes: + op, key, row = ch.get("op"), ch.get("key"), _dec(ch.get("row")) + if op not in ("edit", "add", "delete"): + raise StoreError(400, f"모르는 op 「{op}」") + if op != "add": + at = next((i for i, r in enumerate(items) if str(r.get("열쇠")) == key), None) + if at is None: + raise StoreError(404, f"{file} · 없는 열쇠 「{key}」") + if op != "delete" and not (isinstance(row, dict) and "열쇠" in row): + raise StoreError(400, f"{file} · row 에 열쇠 없음") + new_key = str(row["열쇠"]) if op != "delete" else None + taken = {str(r.get("열쇠")) for i, r in enumerate(items) if op == "add" or i != at} + if new_key in taken: + raise StoreError(400, f"{file} · 열쇠 겹침 「{new_key}」") + if op == "add": + items.append(row) + elif op == "edit": + items[at] = row + else: + del items[at] + + +def save(batch: list[dict]) -> list[dict]: + with _LOCK: + stale = [p.get("file", "") for p in batch if read(p.get("file", ""))[1] != p.get("version")] + if stale: + raise StoreError(409, {"stale": stale}) + before, after = cm.load(folder=FOLDER), cm.load(folder=FOLDER) # after = 고칠 사본 + names = [part["file"] for part in batch] + for part in batch: + _apply(after[part["file"]], part.get("changes") or [], part["file"]) + old = set(cm.check_saved(before, names)) + new = [x for x in cm.check_saved(after, names) if x not in old] + if new: + raise StoreError(422, {"errors": new}) + out = [] + for name in dict.fromkeys(names): + text = dump(after[name]) + tmp = FOLDER / f".{name}.tmp" + tmp.write_text(text, encoding="utf-8", newline="\n") + os.replace(tmp, FOLDER / name) + out.append({"file": name, "version": version_of(text.encode("utf-8"))}) + return out + + +# ── 쓰기 모양 ───────────────────────────────────────────────────────── +def _atom(v) -> str: + if isinstance(v, Decimal): + return str(v) + return json.dumps(v, ensure_ascii=False) + + +def _flat(v) -> str: + if isinstance(v, dict): + return ( + "{ " + ", ".join(f"{_atom(k)}: {_flat(x)}" for k, x in v.items()) + " }" if v else "{}" + ) + if isinstance(v, list): + return "[" + ", ".join(_flat(x) for x in v) + "]" + return _atom(v) + + +def dump(v, indent: int = 0, width: int = 160) -> str: + """UTF-8 · 들여쓰기 2칸 · 한 줄에 들어가는 묶음은 한 줄로.""" + flat = _flat(v) + if indent and len(flat) + indent <= width or not isinstance(v, (dict, list)) or not v: + return flat + ("" if indent else "\n") + pad = " " * (indent + 2) + if isinstance(v, dict): + body = [f"{pad}{_atom(k)}: {dump(x, indent + 2, width)}" for k, x in v.items()] + text = "{\n" + ",\n".join(body) + "\n" + " " * indent + "}" + else: + text = ( + "[\n" + + ",\n".join(pad + dump(x, indent + 2, width) for x in v) + + "\n" + + " " * indent + + "]" + ) + return text + ("" if indent else "\n") diff --git a/main.py b/main.py index 85f2344a..916efeaa 100644 --- a/main.py +++ b/main.py @@ -76,6 +76,7 @@ from B09_Estimation.B09_Estimation_Router_MaterialPrices import ( router as b09_material_prices_router, ) from Z01_MasterData.Z01_MasterData_Router import router as z01_master_data_router +from M01_MasterData.M01_MasterData_Router import router as m01_master_data_router from common_util.common_util_audit import note_api_call, record_call_burst from common_util.common_util_auth import ( require_company, @@ -659,6 +660,7 @@ app.include_router(b09_material_prices_router, dependencies=protected_with_compa # Z01 마스터 데이터 — 회사·프로젝트가 아니라 시스템 관리자만 본다(2026-09-15 브레인 Z01). system_admin_only = [Depends(verify_session), Depends(require_system_admin)] app.include_router(z01_master_data_router, dependencies=system_admin_only) +app.include_router(m01_master_data_router, dependencies=system_admin_only) # 개발 전용 잠금 해제 — 다른 라우터와 **같은 보호**를 받는다(로그인·회사·프로젝트 접근). # 그 위에 서버가 환경까지 한 번 더 본다. app.include_router(dev_unlock_router, dependencies=protected_with_company) diff --git a/resources/master_data/_화면_계약.md b/resources/master_data/_화면_계약.md index cc4e524c..aa0d76eb 100644 --- a/resources/master_data/_화면_계약.md +++ b/resources/master_data/_화면_계약.md @@ -25,7 +25,7 @@ - `rows` 는 요소·로직 파일은 `줄`, 표형 파일(소요량·계수)은 `표` 를 줄로 봄. - `q` = 이름 찾기 — `열쇠`·`이름` 에 든 글(대소문자 무시). -- `chapter` = 파일 이름의 장(`13장` · `01장`) · 없으면 "". +- `chapter` = 파일 이름의 장(`공통3장` · `13장` · `01장`) · 없으면 "". - `blocked` = 로직 검사(`check_master.py 로직`)에 걸림 · `reasons` = 걸린 까닭 글. ## 3. 시험 계산 diff --git a/resources/master_data/scripts/check_master.py b/resources/master_data/scripts/check_master.py index 0bee4a7b..84224b7a 100644 --- a/resources/master_data/scripts/check_master.py +++ b/resources/master_data/scripts/check_master.py @@ -59,10 +59,10 @@ ROW_KEYS = { } -def load(names: list[str] | None = None) -> dict[str, dict]: - """첫 층 JSON 모두(밑줄 파일 빼고) — 수는 Decimal.""" +def load(names: list[str] | None = None, folder: Path = MASTER) -> dict[str, dict]: + """첫 층 JSON 모두(밑줄 파일 빼고) — 수는 Decimal. `folder` = 마스터 폴더(M01 시험은 사본).""" out = {} - for path in sorted(MASTER.glob("*.json")): + for path in sorted(Path(folder).glob("*.json")): if path.name.startswith("_") or ( names and path.name not in names and path.stem not in names ): @@ -73,8 +73,19 @@ def load(names: list[str] | None = None) -> dict[str, dict]: return out -def master() -> mf.Master: - return mf.Master(load()) +def master(folder: Path = MASTER) -> mf.Master: + return mf.Master(load(folder=folder)) + + +def check_saved(files: dict[str, dict], changed: list[str]) -> list[str]: + """저장 전 검사(M01) — 고친 파일의 틀 + 모든 로직의 변수. `files` = 폴더 전부(고친 뒤).""" + out = [x for name in changed for x in check_form(name, files[name])] + return out + check_logics(files, mf.Master(files)) + + +def calc(files: dict[str, dict], ref: str, given: dict) -> dict: + """시험 계산(M01) — `ref` = 「원문:로직 열쇠」 · 멈추면 FormulaError.""" + return mf.run(mf.Master(files), ref, given) # ── (1) 틀 ──────────────────────────────────────────────────────────── @@ -228,7 +239,12 @@ def check_body(files: dict[str, dict]) -> list[str]: if not text: out.append(f"{where} · 본문 절 못 찾음 「{table.get('출처')}」") continue - mine = _flat(table.get("줄")) | _flat(table.get("기준")) | _flat(table.get("주")) | _flat(table.get("값칸")) + mine = ( + _flat(table.get("줄")) + | _flat(table.get("기준")) + | _flat(table.get("주")) + | _flat(table.get("값칸")) + ) fake = mine - numbers_in(text) if fake: out.append(f"{where} · 허구 {len(fake)} {sorted(fake, key=Decimal)}") diff --git a/resources/tester/test_m01_api.py b/resources/tester/test_m01_api.py new file mode 100644 index 00000000..35c73fdc --- /dev/null +++ b/resources/tester/test_m01_api.py @@ -0,0 +1,176 @@ +"""M01 마스터 데이터 API — 계약 `resources/master_data/_화면_계약.md`. + +임시 폴더에 master_data 첫 층 JSON 사본을 두고 읽기 · 시험 계산 · 저장 · 409 · 422. +""" + +from __future__ import annotations + +import json +import shutil +from decimal import Decimal +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_KEY = ( + "건설품셈", + "3-3-1 암발파(미진동굴착 TYPE-Ⅰ)", +) # 화약취급공 1016 · 보통인부 1002 + + +@pytest.fixture +def client(tmp_path: Path, monkeypatch) -> TestClient: + for path in REAL.glob("*.json"): + if not path.name.startswith("_"): + 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": inputs} + ) + assert res.status_code == 200, res.text + return res.json() + + +def _labor_row(client: TestClient, key: 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["열쇠"] == key), 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 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 = "소요량_건설품셈_공통3장.json" + listed = _get(client, "/api/m01/tables", file=file, q="암발파") + one = next(t for t in listed["tables"] if t["열쇠"] == LOGIC_KEY) + assert one["count"] >= 1 and "줄" not in one + table = _get(client, "/api/m01/table", file=file, key=LOGIC_KEY)["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: + listed = _get(client, "/api/m01/logics", book=LOGIC_BOOK, chapter="공통3장", q="암발파")[ + "logics" + ] + assert LOGIC_KEY in [x["열쇠"] for x in listed] + assert all(x["chapter"] == "공통3장" and not x["blocked"] for x in listed) + assert _get(client, "/api/m01/logics", blocked=1)["logics"] == [] + one = _get(client, "/api/m01/logic", book=LOGIC_BOOK, key=LOGIC_KEY) + assert one["file"] == "로직_건설품셈_공통3장.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": "1016", "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: + _, version = _labor_row(client, "1016") + raw = (store.FOLDER / LABOR).read_bytes() + gone = {"file": LABOR, "version": version, "changes": [{"op": "delete", "key": "1016"}]} + res = client.post("/api/m01/save", json={"files": [gone]}) + assert res.status_code == 422 and any("1016" 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]}]} + assert client.post("/api/m01/save", json={"files": [dup]}).status_code == 400 + assert (store.FOLDER / LABOR).read_bytes() == raw + + +def test_더함_지움_로직_파일도_같은_길(client: TestClient) -> None: + one = _get(client, "/api/m01/logic", book=LOGIC_BOOK, key=LOGIC_KEY) + copy = {**one["logic"], "열쇠": LOGIC_KEY + " 사본"} + 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 + assert _get(client, "/api/m01/logic", book=LOGIC_BOOK, key=copy["열쇠"])["blocked"] is False + drop = { + "file": one["file"], + "version": res.json()["files"][0]["version"], + "changes": [{"op": "delete", "key": copy["열쇠"]}], + } + assert client.post("/api/m01/save", json={"files": [drop]}).status_code == 200 + assert ( + client.get("/api/m01/logic", params={"book": LOGIC_BOOK, "key": copy["열쇠"]}).status_code + == 404 + ) + + +def test_쓰기_모양은_읽은_값과_같음() -> None: + for name in ("소요량_건설품셈_기계설비13장.json", "로직_건설품셈_토목2장.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_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