From e1c71c3e8b4b37efc68933f54d9ce638aa653082 Mon Sep 17 00:00:00 2001 From: umsangdon Date: Fri, 18 Sep 2026 23:47:33 +0900 Subject: [PATCH] =?UTF-8?q?feat(Z01):=20=EA=B3=B5=EC=A2=85=20=ED=8E=B8?= =?UTF-8?q?=EC=A7=91=20=EC=A0=80=EC=9E=A5=EC=86=8C=C2=B7API=20=E2=80=94=20?= =?UTF-8?q?=EB=AA=A9=EB=A1=9D=C2=B7=EC=83=81=EC=84=B8=C2=B7=EC=A0=84?= =?UTF-8?q?=EC=B2=B4=20=EC=A4=84=20=EC=A0=80=EC=9E=A5=C2=B7=EC=B4=88?= =?UTF-8?q?=EA=B8=B0=EA=B0=92=EC=9C=BC=EB=A1=9C=20(PLAN=201-3)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - GET /work-items(상태별 개수) · GET/PUT /work-items/{key} · POST …/reset 신설, 시스템관리자 전용 묶음에 등록 - 저장은 한 트랜잭션 — 낡은 revision 409 · 없는 자원 열쇠·남의 줄 id 422 · 실패 시 전부 되돌림 - 빠진 줄은 초기값 줄이면 숨김(초기값으로 때 살림) · 관리자 추가 줄이면 지움 - 검사·줄 비교·되돌리기 규칙은 WorkItems.py 순수 함수 · SQL 은 Repository_WorkItems.py - 옛 기초단가 표의 공종 축 보기(JSON 읽기)는 그대로 둠 — BasePrices.py(sub_laptop_2 몫)가 동기로 부름 - 시험 15개(test_z01_work_item_edit.py) — DB 없이 가짜 커넥션으로 409·422·롤백 확인 Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_016WRFJmmhPi4exAzHgPZHMT --- .../Z01_MasterData_Repository_WorkItems.py | 181 ++++++++++++++ .../Z01_MasterData_Router_WorkItems.py | 152 +++++++++++ Z01_MasterData/Z01_MasterData_WorkItems.py | 132 ++++++++++ main.py | 2 + resources/tester/test_z01_work_item_edit.py | 235 ++++++++++++++++++ 5 files changed, 702 insertions(+) create mode 100644 Z01_MasterData/Z01_MasterData_Repository_WorkItems.py create mode 100644 Z01_MasterData/Z01_MasterData_Router_WorkItems.py create mode 100644 resources/tester/test_z01_work_item_edit.py diff --git a/Z01_MasterData/Z01_MasterData_Repository_WorkItems.py b/Z01_MasterData/Z01_MasterData_Repository_WorkItems.py new file mode 100644 index 00000000..11f0658a --- /dev/null +++ b/Z01_MasterData/Z01_MasterData_Repository_WorkItems.py @@ -0,0 +1,181 @@ +"""Z01 공종 저장소 — `master_work_item`·`master_work_item_row`·`master_revision` 읽기·쓰기(SQL 만). + +무엇을 쓸지는 `Z01_MasterData_WorkItems` 순수 함수(`plan_rows`·`reset_plan`)가 정함 · +여기는 받은 대로 씀. +표 모양 = `db_management/020_master_cost.sql`(PLAN 1-0 계약). 트랜잭션은 부르는 쪽(라우터)이 엶. +""" + +from __future__ import annotations + +import json +from typing import Any + +import aiomysql + +HEAD_COLUMNS = ( + "id, axis, work_item_key, code, number, name, path_name, parent_key, parent_mode, " + "basis_quantity, basis_unit, status, notes, source, seed, edition, sort_order, " + "updated_by, updated_at" +) +ROW_COLUMNS = ( + "id, variant, resource_kind, resource_key, resource_name, resource_spec, amount, " + "amount_unit, alternative_amount, group_ratio_pct, raw_cell, pum_table_id, seed, " + "sort_order, is_deleted" +) +_JSON = ("notes", "source", "seed") + + +def _parsed(row: dict[str, Any]) -> dict[str, Any]: + """JSON 칸은 MariaDB 가 글자로 돌려줌 — 풀어서 씀.""" + for key in _JSON: + if isinstance(row.get(key), (str, bytes)): + row[key] = json.loads(row[key]) + return row + + +def _like(q: str) -> str: + return "%" + q.replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_") + "%" + + +async def list_items( + connection: aiomysql.Connection, *, axis: str, status: str, q: str, page: int, size: int +) -> dict[str, Any]: + """목록 한 쪽 + 상태별 개수(상태 거르기 전 · 탭 숫자용). total 은 개수에서 셈(질의 하나 덜).""" + where, params = ["1"], [] + if axis: + where.append("w.axis = %s") + params.append(axis) + if q: + where.append("(w.work_item_key LIKE %s OR w.number LIKE %s OR w.path_name LIKE %s)") + params += [_like(q)] * 3 + picked, picked_params = list(where), list(params) + if status: + picked.append("w.status = %s") + picked_params.append(status) + async with connection.cursor(aiomysql.DictCursor) as cursor: + await cursor.execute( + "SELECT w.status, COUNT(*) AS n FROM master_work_item w " + f"WHERE {' AND '.join(where)} GROUP BY w.status", + params, + ) + counts = {row["status"]: int(row["n"]) for row in await cursor.fetchall()} + await cursor.execute( + "SELECT w.work_item_key, w.axis, w.number, w.name, w.path_name, w.status, " + "(SELECT COUNT(*) FROM master_work_item_row r " + " WHERE r.work_item_id = w.id AND r.is_deleted = 0) AS row_count " + f"FROM master_work_item w WHERE {' AND '.join(picked)} " + "ORDER BY w.axis, w.sort_order, w.id LIMIT %s OFFSET %s", + [*picked_params, size, (page - 1) * size], + ) + items = list(await cursor.fetchall()) + total = counts.get(status, 0) if status else sum(counts.values()) + return {"items": items, "total": total, "counts": counts} + + +async def fetch_item( + connection: aiomysql.Connection, key: str, *, lock: bool = False +) -> dict[str, Any] | None: + async with connection.cursor(aiomysql.DictCursor) as cursor: + await cursor.execute( + f"SELECT {HEAD_COLUMNS} FROM master_work_item WHERE work_item_key = %s" + + (" FOR UPDATE" if lock else ""), + (key,), + ) + row = await cursor.fetchone() + return _parsed(row) if row else None + + +async def fetch_rows( + connection: aiomysql.Connection, item_id: int, *, with_deleted: bool = False +) -> list[dict[str, Any]]: + """구성 줄 — 기본은 살아 있는 줄만 · [초기값으로] 는 숨긴 줄까지.""" + async with connection.cursor(aiomysql.DictCursor) as cursor: + await cursor.execute( + f"SELECT {ROW_COLUMNS} FROM master_work_item_row WHERE work_item_id = %s" + + ("" if with_deleted else " AND is_deleted = 0") + + " ORDER BY sort_order, id", + (item_id,), + ) + return [_parsed(row) for row in await cursor.fetchall()] + + +async def revision(connection: aiomysql.Connection, scope: str, *, lock: bool = False) -> int: + """지금 revision(없으면 0). lock 이면 줄을 만들어 잡음 — 같은 공종 저장이 줄 서게.""" + async with connection.cursor() as cursor: + if lock: + await cursor.execute( + "INSERT IGNORE INTO master_revision (scope, revision) VALUES (%s, 0)", (scope,) + ) + await cursor.execute( + "SELECT revision FROM master_revision WHERE scope = %s" + + (" FOR UPDATE" if lock else ""), + (scope,), + ) + row = await cursor.fetchone() + return int(row[0]) if row else 0 + + +async def bump_revision(connection: aiomysql.Connection, scope: str) -> None: + async with connection.cursor() as cursor: + await cursor.execute( + "INSERT INTO master_revision (scope, revision) VALUES (%s, 1) " + "ON DUPLICATE KEY UPDATE revision = revision + 1", + (scope,), + ) + + +async def known_resource_keys(connection: aiomysql.Connection, keys: set[str]) -> set[str]: + """그중 기초단가(`master_base_price.row_key`)에 실재하는 열쇠.""" + if not keys: + return set() + marks = ", ".join(["%s"] * len(keys)) + async with connection.cursor() as cursor: + await cursor.execute( + "SELECT DISTINCT row_key FROM master_base_price " + f"WHERE is_deleted = 0 AND row_key IN ({marks})", + list(keys), + ) + return {row[0] for row in await cursor.fetchall()} + + +async def write( + connection: aiomysql.Connection, + item_id: int, + head: dict[str, Any], + plan: dict[str, list[Any]], + by: Any, +) -> None: + """머리 칸 + 줄 쓰기(`plan_rows`·`reset_plan` 모양). + + 칸 이름은 순수 함수의 고정 목록에서만 옴(요청 글자가 SQL 에 안 들어감). + """ + async with connection.cursor() as cursor: + sets = "".join(f"{field} = %s, " for field in head) + await cursor.execute( + f"UPDATE master_work_item SET {sets}updated_by = %s, updated_at = NOW() WHERE id = %s", + [*head.values(), by, item_id], + ) + for row in plan["updates"]: # 줄마다 칸이 다를 수 있음(seed 에 있는 칸만) — 한 공종 수십 줄 + values = {k: v for k, v in row.items() if k != "id"} + sets = "".join(f"{field} = %s, " for field in values) + await cursor.execute( + f"UPDATE master_work_item_row SET {sets}is_deleted = 0 " + "WHERE id = %s AND work_item_id = %s", + [*values.values(), row["id"], item_id], + ) + if plan["inserts"]: + fields = list(plan["inserts"][0]) + await cursor.executemany( + f"INSERT INTO master_work_item_row (work_item_id, {', '.join(fields)}) " + f"VALUES (%s, {', '.join(['%s'] * len(fields))})", + [[item_id, *(row[f] for f in fields)] for row in plan["inserts"]], + ) + for ids, sql in ( + (plan["hide"], "UPDATE master_work_item_row SET is_deleted = 1"), + (plan["drop"], "DELETE FROM master_work_item_row"), + ): + if ids: + await cursor.execute( + f"{sql} WHERE work_item_id = %s AND id IN ({', '.join(['%s'] * len(ids))})", + [item_id, *ids], + ) diff --git a/Z01_MasterData/Z01_MasterData_Router_WorkItems.py b/Z01_MasterData/Z01_MasterData_Router_WorkItems.py new file mode 100644 index 00000000..dfbac857 --- /dev/null +++ b/Z01_MasterData/Z01_MasterData_Router_WorkItems.py @@ -0,0 +1,152 @@ +"""Z01 공종 편집 라우터 — 공종 머리·구성 줄을 관리자가 고치고 메움(2026-09-18 PLAN 1-0 계약). + + GET /api/master-data/work-items?axis&status&q&page&size 목록 + 상태별 개수 + GET /api/master-data/work-items/{key} 머리(+revision)·구성 줄·원문 표·[주] + PUT /api/master-data/work-items/{key} 전체 줄 보내기 + → 409(낡은 revision) · 422(검사) + POST /api/master-data/work-items/{key}/reset 머리·줄을 초기값으로(추가 줄 삭제) +쓰기는 한 트랜잭션(전부 아니면 전무) · 응답은 새 GET 모양 그대로(화면이 다시 안 불러도 됨). +⚠ 권한은 등록하는 쪽(`main.py`)이 시스템 관리자 전용 묶음으로 붙임. +""" + +from __future__ import annotations + +from decimal import Decimal +from typing import Any, Literal + +from fastapi import APIRouter, Depends, HTTPException +from pydantic import BaseModel, ConfigDict, Field + +from common_util.common_util_auth import verify_session +from config.config_db import get_db_pool +from Z01_MasterData import Z01_MasterData_Repository_WorkItems as repo +from Z01_MasterData import Z01_MasterData_Tables as tables +from Z01_MasterData import Z01_MasterData_WorkItems as work_items + +router = APIRouter(prefix="/api/master-data", tags=["Z01 MasterData WorkItems"]) + +Status = Literal[work_items.STATUSES] # 튜플을 펼침 — 상태 이름은 한 곳(`STATUSES`) +MAX_ROWS = 500 + + +class _Strict(BaseModel): + model_config = ConfigDict(extra="forbid") + + +class ItemHead(_Strict): + name: str = Field(min_length=1) + basis_quantity: Decimal | None = Field(default=None, ge=0) + basis_unit: str | None = None + status: Status + + +class ItemRow(_Strict): + id: int | None = None # 있으면 수정 · 없으면 추가 + variant: str = "" + resource_kind: Literal["labor", "machine", "material", "etc"] + resource_key: str | None = Field(default=None, min_length=1) # 못 맞춘 줄은 null(빈 글자 금지) + resource_name: str = "" + resource_spec: str = "" + amount: Decimal | None = Field(default=None, ge=0) + amount_unit: str = "" + alternative_amount: Decimal | None = Field(default=None, ge=0) + group_ratio_pct: Decimal | None = Field(default=None, ge=0) + raw_cell: str | None = None + pum_table_id: str = "" + + +class WorkItemSave(_Strict): + base_revision: int + item: ItemHead + rows: list[ItemRow] = Field(max_length=MAX_ROWS) # 전체 줄 — 빠진 줄은 삭제 + + +@router.get("/work-items") +async def list_work_items( + axis: Literal["", "forest", "const"] = "", + status: Status | Literal[""] = "", + q: str = "", + page: int = 1, + size: int = tables.DEFAULT_PAGE_SIZE, +) -> dict[str, Any]: + page, size = max(page, 1), max(1, min(size, tables.MAX_PAGE_SIZE)) + async with get_db_pool().acquire() as connection: + found = await repo.list_items( + connection, axis=axis, status=status, q=q.strip(), page=page, size=size + ) + counts = {s: found["counts"].get(s, 0) for s in work_items.STATUSES} + return {"status": "success", **found, "counts": counts, "page": page, "size": size} + + +async def _detail(connection: Any, key: str) -> dict[str, Any]: + head = await repo.fetch_item(connection, key) + if head is None: + raise HTTPException(status_code=404, detail=f"없는 공종: {key}") + rows = await repo.fetch_rows(connection, head["id"]) + revision = await repo.revision(connection, work_items.revision_scope(key)) + return {"status": "success", **work_items.detail(head, rows, revision)} + + +@router.get("/work-items/{key}") +async def get_work_item(key: str) -> dict[str, Any]: + async with get_db_pool().acquire() as connection: + return await _detail(connection, key) + + +async def _locked(connection: Any, key: str) -> tuple[dict[str, Any], int]: + """공종 머리 + revision 을 잡음(같은 공종 저장이 줄 섬).""" + head = await repo.fetch_item(connection, key, lock=True) + if head is None: + raise HTTPException(status_code=404, detail=f"없는 공종: {key}") + return head, await repo.revision(connection, work_items.revision_scope(key), lock=True) + + +@router.put("/work-items/{key}") +async def put_work_item( + key: str, body: WorkItemSave, session: dict = Depends(verify_session) +) -> dict[str, Any]: + rows = [row.model_dump() for row in body.rows] + async with get_db_pool().acquire() as connection: + await connection.begin() + try: + head, current = await _locked(connection, key) + if current != body.base_revision: + raise HTTPException( + status_code=409, + detail="다른 곳에서 먼저 고쳤음 — 다시 불러올 것" + f"(지금 {current} · 보낸 {body.base_revision})", + ) + existing = await repo.fetch_rows(connection, head["id"]) + known = await repo.known_resource_keys( + connection, {row["resource_key"] for row in rows if row["resource_key"]} + ) + errors = work_items.check_rows(rows, {row["id"] for row in existing}, known) + if errors: + raise HTTPException(status_code=422, detail="\n".join(errors)) + plan = work_items.plan_rows(existing, rows) + await repo.write( + connection, head["id"], body.item.model_dump(), plan, session.get("user_id") + ) + await repo.bump_revision(connection, work_items.revision_scope(key)) + await connection.commit() + except BaseException: + await connection.rollback() + raise + return await _detail(connection, key) + + +@router.post("/work-items/{key}/reset") +async def reset_work_item(key: str, session: dict = Depends(verify_session)) -> dict[str, Any]: + async with get_db_pool().acquire() as connection: + await connection.begin() + try: + head, _current = await _locked(connection, key) + rows = await repo.fetch_rows(connection, head["id"], with_deleted=True) + plan = work_items.reset_plan(head, rows) + await repo.write(connection, head["id"], plan["head"], plan, session.get("user_id")) + await repo.bump_revision(connection, work_items.revision_scope(key)) + await connection.commit() + except BaseException: + await connection.rollback() + raise + return await _detail(connection, key) diff --git a/Z01_MasterData/Z01_MasterData_WorkItems.py b/Z01_MasterData/Z01_MasterData_WorkItems.py index 23c8d85b..c62360a5 100644 --- a/Z01_MasterData/Z01_MasterData_WorkItems.py +++ b/Z01_MasterData/Z01_MasterData_WorkItems.py @@ -7,11 +7,15 @@ — 빠지면 발파암(절취 0.1 + 깍기 0.9 + 집토 1) 금액이 틀어짐. `@id` = 불변 열쇠(FW-·CW-) — 없는 줄은 목차 코드로 서고 알림에 드러냄. ⚠ 읽기만 — 고칠 칸 없음(원문 값 풀기 8-7 뒤). 원본 `work_item_code`(FP-)는 B08·B09 가 씀 — 안 건드림. + +공종 편집(DB 정본 · 2026-09-18 PLAN 1-0) — 맨 아래 「공종 편집」 절. `master_work_item`·`_row` 를 +고치는 규칙만 순수 함수로(DB 없이 시험) · SQL 은 `Z01_MasterData_Repository_WorkItems`. """ from __future__ import annotations import json +from decimal import Decimal from functools import lru_cache from pathlib import Path from typing import Any @@ -292,3 +296,131 @@ def extra(kind: str) -> dict[str, Any]: "units": len(rows), }, } + + +# ── 공종 편집 — DB 표 `master_work_item`·`master_work_item_row` 를 고치는 규칙(PLAN 1-0 계약) ── + +STATUSES = ("auto_ok", "partial", "needs_input", "not_item", "reviewed") +#: 관리자가 고치는 칸 — [초기값으로] 도 이 칸만 seed 에서 되살림 +HEAD_FIELDS = ("name", "basis_quantity", "basis_unit", "status") +ROW_FIELDS = ( + "variant", + "resource_kind", + "resource_key", + "resource_name", + "resource_spec", + "amount", + "amount_unit", + "alternative_amount", + "group_ratio_pct", + "raw_cell", + "pum_table_id", +) +_HIDDEN = ("id", "work_item_id", "seed", "source", "notes", "is_deleted") + + +def revision_scope(key: str) -> str: + return f"work_item:{key}" + + +def _same(a: Any, b: Any) -> bool: + """칸 값이 같은가 — 숫자는 값으로(DB `1.500000` = 요청 `1.5` = seed `1.5`) · 빈 글자 = NULL.""" + numbers = (int, float, Decimal) + if isinstance(a, numbers) and isinstance(b, numbers): + return Decimal(str(a)) == Decimal(str(b)) + return (None if a == "" else a) == (None if b == "" else b) + + +def changed_columns(current: dict[str, Any], seed: dict[str, Any] | None, fields) -> list[str]: + """초기값과 다른 칸 — 관리자 추가 줄(seed 없음)은 빈 목록(`is_added` 로 드러남).""" + if seed is None: + return [] + return [f for f in fields if f in seed and not _same(current.get(f), seed[f])] + + +def detail(head: dict[str, Any], rows: list[dict[str, Any]], revision: int) -> dict[str, Any]: + """GET 응답 — 머리(+revision) · 구성 줄 · 원문 표 · [주]. seed 는 안 내보내고 고친 칸 이름만.""" + seed = head.get("seed") + return { + "item": { + **{k: v for k, v in head.items() if k not in _HIDDEN}, + "revision": revision, + "changed_columns": changed_columns(head, seed, HEAD_FIELDS), + }, + "rows": [ + { + **{k: v for k, v in row.items() if k not in _HIDDEN or k == "id"}, + "is_added": row.get("seed") is None, + "changed_columns": changed_columns(row, row.get("seed"), ROW_FIELDS), + } + for row in rows + ], + "source": head.get("source") or [], + "notes": head.get("notes") or [], + } + + +def check_rows( + rows: list[dict[str, Any]], existing_ids: set[int], known_keys: set[str] +) -> list[str]: + """[저장] 서버 검사 중 DB 와 맞춰 볼 것 — 모양(숫자 ≥0 · 종류 · 모르는 칸)은 요청 모델이 봄.""" + errors, seen = [], set() + for number, row in enumerate(rows, 1): + row_id = row.get("id") + if row_id is not None: + if row_id not in existing_ids: + errors.append(f"{number}번째 줄: 이 공종에 없는 줄(id {row_id})") + elif row_id in seen: + errors.append(f"{number}번째 줄: 같은 줄(id {row_id})이 두 번 옴") + seen.add(row_id) + key = row.get("resource_key") + if key is not None and key not in known_keys: + errors.append(f"{number}번째 줄: 기초단가에 없는 자원 열쇠 {key}") + return errors + + +def plan_rows( + existing: list[dict[str, Any]], incoming: list[dict[str, Any]] +) -> dict[str, list[Any]]: + """전체 줄 보내기 → 쓸 것. existing = 지금 살아 있는 줄 · incoming 순서 = sort_order. + + id 있음 → 수정(바뀐 줄만) · id 없음 → 추가 · 빠진 줄 → 초기값 줄은 숨김(`is_deleted` · + [초기값으로] 가 살림) · 관리자 추가 줄은 지움. + """ + by_id = {row["id"]: row for row in existing} + updates, inserts = [], [] + for order, row in enumerate(incoming): + values = {f: row.get(f) for f in ROW_FIELDS} | {"sort_order": order} + old = by_id.get(row.get("id")) + if old is None: + inserts.append(values) + elif any(not _same(old.get(f), v) for f, v in values.items()): + updates.append({"id": old["id"], **values}) + kept = {row.get("id") for row in incoming} + gone = [row for row in existing if row["id"] not in kept] + return { + "updates": updates, + "inserts": inserts, + "hide": [row["id"] for row in gone if row.get("seed") is not None], + "drop": [row["id"] for row in gone if row.get("seed") is None], + } + + +def reset_plan(head: dict[str, Any], rows: list[dict[str, Any]]) -> dict[str, Any]: + """[초기값으로] — 머리·줄을 seed 로(seed 에 있는 칸만) · 숨긴 초기값 줄도 살림 · 추가 줄은 지움. + + rows = 숨긴 줄까지 전부. 쓰기 모양은 `plan_rows` 와 같음(+ `head`). + """ + seed = head.get("seed") or {} + restore = (*ROW_FIELDS, "sort_order") + return { + "head": {f: seed[f] for f in HEAD_FIELDS if f in seed}, + "updates": [ + {"id": row["id"], **{f: row["seed"][f] for f in restore if f in row["seed"]}} + for row in rows + if row.get("seed") is not None + ], + "inserts": [], + "hide": [], + "drop": [row["id"] for row in rows if row.get("seed") is None], + } diff --git a/main.py b/main.py index 85f2344a..3d48f738 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 Z01_MasterData.Z01_MasterData_Router_WorkItems import router as z01_work_items_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(z01_work_items_router, dependencies=system_admin_only) # 개발 전용 잠금 해제 — 다른 라우터와 **같은 보호**를 받는다(로그인·회사·프로젝트 접근). # 그 위에 서버가 환경까지 한 번 더 본다. app.include_router(dev_unlock_router, dependencies=protected_with_company) diff --git a/resources/tester/test_z01_work_item_edit.py b/resources/tester/test_z01_work_item_edit.py new file mode 100644 index 00000000..040b3475 --- /dev/null +++ b/resources/tester/test_z01_work_item_edit.py @@ -0,0 +1,235 @@ +"""Z01 공종 편집 — 규칙(순수 함수) · 요청 모델 · 저장 흐름(409·422·전부 아니면 전무) · 등록. + +(2026-09-18 브레인 PLAN 1-0 계약 · sub_laptop_3) DB 없이 — 저장 흐름은 가짜 커넥션·저장소로 훑음. +""" + +from __future__ import annotations + +import importlib +from contextlib import asynccontextmanager +from decimal import Decimal + +import pytest +from fastapi import FastAPI +from fastapi.routing import APIRoute +from fastapi.testclient import TestClient +from pydantic import ValidationError + +from common_util.common_util_auth import require_system_admin, verify_session +from Z01_MasterData import Z01_MasterData_Router_WorkItems as router_module +from Z01_MasterData import Z01_MasterData_WorkItems as work_items + +SEED_ROW = {"resource_kind": "labor", "resource_key": "L-001", "amount": 1.5, "sort_order": 0} + + +def _row(row_id, seed=SEED_ROW, **values): + base = {"id": row_id, "variant": "", "resource_kind": "labor", "resource_key": "L-001"} + return {**base, "amount": Decimal("1.500000"), "sort_order": 0, "seed": seed, **values} + + +# ── 규칙 ── + + +def test_고친_칸은_초기값과_값으로_비교함() -> None: + assert work_items.changed_columns(_row(1), SEED_ROW, work_items.ROW_FIELDS) == [] + changed = _row(1, amount=Decimal("2")) + assert work_items.changed_columns(changed, SEED_ROW, work_items.ROW_FIELDS) == ["amount"] + assert work_items.changed_columns(_row(1, seed=None), None, work_items.ROW_FIELDS) == [] + + +def test_검사는_남의_줄_겹친_줄_없는_자원을_막음() -> None: + rows = [ + {"id": 1, "resource_key": "L-001"}, + {"id": 1, "resource_key": None}, + {"id": 99, "resource_key": None}, + {"id": None, "resource_key": "없는열쇠"}, + ] + errors = work_items.check_rows(rows, existing_ids={1, 2}, known_keys={"L-001"}) + assert len(errors) == 3 + assert "두 번" in errors[0] and "id 99" in errors[1] and "없는열쇠" in errors[2] + assert work_items.check_rows(rows[:1], {1}, {"L-001"}) == [] + + +def test_전체_줄_보내기는_바뀐_줄만_쓰고_빠진_줄은_숨기거나_지움() -> None: + existing = [_row(1), _row(2, sort_order=1), _row(3, seed=None, sort_order=2)] + same = {f: existing[0].get(f) for f in work_items.ROW_FIELDS} | {"id": 1, "amount": 1.5} + new = {"id": None, "resource_kind": "material", "resource_key": None, "amount": None} + plan = work_items.plan_rows(existing, [same, new]) + assert plan["updates"] == [] # 1.500000 = 1.5 · 순서도 그대로 + assert plan["inserts"][0]["resource_kind"] == "material" + assert plan["inserts"][0]["sort_order"] == 1 + assert plan["hide"] == [2] # 초기값 줄 — 숨김(되살릴 수 있게) + assert plan["drop"] == [3] # 관리자 추가 줄 — 지움 + + moved = work_items.plan_rows(existing[:2], [{**same, "id": 2}, same]) + assert [u["id"] for u in moved["updates"]] == [2, 1] # 순서만 바뀌어도 sort_order 를 씀 + + +def test_초기값으로는_seed_칸만_살리고_추가_줄은_지움() -> None: + head = {"id": 7, "name": "고친 이름", "seed": {"name": "원 이름", "status": "partial"}} + rows = [_row(1, amount=Decimal("9")), _row(2, is_deleted=1), _row(3, seed=None)] + plan = work_items.reset_plan(head, rows) + assert plan["head"] == {"name": "원 이름", "status": "partial"} + assert [u["id"] for u in plan["updates"]] == [1, 2] + assert plan["updates"][0]["amount"] == 1.5 + assert plan["drop"] == [3] and plan["inserts"] == [] and plan["hide"] == [] + + +def test_상세는_seed_를_안_내보내고_고친_칸만_알림() -> None: + head = {"id": 7, "work_item_key": "FW-00001", "name": "새", "status": "partial"} + head |= {"seed": {"name": "옛"}, "source": [{"pum_table_id": "T1"}], "notes": ["[주] 1"]} + out = work_items.detail(head, [_row(1), _row(2, seed=None)], revision=4) + assert out["item"]["revision"] == 4 and out["item"]["changed_columns"] == ["name"] + assert "seed" not in out["item"] and "id" not in out["item"] + assert out["source"] == [{"pum_table_id": "T1"}] and out["notes"] == ["[주] 1"] + assert [r["is_added"] for r in out["rows"]] == [False, True] + assert out["rows"][0]["id"] == 1 and "seed" not in out["rows"][0] + + +# ── 요청 모델 ── + + +def _body(**row): + item = {"name": "공종", "status": "reviewed"} + return {"base_revision": 3, "item": item, "rows": [{"resource_kind": "labor", **row}]} + + +@pytest.mark.parametrize( + "row", + [{"amount": -1}, {"resource_key": ""}, {"resource_kind": "oil"}, {"unknown": 1}], +) +def test_요청_모델은_음수_빈열쇠_모르는_종류_모르는_칸을_막음(row) -> None: + with pytest.raises(ValidationError): + router_module.WorkItemSave.model_validate(_body(**row)) + + +# ── 저장 흐름(가짜 커넥션) ── + + +class _Connection: + def __init__(self) -> None: + self.log: list[str] = [] + + async def begin(self) -> None: + self.log.append("begin") + + async def commit(self) -> None: + self.log.append("commit") + + async def rollback(self) -> None: + self.log.append("rollback") + + +@pytest.fixture +def client(monkeypatch): + connection = _Connection() + + class _Pool: + @asynccontextmanager + async def acquire(self): + yield connection + + head = {"id": 7, "work_item_key": "FW-00001", "name": "공종", "status": "partial"} + head |= {"seed": {"name": "공종"}, "source": [], "notes": []} + + async def fetch_item(_c, key, lock=False): + return dict(head) if key == "FW-00001" else None + + async def fetch_rows(_c, _item_id, with_deleted=False): + return [_row(1)] + + async def revision(_c, _scope, lock=False): + return 3 + + async def known_resource_keys(_c, keys): + return keys & {"L-001"} + + async def write(_c, *args): + connection.log.append("write") + + async def bump_revision(_c, _scope): + connection.log.append("bump") + + async def list_items(_c, **picked): + connection.log.append(picked["status"]) + return {"items": [], "total": 2, "counts": {"partial": 2}} + + monkeypatch.setattr(router_module, "get_db_pool", lambda: _Pool()) + for name, fake in { + "list_items": list_items, + "fetch_item": fetch_item, + "fetch_rows": fetch_rows, + "revision": revision, + "known_resource_keys": known_resource_keys, + "write": write, + "bump_revision": bump_revision, + }.items(): + monkeypatch.setattr(router_module.repo, name, fake) + app = FastAPI() + app.include_router(router_module.router) + app.dependency_overrides[verify_session] = lambda: {"user_id": 42} + test_client = TestClient(app) + test_client.log = connection.log + return test_client + + +URL = "/api/master-data/work-items/FW-00001" + + +def test_낡은_revision_은_409_이고_아무것도_안_씀(client) -> None: + response = client.put(URL, json={**_body(id=1, resource_key="L-001"), "base_revision": 2}) + assert response.status_code == 409 + assert client.log == ["begin", "rollback"] + + +def test_없는_자원_열쇠는_422_이고_아무것도_안_씀(client) -> None: + response = client.put(URL, json=_body(resource_key="없는열쇠")) + assert response.status_code == 422 and "없는열쇠" in response.json()["detail"] + assert client.log == ["begin", "rollback"] + + +def test_맞는_저장은_쓰고_revision_올리고_새_상세를_줌(client) -> None: + response = client.put(URL, json=_body(id=1, resource_key="L-001", amount=2)) + assert response.status_code == 200, response.text + assert client.log == ["begin", "write", "bump", "commit"] + assert response.json()["status"] == "success" and response.json()["item"]["revision"] == 3 + + +def test_목록은_상태별_개수를_다_채우고_모르는_상태는_막음(client) -> None: + body = client.get("/api/master-data/work-items?status=").json() + assert body["counts"] == {s: 2 if s == "partial" else 0 for s in work_items.STATUSES} + assert client.get("/api/master-data/work-items?status=needs_input").status_code == 200 + assert client.get("/api/master-data/work-items?status=bogus").status_code == 422 + assert client.log == ["", "needs_input"] + + +def test_초기값으로와_없는_공종(client) -> None: + assert client.post(URL + "/reset").status_code == 200 + assert client.log == ["begin", "write", "bump", "commit"] + assert client.get("/api/master-data/work-items/FW-99999").status_code == 404 + + +# ── 등록 ── + + +def _calls(dependant) -> set: + found = set() + for sub in dependant.dependencies: + found.add(sub.call) + found |= _calls(sub) + return found + + +def test_공종_편집_경로는_시스템_관리자만() -> None: + app = importlib.import_module("main").app + routes = {(r.path, m) for r in app.routes if isinstance(r, APIRoute) for m in r.methods} + wanted = { + ("/api/master-data/work-items", "GET"), + ("/api/master-data/work-items/{key}", "GET"), + ("/api/master-data/work-items/{key}", "PUT"), + ("/api/master-data/work-items/{key}/reset", "POST"), + } + assert wanted <= routes + for route in app.routes: + if isinstance(route, APIRoute) and route.path.startswith("/api/master-data/work-items"): + assert {verify_session, require_system_admin} <= _calls(route.dependant)