From 6fe4056d9892ab37c7b64bbe86ad5533517bfb4a Mon Sep 17 00:00:00 2001 From: umsangdon Date: Fri, 18 Sep 2026 23:53:50 +0900 Subject: [PATCH] =?UTF-8?q?feat(Z01):=20=EA=B8=B0=EC=B4=88=EB=8B=A8?= =?UTF-8?q?=EA=B0=80=20=EC=A0=95=EB=B3=B8=EC=9D=84=20DB=20=EB=A1=9C=20?= =?UTF-8?q?=E2=80=94=20=EC=A0=80=EC=9E=A5=EC=86=8C=20=EC=8B=A0=EC=84=A4=20?= =?UTF-8?q?=C2=B7=20=EC=9D=BC=EA=B4=84=20=EC=A0=80=EC=9E=A5=C2=B7=EC=A4=84?= =?UTF-8?q?=20=EC=B6=94=EA=B0=80/=EC=82=AD=EC=A0=9C=C2=B7=EC=B4=88?= =?UTF-8?q?=EA=B8=B0=EA=B0=92=20=EB=90=98=EB=8F=8C=EB=A6=AC=EA=B8=B0=20API?= =?UTF-8?q?=20=C2=B7=20=EB=8D=AE=EA=B0=9C=20=EC=B8=B5=20=EC=82=AD=EC=A0=9C?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Z01_MasterData_Repository_BasePrices.py 신설 — master_base_price·master_revision Raw SQL(읽기·줄 수·판번호·저장·되돌리기) · 판번호 줄을 잠그고 다시 봄 · 틀리면 통째로 되돌림 - GET /base-prices/{kind} 가 DB 를 읽음 · 응답에 revision · 줄마다 changed_columns·is_added 추가 - PUT /base-prices/{kind} {base_revision, edits, added, deleted} — 판 다르면 409 · 하나라도 틀리면 아무것도 안 씀 · edits 500 상한 · 모르는 칸 거절 - POST /base-prices/{kind}/reset {row_keys} — 주입 줄은 data := seed(지운 줄도 되살림) · 추가 줄은 지움 - 기계 시간당 단가는 DB 노임·유가로 읽을 때마다 셈 · 제원은 줄 값에서(추가한 기종도 섬) - 옛 덮개 층 삭제 — Z01_MasterData_Overrides.py · 칸 단위 PUT · /overrides·/overrides/clear - 새 줄 열쇠는 서버가 줌(added/{난수}) · 주입 data 에 '@' 칸이 없어도 축 잠금은 파일 원본으로 섬 - 시험: DB 가짜(helper_z01_fake_repo.py) 로 API 손질 · 저장소 SQL 흐름은 가짜 커넥션으로 잼 Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01V1MKKZKpUHTPKb513FneU8 --- Z01_MasterData/Z01_MasterData_BasePrices.py | 269 +++++---- .../Z01_MasterData_BasePrices_Machine.py | 34 +- .../Z01_MasterData_BasePrices_Tables.py | 2 +- Z01_MasterData/Z01_MasterData_Overrides.py | 202 ------- .../Z01_MasterData_Repository_BasePrices.py | 168 ++++++ Z01_MasterData/Z01_MasterData_Router.py | 152 +++-- resources/tester/helper_z01_fake_repo.py | 92 +++ .../tester/test_master_labels_base_prices.py | 39 +- resources/tester/test_z01_base_prices.py | 550 +++++++++--------- resources/tester/test_z01_base_tables.py | 60 +- resources/tester/test_z01_work_items.py | 17 +- 11 files changed, 883 insertions(+), 702 deletions(-) delete mode 100644 Z01_MasterData/Z01_MasterData_Overrides.py create mode 100644 Z01_MasterData/Z01_MasterData_Repository_BasePrices.py create mode 100644 resources/tester/helper_z01_fake_repo.py diff --git a/Z01_MasterData/Z01_MasterData_BasePrices.py b/Z01_MasterData/Z01_MasterData_BasePrices.py index 9a3d7342..0ccb5576 100644 --- a/Z01_MasterData/Z01_MasterData_BasePrices.py +++ b/Z01_MasterData/Z01_MasterData_BasePrices.py @@ -1,16 +1,20 @@ """Z01 기초단가 다섯 — 노임·기계·자재·유가·요율을 kind 마다 **한 표**로(2026-09-15 사용자 지시 · 브레인 계약). 표 수준(한 번): columns · total · editable · locked{열: 왜 못 고치나} · formula{열: 식} · notice(요율만) -줄 수준(해당할 때만): `@id`(자료 제 열쇠) · `@overrides`(고친 줄) · `@formula`(계산값 줄) +줄 수준: `@id`(자료 제 열쇠) · `changed_columns`(초기값과 다른 칸) · `is_added`(관리자 추가 줄) · `@formula`(계산값 줄) `@id` — 줄 차례가 아니라 자료 열쇠(브레인 승인): 노임 `{dataset_id}/{occupation_code}` · 기계 machine_code · 자재 item_code · 유가 `{scope}/{변수}[/{sido_code}]`(원본 값 그대로 · 열 이름은 이름표 것) · 요율 `{변수}[/{목록}/{구간칸=값;…}]`(구간 칸 이름 차례 · 값·식 칸 뺌) -원본 파일 자리는 이름표 파일이 가리키는 곳(`Z01_MasterData_Tables`) — 원본은 읽기만, 고친 값은 덮개(`Z01_MasterData_Overrides`). +정본 = DB `master_base_price`(2026-09-18 PLAN 1-0 · `Z01_MasterData_Repository_BasePrices`) — 파일 원본(`base_rows`)은 + 초기값 1회 주입(`Z01_MasterData_Seed`)의 재료 · 공종 축 둘만 아직 파일 읽기. """ from __future__ import annotations +import asyncio import json +import math import re +import uuid from datetime import date from decimal import Decimal from functools import lru_cache @@ -19,7 +23,7 @@ from typing import Any from Z01_MasterData import Z01_MasterData_BasePrices_Machine as machine from Z01_MasterData import Z01_MasterData_BasePrices_Tables as base_tables -from Z01_MasterData import Z01_MasterData_Overrides as overrides +from Z01_MasterData import Z01_MasterData_Repository_BasePrices as repo from Z01_MasterData import Z01_MasterData_Tables as tables from Z01_MasterData import Z01_MasterData_WorkItems as work_items @@ -28,7 +32,7 @@ KINDS = ("labor", "machine", "material", "oil", "rate", *base_tables.SPEC, *work SOURCES_DIR = ( tables.RESOURCES / "data_master_sources" ) # 출처표(데스크탑 서브) — 어디서 받고 최신이 무엇인지 -_KEY = "자료 열쇠 — 바꾸면 덮개·다른 표가 이 줄을 못 찾음" +_KEY = "자료 열쇠 — 바꾸면 다른 표가 이 줄을 못 찾음" _SOURCE = "공표 원문 칸 — 원본 갱신으로만 바뀜" _RATE_EDITABLE = ( "rate_percent", @@ -39,14 +43,14 @@ _RATE_EDITABLE = ( "manager_thresholds.default_estimated_amount_krw", "manager_thresholds.civil_main_work_estimated_amount_krw", ) -#: ⚠ **지금 상태를 먼저** — 덮개를 읽는 곳이 Z01 밖에 0(2026-09-17 서브 훑기 · 브레인 「없으면 없다고」). +#: ⚠ **지금 상태를 먼저** — 고친 값을 읽는 곳이 Z01 밖에 0(2026-09-17 서브 훑기 · 브레인 「없으면 없다고」). #: 뒤 문장은 프로젝트 복사(요구 14)가 붙으면 참이 되는 약속이라 지우지 않고 「그 길이 붙으면」으로 둠. _SCOPE_NOTICE = ( "⚠ **아직 어디에도 안 쓰임** — 고친 값은 마스터에만 쌓임 · 프로젝트로 옮기는 길(프로젝트 복사)이 아직 없음." " 그 길이 붙으면: 여기서 고친 값은 **새로 만드는 프로젝트부터** 쓰임 — 이미 만든 프로젝트 금액은 안 바뀜" "(프로젝트는 만들 때 뜬 사본을 봄)" ) -_RATE_NOTICE = "법이 정한 값 — 고치면 원본 갱신 때 구간 이름이 바뀌어 고친 값이 주인을 잃기 쉬움(주인 없음으로 남음)" +_RATE_NOTICE = "법이 정한 값 — 고시가 바뀔 때만 고칠 것(구간 칸은 규칙이라 못 고침)" def _source(file_id: str) -> Path: @@ -277,7 +281,7 @@ def merge_same_rows(rows: list[dict[str, Any]]) -> list[dict[str, Any]]: def base_rows(kind: str) -> list[dict[str, Any]]: - """덮개 얹기 전 원본 줄 — 기계는 입력 칸까지(계산 칸은 덮개를 얹은 뒤 셈).""" + """파일 원본 줄 — 초기값 주입의 재료 · 기계는 입력 칸까지(계산 칸은 읽을 때 셈).""" if kind == "machine": doc, name = _doc("mach_base") return machine.base_rows(name, edition_of(doc)) @@ -290,33 +294,67 @@ def base_rows(kind: str) -> list[dict[str, Any]]: return _BUILDERS[kind]() -def _machine_inputs() -> machine.Inputs: - """기계가 끌어 쓰는 밑값 — **덮개를 얹은** 노임(건설)·전국평균 유가 + 그 판들의 기준일.""" - labor = [r for r in overrides.overlay("labor", _labor()) if r["source"] == "labor_const"] +def _machine_inputs(labor: list[dict[str, Any]], oil: list[dict[str, Any]]) -> machine.Inputs: + """기계가 끌어 쓰는 밑값 — DB 의 노임(건설)·전국평균 유가 + 그 판들의 기준일(빈 값 줄은 뺌).""" + labor = [r for r in labor if r.get("source") == "labor_const" and r.get("occupation_code")] oil = [ - r for r in overrides.overlay("oil", _oil()) if r["@id"] == f"national_average/{r['fuel']}" + r + for r in oil + if r["@id"] == f"national_average/{r.get('fuel')}" and r.get("price_krw_per_l") is not None ] return machine.Inputs( wages={ str(r["occupation_code"]): None - if r["daily_wage_krw"] is None + if r.get("daily_wage_krw") is None else Decimal(str(r["daily_wage_krw"])) for r in labor }, - wage_date=labor[0]["effective_date"] if labor else None, + wage_date=labor[0].get("effective_date") if labor else None, oil={r["fuel"]: Decimal(str(r["price_krw_per_l"])) for r in oil}, - oil_dates={r["fuel"]: r.get("date") or r["effective_date"] for r in oil}, + oil_dates={r["fuel"]: r.get("date") or r.get("effective_date") for r in oil}, operating_date=edition_of(_doc("pum_const")[0]), ) -def rows(kind: str) -> list[dict[str, Any]]: - """덮개를 얹은 줄 — 기계는 그 위에서 시간당 단가를 다시 셈.""" - out = overrides.overlay(kind, base_rows(kind)) - if kind == "machine": - inputs = _machine_inputs() - out = [machine.compute(r, inputs) for r in out] - return out +def same(a: Any, b: Any) -> bool: + """값 비교 — 215907 과 215907.0 은 같음 · None 은 None 과만 같음.""" + if a is None or b is None: + return a is None and b is None + if isinstance(a, (int, float)) and isinstance(b, (int, float)): + return Decimal(str(a)) == Decimal(str(b)) + return a == b + + +def _row(record: dict[str, Any]) -> dict[str, Any]: + """DB 한 줄 → 표 줄. `@data` = 저장된 값 그대로(쓸 때 여기에 고친 칸만 얹음 · 기계 계산 칸이 섞이지 않게).""" + data, seed = record["data"], record["seed"] + changed = [] + if seed is not None: + changed = [ + c + for c in dict.fromkeys([*seed, *data]) + if not c.startswith("@") and not same(data.get(c), seed.get(c)) + ] + row = {"@id": record["row_key"], **data} + row.update( + {"@id": record["row_key"], "@data": data, "@added": seed is None, "@changed": changed} + ) + return row + + +async def _stored(kind: str) -> list[dict[str, Any]]: + return [_row(r) for r in await repo.load(kind)] + + +async def rows(kind: str) -> list[dict[str, Any]]: + """DB 정본 줄(공종 축 둘만 파일) — 기계는 DB 노임·유가로 시간당 단가를 읽을 때마다 셈(식은 코드 · 브레인).""" + if kind in work_items.KINDS: + return work_items.base_rows(kind) + if kind != "machine": + return await _stored(kind) + out, labor, oil = await asyncio.gather(_stored("machine"), _stored("labor"), _stored("oil")) + inputs = _machine_inputs(labor, oil) + return [machine.compute(r, inputs) for r in out] def _columns(kind: str, all_rows: list[dict[str, Any]]) -> list[str]: @@ -446,45 +484,26 @@ def group_label(group: str) -> str: return named.get("name_ko") or group -def kinds() -> list[dict[str, Any]]: +async def kinds() -> list[dict[str, Any]]: """화면이 상자를 세울 목록 — **서버가 냄**(화면이 제 코드에 들면 새 kind 가 조용히 안 뜸). - 갈래 · 갈래 한글 이름 · kind · 한글 이름(이름표) · 줄 수. + 갈래 · 갈래 한글 이름 · kind · 한글 이름(이름표) · 줄 수(DB · 공종 축은 파일). """ + counted = await repo.counts() return [ { "group": kind_group(kind), "group_label": group_label(kind_group(kind)), "kind": kind, "label": kind_label(kind), - "rows": len(rows(kind)), + "rows": len(work_items.base_rows(kind)) + if kind in work_items.KINDS + else counted.get(kind, 0), } for kind in KINDS ] -def row_label(kind: str, row: dict[str, Any] | None) -> str: - if row is None: - return "" - if kind == "labor": - return str(row.get("occupation_name") or "") - if kind in work_items.KINDS: - return str(row.get("name") or "") - if kind == "machine": - return f"{row.get('machine_name') or ''} {row.get('specification') or ''}".strip() - if kind == "material": - return str(row.get("specification") or row.get("classification_name") or "") - if kind == "oil": - return f"{row.get('source_product_name') or row.get('fuel')} {row.get('region_name') or '전국평균'}" - if kind in base_tables.SPEC: # 기초데이터 아홉 — 축 칸을 이어 사람이 읽을 한 줄로 - return " · ".join(str(row[c]) for c in row.get("@axis", ()) if row.get(c) is not None) - named = (tables.load_labels().get("value_overrides") or {}).get( - f"rates::variables/{row['variable']}" - ) or {} - bracket = row["@id"].split("/", 2)[2].replace(";", " · ") if row["part"] else "" - return f"{named.get('name_ko') or row['variable']} {bracket}".strip() - - def column_meta(kind: str, key: str) -> dict[str, Any]: """열 이름 — 이름표 `merged_tables[kind]` → `columns[열key]` → 영문 key(이름은 이름표에만 · 브레인 ②).""" labels = tables.load_labels() @@ -501,27 +520,12 @@ def column_meta(kind: str, key: str) -> dict[str, Any]: } -#: 뜻 칸에서 뺄 자리 — 판·공표 날짜는 갱신 때마다 바뀌어 「뜻 바뀜」과 구별이 안 됨 -_NOT_MEANING = ("effective_date", "date", "notice_datetime") - - -def meaning_of(kind: str, row: dict[str, Any], editable: list[str]) -> dict[str, Any]: - """그 줄의 **값이 아닌 글자 칸**(이름·규격·단위·인도조건 …) — 이름표가 숨김 아니라 한 열만. - - 갱신으로 같은 코드에 다른 기준이 붙으면(노임 국가유산직종 개편 · 자재 KS 개정) 값만으로는 못 잡음. - """ - out = {} - for key, value in row.items(): - if key.startswith("@") or key in editable or key in _NOT_MEANING or key.endswith("_date"): - continue - if isinstance(value, str) and value.strip() and not column_meta(kind, key)["hidden"]: - out[key] = value - return out +_INNER = ("@source", "@axis", "@data", "@added", "@changed") def public(row: dict[str, Any]) -> dict[str, Any]: - """내보낼 줄 — 안쪽 칸(`@source`·`@axis`)은 뺌.""" - return {k: v for k, v in row.items() if k not in ("@source", "@axis")} + """내보낼 줄 — 안쪽 칸(`@source`·`@axis`·`@data`…)은 뺌.""" + return {k: v for k, v in row.items() if k not in _INNER} #: 찾기만 같게 보는 글자 — 원문이 물결표·가운뎃점을 여러 벌로 씀(∼ 26 · ～ 8 · ․ · ㆍ) · 자간 공백(「굴 삭 기」). @@ -538,6 +542,7 @@ def fold(text: str) -> str: def table( kind: str, + all_rows: list[dict[str, Any]], page: int = 1, size: int = tables.DEFAULT_PAGE_SIZE, q: str = "", @@ -545,7 +550,7 @@ def table( desc: bool = False, filters: dict[str, str] | None = None, ) -> dict[str, Any]: - all_rows = rows(kind) + """한 표 한 쪽 — `all_rows` 는 `rows(kind)` 가 준 전체(DB 읽기와 셈을 떼어 둠).""" columns = _columns(kind, all_rows) needle = fold(q) hits = ( @@ -569,7 +574,14 @@ def table( "columns": [column_meta(kind, c) for c in columns], # 못 고치는 칸(계산값)도 **세우기는 됨** — 화면은 이 목록으로 눌릴 제목을 가림 "sortable": list(columns), - "rows": [public(r) for r in hits[start : start + size]], + "rows": [ + { + **public(r), + "changed_columns": r.get("@changed", []), + "is_added": r.get("@added", False), + } + for r in hits[start : start + size] + ], "total": len(hits), **spec(kind, columns), "notice": notice(kind), @@ -578,49 +590,84 @@ def table( } -def edit(kind: str, row_id: str, values: dict[str, Any], by: Any) -> tuple[int, dict[str, Any]]: - """고치기 — (상태 코드, 고친 뒤 그 줄 또는 막은 까닭).""" - base = next((r for r in base_rows(kind) if r["@id"] == row_id), None) - if base is None: - return 404, {"message": f"없는 줄: {row_id}"} - info = spec(kind, _columns(kind, rows(kind))) - for column, value in values.items(): - if column in (base.get("@axis") or ()): - return 400, {"message": f"못 고치는 칸 {column} — {_KEY}"} - if column not in info["editable"] or column not in base: - reason = info["locked"].get(column) or "이 줄에 없는 칸" - return 400, {"message": f"못 고치는 칸 {column} — {reason}"} - problem = overrides.check_value(value) +class SaveError(ValueError): + """[저장] 을 막는 까닭 — 하나라도 있으면 아무것도 안 씀(전부 아니면 전무).""" + + def __init__(self, status: int, message: str) -> None: + super().__init__(message) + self.status = status + + +def check_value(value: Any) -> str | None: + """값 칸에 넣을 수 있나 — 막는 까닭(없으면 None). null 은 빈 값(미공표 노임처럼).""" + if value is None: + return None + if isinstance(value, bool) or not isinstance(value, (int, float)): + return "숫자만 넣을 수 있음" + if not math.isfinite(value) or value < 0: + return "0 이상 유한한 수만 넣을 수 있음" + return None + + +def _axis_of(kind: str) -> dict[str, tuple[str, ...]]: + """줄마다 축 칸 — 파일 원본에서(표 모양이라 값 고치기로 안 바뀜 · 주입 data 에 `@axis` 가 없어도 섬).""" + if kind not in base_tables.SPEC: + return {} + return {r["@id"]: tuple(r.get("@axis") or ()) for r in base_rows(kind)} + + +def _cell_problem( + row: dict[str, Any], column: str, value: Any, info: dict[str, Any], axis: dict, fillable: set +) -> str | None: + """한 칸 고치기를 막는 까닭 — 추가 줄은 저장 칸이면 글자 칸도(이름·규격을 사람이 넣은 줄).""" + if row["@added"]: + if column not in fillable: + return f"못 고치는 칸 {column} — {info['locked'].get(column) or '이 표에 없는 칸'}" + elif column in axis.get(row["@id"], ()): + return f"못 고치는 칸 {column} — {_KEY}" + elif column not in info["editable"] or column not in row: + return f"못 고치는 칸 {column} — {info['locked'].get(column) or '이 줄에 없는 칸'}" + if column in info["editable"]: + problem = check_value(value) + return f"{column}: {problem}" if problem else None + if isinstance(value, (dict, list)): + return f"{column}: 글자나 숫자만 넣을 수 있음" + return None + + +def plan_save( + kind: str, + shown: list[dict[str, Any]], + edits: list[dict[str, Any]], + added: list[dict[str, Any]], + deleted: list[str], +) -> tuple[dict[str, dict[str, Any]], dict[str, dict[str, Any]], list[str]]: + """[저장] 한 번 — 다 맞춰 본 뒤 쓸 것만: (고친 줄 data · 새 줄 data · 지울 줄). 막히면 SaveError. + + `shown` = `rows(kind)`(기계는 계산 칸까지 붙은 것) — 쓰는 값은 그 줄 `@data` 에 고친 칸만 얹음. + 새 줄 열쇠 = `added/{난수}`(브레인 계약에 열쇠 규칙이 없어 서버가 줌 · 응답으로 알림). + """ + by_key = {r["@id"]: r for r in shown} + info = spec(kind, _columns(kind, shown)) + axis = _axis_of(kind) + fillable = {c for r in shown for c in r["@data"] if not c.startswith("@")} # 계산 칸 제외 + changed: dict[str, dict[str, Any]] = {} + for edit in edits: + row = by_key.get(edit["row_key"]) + if row is None: + raise SaveError(404, f"없는 줄: {edit['row_key']}") + problem = _cell_problem(row, edit["column"], edit["value"], info, axis, fillable) if problem: - return 400, {"message": f"{column}: {problem}"} - overrides.write(kind, base, values, by, meaning_of(kind, base, info["editable"])) - return 200, public(next(r for r in rows(kind) if r["@id"] == row_id)) - - -def override_items( - kind: str = "", state: str = "", page: int = 1, size: int = tables.DEFAULT_PAGE_SIZE -) -> dict[str, Any]: - """모아 보기 — 서버가 정렬(원본 바뀜 → 주인 없음 → 고쳐짐) · 거름(kind·state) · 쪽 · kind 별 state 수.""" - out = [] - counts: dict[str, dict[str, int]] = {} - for each in KINDS: # 수는 늘 다섯 다 셈 — 거름은 줄에만 - base = base_rows(each) - by_id = {r["@id"]: r for r in base} - for item in overrides.items(each, base): - counts.setdefault(each, {}).setdefault(item["state"], 0) - counts[each][item["state"]] += 1 - if (kind and each != kind) or (state and item["state"] != state): - continue - item["row_label"] = row_label(each, by_id.get(item["row_id"])) - meta = column_meta(each, item["column"]) - item["column_label"] = meta["label"] - item["unit"] = meta["unit"] - out.append(item) - out.sort(key=lambda i: overrides.STATE_ORDER.index(i["state"])) - size = max(1, min(size, tables.MAX_PAGE_SIZE)) - start = (max(page, 1) - 1) * size - return {"items": out[start : start + size], "total": len(out), "counts": counts} - - -def clear_overrides(kind: str, state: str) -> int: - return overrides.clear(kind, state, base_rows(kind)) + raise SaveError(400, problem) + changed.setdefault(row["@id"], dict(row["@data"]))[edit["column"]] = edit["value"] + new: dict[str, dict[str, Any]] = {} + for data in added: + for column, value in data.items(): + problem = _cell_problem({"@added": True}, column, value, info, axis, fillable) + if problem: + raise SaveError(400, problem) + new[f"added/{uuid.uuid4().hex[:12]}"] = dict(data) + missing = [key for key in deleted if key not in by_key] + if missing: + raise SaveError(404, f"없는 줄: {', '.join(missing)}") + return changed, new, list(dict.fromkeys(deleted)) diff --git a/Z01_MasterData/Z01_MasterData_BasePrices_Machine.py b/Z01_MasterData/Z01_MasterData_BasePrices_Machine.py index 350a84c8..e0308862 100644 --- a/Z01_MasterData/Z01_MasterData_BasePrices_Machine.py +++ b/Z01_MasterData/Z01_MasterData_BasePrices_Machine.py @@ -1,7 +1,7 @@ """Z01 기초단가 — 기계 한 표: 기종 한 줄 + 시간당 단가(계산값) · 산출근거(2026-09-15 브레인 계약). 기계단가는 취득가 613 · 손료계수 387 · 운전경비(연료·잡재료·조종원 — 건설품셈 8-4) · 노임 · 유가가 계산으로 물림. - 고치는 칸 = 취득가 · 손료계수(이 표) — 노임·유가는 그 표에서 고치고 여기는 **덮개 얹은 값**을 끌어 씀 + 고치는 칸 = 취득가 · 손료계수(이 표) — 노임·유가는 그 표에서 고치고 여기는 **DB 의 지금 값**을 끌어 씀 계산 칸 = 손료 · 연료 · 조종원 · 시간당 단가 — editable 에서 빼고 locked 에 까닭 ⚠ 셈은 B09 함수를 그대로 부름(`load_machine_catalog` · `load_operating_records` · `hourly_machine_cost`) — 식을 두 벌로 안 둠. `hourly_cost_of`(B09) 와 같은 값인지 `test_z01_base_prices.py` 가 대조. 반만 선 값(연료·조종원 빠짐)은 시간당 단가를 안 냄. @@ -9,13 +9,14 @@ from __future__ import annotations -from dataclasses import dataclass, replace +from dataclasses import dataclass from decimal import Decimal from functools import lru_cache from typing import Any from B09_Estimation.B09_Estimation_MachineCost import ( OPERATOR_WAGE_FORMULA, + MachineSpec, hourly_machine_cost, load_machine_catalog, ) @@ -40,7 +41,7 @@ _FORMULA = { _LOCKED_COMPUTED = "계산값 — 밑값(취득가·손료계수는 이 표 · 노임·유가는 그 표)을 고치면 따라 바뀜" _LOCKED_OPERATING = "건설품셈 8-4 운전경비표(로직)에서 옴 — 기초단가가 아님" _LOCKED = { - "machine_code": "자료 열쇠 — 바꾸면 덮개·다른 표가 이 줄을 못 찾음", + "machine_code": "자료 열쇠 — 바꾸면 다른 표가 이 줄을 못 찾음", "machine_name": "공표 원문 칸 — 원본 갱신으로만 바뀜", "specification": "공표 원문 칸 — 원본 갱신으로만 바뀜", "economic_life_hours": "공표 원문 칸 — 원본 갱신으로만 바뀜", @@ -50,8 +51,8 @@ _LOCKED = { "misc_material_percent": _LOCKED_OPERATING, "operator_occupation_code": _LOCKED_OPERATING, "operator_person_per_day": _LOCKED_OPERATING, - "fuel_price_krw_per_l": "유가 표에서 고침 — 여기는 덮개 얹은 전국평균을 끌어 씀", - "operator_daily_wage_krw": "노임 표에서 고침 — 여기는 덮개 얹은 건설 노임을 끌어 씀", + "fuel_price_krw_per_l": "유가 표에서 고침 — 여기는 전국평균을 끌어 씀", + "operator_daily_wage_krw": "노임 표에서 고침 — 여기는 건설 노임을 끌어 씀", "hourly_note": "계산이 왜 안 섰는지 — 밑값이 비면 섬", "effective_date": "원본 판 기준일 — 원본 갱신으로만 바뀜", "operating_effective_date": "운전경비 원단위가 나온 건설품셈 판 기준일", @@ -75,7 +76,7 @@ def _fmt(value: Decimal | int | float) -> str: @dataclass(frozen=True) class Inputs: - """기계 셈이 다른 표에서 끌어 오는 값 — 덮개를 얹은 뒤 · 판 기준일과 함께.""" + """기계 셈이 다른 표에서 끌어 오는 값 — DB 지금 값 · 판 기준일과 함께.""" wages: dict[str, Decimal | None] wage_date: str | None @@ -123,10 +124,13 @@ def base_rows(source_name: str, effective_date: str | None) -> list[dict[str, An def compute(row: dict[str, Any], inputs: Inputs) -> dict[str, Any]: - """덮개 얹은 줄 → 시간당 단가 칸을 붙인 새 줄. B09 `hourly_cost_of` 와 같은 길.""" - catalog, records = _inputs() + """DB 줄 → 시간당 단가 칸을 붙인 새 줄. B09 `hourly_cost_of` 와 같은 길. + + 기종 제원은 줄 값에서(카탈로그 아님) — 관리자가 추가한 줄도 섬 · 운전경비 원단위는 코드로 찾아 없으면 빈칸. + """ + _, records = _inputs() code = row["@id"] - coef = row["loss_coefficient_per_hour"] + coef = row.get("loss_coefficient_per_hour") record = records.get(code) out = { **row, @@ -137,13 +141,17 @@ def compute(row: dict[str, Any], inputs: Inputs) -> dict[str, Any]: "operator_wage_date": None, **dict.fromkeys(COMPUTED), } - if coef is None: - out["hourly_note"] = "손료계수 없음 — 시간당 단가 안 섬" + if coef is None or row.get("price_thousand_krw") is None: + out["hourly_note"] = f"{'손료계수' if coef is None else '취득가'} 없음 — 시간당 단가 안 섬" return out - spec = replace( - catalog.machines[code], + spec = MachineSpec( + machine_code=code, + name=str(row.get("machine_name") or ""), + specification=str(row.get("specification") or ""), price_thousand_krw=Decimal(str(row["price_thousand_krw"])), loss_coefficient_per_hour=Decimal(str(coef)), + economic_life_hours=row.get("economic_life_hours"), + annual_standard_hours=row.get("annual_standard_hours"), ) liters = fuel_price = wage = None if record is not None: diff --git a/Z01_MasterData/Z01_MasterData_BasePrices_Tables.py b/Z01_MasterData/Z01_MasterData_BasePrices_Tables.py index a89bdf7f..a0f55798 100644 --- a/Z01_MasterData/Z01_MasterData_BasePrices_Tables.py +++ b/Z01_MasterData/Z01_MasterData_BasePrices_Tables.py @@ -4,7 +4,7 @@ 태움 — 토량환산 · 자재 할증 · 거푸집 전용 · 철근 갈래 · 목재 갈래 · 돌쌓기 경사·뒷길이·돌 종류·갈래 안 태움 — 이음표(`type_map`·`form_map`·`bond.codes`) · 품셈 원문 표(coef `surcharge_*`) · 글·방침 · 구조물 실측(울진 한 현장 관찰값) · 기슭막이 교본값(값과 글이 한 줄에 섞임) -⚠ 축 칸(`@axis`)은 못 고침 — 그 칸이 `@id` 를 이루므로 바꾸면 덮개가 줄을 못 찾음. +⚠ 축 칸(`@axis`)은 못 고침 — 그 칸이 `@id` 를 이루므로 바꾸면 다른 표가 줄을 못 찾음. ⚠ 원문 중복(coef 「암괴…점토」 두 줄)은 **값이 같을 때만** 한 줄로 합침 — 갈리면 `DuplicateRowError`(브레인 조건). """ diff --git a/Z01_MasterData/Z01_MasterData_Overrides.py b/Z01_MasterData/Z01_MasterData_Overrides.py deleted file mode 100644 index 9cdc5a8e..00000000 --- a/Z01_MasterData/Z01_MasterData_Overrides.py +++ /dev/null @@ -1,202 +0,0 @@ -"""Z01 기초단가 덮개 — 사람이 고친 값만 담는 층(2026-09-15 브레인 틀 · 사용자 지시 「기초단가를 고칠 수 있게」). - - 원본 resources/data_*/… 읽기만 — 주기 갱신은 이 파일을 통째로 갈아끼움 - 덮개 resources/data_master_override/{kind}.json 고친 것만 — 주기가 달라 kind 마다 한 파일 - 읽기 원본 줄 위에 덮개를 얹음 · 줄 수준 `@overrides[열] = {original, value, state, current}` - -state 셋(판정은 **여기 한 곳** — 화면이 따로 재지 않음): - edited 고쳐짐 — 원본 그대로(덮개를 쓸 때 본 원본 = 지금 원본) - source_changed 고쳐짐 — 원본이 바뀜 · ⚠ 덮개 값을 그대로 씀(자동으로 원본을 쓰면 금액이 말없이 바뀜 — 사람이 정함) - orphan 주인 없음 — 갱신으로 그 줄·칸이 사라짐 · 버리지 않고 모아 보기에 남김 -되돌리기(「새 원본 값으로 받기」 도 같음) = 값 null → 그 칸을 덮개에서 뺌. -""" - -from __future__ import annotations - -import json -import math -import threading -from datetime import datetime -from decimal import Decimal -from pathlib import Path -from typing import Any - -from common_util.common_util_json import atomic_write_json - -OVERRIDE_DIR = Path(__file__).resolve().parent.parent / "resources" / "data_master_override" -EDITED, SOURCE_CHANGED, ORPHAN = "edited", "source_changed", "orphan" -#: ⚠ 넷째 — 값이 아니라 **줄의 뜻**이 바뀜(같은 코드에 다른 기준: 노임 문화재→국가유산 · 자재 KS 개정 규격). -#: 값이 틀린 것보다 **다른 물건에 값이 붙은 것**이 더 나쁨 → 맨 위(브레인 2026-09-16). -MEANING_CHANGED = "meaning_changed" -STATE_ORDER = (MEANING_CHANGED, SOURCE_CHANGED, ORPHAN, EDITED) # 모아 보기 차례 — 센 것이 맨 위 -_LOCK = threading.Lock() # ponytail: 한 프로세스 잠금 — 서버를 여럿 띄우면 파일 잠금이 필요 - - -def _path(kind: str) -> Path: - return OVERRIDE_DIR / f"{kind}.json" - - -def load(kind: str) -> dict[str, dict[str, dict[str, Any]]]: - """`@id` → 열 → {value, original, source, at, by}. 파일이 없으면 빈 덮개.""" - path = _path(kind) - if not path.is_file(): - return {} - return json.loads(path.read_text(encoding="utf-8")).get("overrides") or {} - - -def same(a: Any, b: Any) -> bool: - """값 비교 — 215907 과 215907.0 은 같음 · None 은 None 과만 같음.""" - if a is None or b is None: - return a is None and b is None - if isinstance(a, (int, float)) and isinstance(b, (int, float)): - return Decimal(str(a)) == Decimal(str(b)) - return a == b - - -def check_value(value: Any) -> str | None: - """넣을 수 있는 값인가 — 막는 까닭(없으면 None). null 은 되돌리기라 통과.""" - if value is None: - return None - if isinstance(value, bool) or not isinstance(value, (int, float)): - return "숫자만 넣을 수 있음" - if not math.isfinite(value) or value < 0: - return "0 이상 유한한 수만 넣을 수 있음" - return None - - -#: 이름 칸이 여럿인 자료가 있어(자재 품명·규격·단위·인도조건) **서버가 한 줄로 합침** — 화면이 합치면 판정이 두 벌. -_JOIN = " · " - - -def meaning_line(entry: dict[str, Any], row: dict[str, Any] | None) -> tuple[str, str]: - """(고칠 때 그 줄 · 지금 그 줄) 한 줄씩 — 달라진 데가 없으면 빈 둘.""" - stored = entry.get("meaning") or {} - keys = [k for k in stored if row is not None and k in row] - if not keys or all(row[k] == stored[k] for k in keys): - return "", "" - return ( - _JOIN.join(str(stored[k]) for k in keys), - _JOIN.join(str(row[k]) for k in keys), - ) - - -def mark(entry: dict[str, Any], current: Any, row: dict[str, Any] | None = None) -> dict[str, Any]: - original, value = entry.get("original"), entry.get("value") - was, now = meaning_line(entry, row) - if was: - state = MEANING_CHANGED - else: - state = EDITED if same(original, current) else SOURCE_CHANGED - out = {"original": original, "value": value, "state": state, "current": current} - if was: # ⚠ meaning_changed 인데 비어 오면 서버 버그 — 늘 함께 나감 - out["was"], out["now"] = was, now - return out - - -def overlay(kind: str, rows: list[dict[str, Any]]) -> list[dict[str, Any]]: - """원본 줄(바꾸지 않음) 위에 덮개를 얹은 **새 줄들**.""" - store = load(kind) - out = [] - for row in rows: - entries = store.get(row["@id"]) - if not entries: - out.append(row) - continue - row = dict(row) - marks = {} - for column, entry in entries.items(): - if column not in row: - continue # 주인 없는 칸 — 모아 보기에만 - marks[column] = mark(entry, row[column], row) - row[column] = entry.get("value") - if marks: - row["@overrides"] = marks - out.append(row) - return out - - -def write( - kind: str, - base_row: dict[str, Any], - values: dict[str, Any], - by: Any, - meaning: dict[str, Any] | None = None, -) -> None: - """고친 값을 덮개에 씀 — null 이거나 지금 원본과 같으면 그 칸을 뺌. 원본은 안 건드림.""" - row_id = base_row["@id"] - with _LOCK: - store = load(kind) - entries = dict(store.get(row_id) or {}) - for column, value in values.items(): - current = base_row.get(column) - if value is None or same(value, current): - entries.pop(column, None) - continue - entries[column] = { - "value": value, - "original": current, # 이 값을 보고 고침 — 갱신 뒤 원본과 달라지면 source_changed - "source": base_row.get("@source", ""), - # 값이 아닌 글자 칸 — 갱신으로 **줄의 뜻**이 바뀌면 이것으로 잡음 - "meaning": meaning or {}, - "at": datetime.now().astimezone().isoformat(timespec="seconds"), - "by": by, - } - if entries: - store[row_id] = entries - else: - store.pop(row_id, None) - _save(kind, store) - - -def _save(kind: str, store: dict[str, dict[str, dict[str, Any]]]) -> None: - atomic_write_json( - _path(kind), - { - "schema_version": "1.0", - "kind": kind, - "note": "사람이 고친 기초단가만 — 원본 파일은 그대로 · Z01 마스터 데이터 화면이 씀", - "overrides": store, - }, - ) - - -def clear(kind: str, state: str, base_rows: list[dict[str, Any]]) -> int: - """한 kind 의 그 state 덮개를 한꺼번에 뺌 — 품목 퇴출로 여럿이 주인을 잃는 갱신 뒤(브레인 ②). 뺀 수.""" - with _LOCK: - doomed = {(i["row_id"], i["column"]) for i in items(kind, base_rows) if i["state"] == state} - if not doomed: - return 0 - store = load(kind) - for row_id, column in doomed: - store[row_id].pop(column, None) - if not store[row_id]: - del store[row_id] - _save(kind, store) - return len(doomed) - - -def items(kind: str, base_rows: list[dict[str, Any]]) -> list[dict[str, Any]]: - """모아 보기 한 kind 몫 — 주인 없는 것까지 · 차례는 부르는 쪽이 STATE_ORDER 로.""" - by_id = {r["@id"]: r for r in base_rows} - out = [] - for row_id, entries in load(kind).items(): - row = by_id.get(row_id) - for column, entry in entries.items(): - if row is None or column not in row: - current, marked = None, {"state": ORPHAN} - else: - current = row[column] - marked = mark(entry, current, row) - out.append( - { - "kind": kind, - "row_id": row_id, - "column": column, - "original": entry.get("original"), - "current": current, - "value": entry.get("value"), - "state": marked["state"], - **({"was": marked["was"], "now": marked["now"]} if "was" in marked else {}), - } - ) - return out diff --git a/Z01_MasterData/Z01_MasterData_Repository_BasePrices.py b/Z01_MasterData/Z01_MasterData_Repository_BasePrices.py new file mode 100644 index 00000000..a3013799 --- /dev/null +++ b/Z01_MasterData/Z01_MasterData_Repository_BasePrices.py @@ -0,0 +1,168 @@ +"""Z01 기초단가 저장소 — `master_base_price` · `master_revision` (020 · PLAN 1-0 계약 · 2026-09-18). + +정본 = DB. 줄 = (kind, row_key) 한 줄 · `data` 는 지금 값 한 줄 전체 · `seed` 는 주입 때 초기값(관리자 추가 줄은 NULL). +판번호 = `master_revision.revision`(scope `base:{kind}`) — 쓸 때마다 +1 · 읽어 간 판과 다르면 안 씀(409). +⚠ 여기는 SQL 만 — 무엇을 고칠 수 있나·값 검사는 `Z01_MasterData_BasePrices` 가 함. +""" + +from __future__ import annotations + +import json +from typing import Any + +import aiomysql + +from config.config_db import get_db_pool + + +class RevisionConflict(Exception): + """읽어 간 판번호가 지금 판과 다름 — 그 사이 누가 저장함.""" + + +class UnknownRows(LookupError): + """없는 줄 열쇠 — 되돌릴 줄을 못 찾음.""" + + +def _scope(kind: str) -> str: + return f"base:{kind}" + + +def _json(text: Any) -> Any: + return json.loads(text) if isinstance(text, (str, bytes)) else text + + +def _dump(data: dict[str, Any]) -> str: + return json.dumps(data, ensure_ascii=False, default=str) + + +async def load(kind: str) -> list[dict[str, Any]]: + """지운 줄 뺀 전부 — 넣은 차례(id) 그대로. {row_key, data, seed}.""" + pool = get_db_pool() + async with pool.acquire() as connection, connection.cursor(aiomysql.DictCursor) as cursor: + await cursor.execute( + """SELECT row_key, data, seed FROM master_base_price + WHERE kind = %s AND is_deleted = 0 ORDER BY id""", + (kind,), + ) + return [ + {"row_key": r["row_key"], "data": _json(r["data"]), "seed": _json(r["seed"])} + for r in await cursor.fetchall() + ] + + +async def revision(kind: str) -> int: + pool = get_db_pool() + async with pool.acquire() as connection, connection.cursor() as cursor: + await cursor.execute( + "SELECT revision FROM master_revision WHERE scope = %s", (_scope(kind),) + ) + row = await cursor.fetchone() + return int(row[0]) if row else 0 + + +async def counts() -> dict[str, int]: + """kind 마다 줄 수(지운 줄 뺌) — 표 목록 한 번에.""" + pool = get_db_pool() + async with pool.acquire() as connection, connection.cursor() as cursor: + await cursor.execute( + "SELECT kind, COUNT(*) FROM master_base_price WHERE is_deleted = 0 GROUP BY kind" + ) + return {kind: int(n) for kind, n in await cursor.fetchall()} + + +async def _lock_revision(cursor: Any, kind: str) -> int: + """판번호 줄을 잠그고 읽음 — 없으면 0 으로 세움(같은 kind 쓰기는 여기서 한 줄로 섬).""" + await cursor.execute( + "INSERT IGNORE INTO master_revision (scope, revision) VALUES (%s, 0)", (_scope(kind),) + ) + await cursor.execute( + "SELECT revision FROM master_revision WHERE scope = %s FOR UPDATE", (_scope(kind),) + ) + return int((await cursor.fetchone())[0]) + + +async def _bump(cursor: Any, kind: str) -> None: + await cursor.execute( + "UPDATE master_revision SET revision = revision + 1 WHERE scope = %s", (_scope(kind),) + ) + + +async def save( + kind: str, + base_revision: int, + changed: dict[str, dict[str, Any]], + added: dict[str, dict[str, Any]], + deleted: list[str], + by: int | None, +) -> int: + """한 트랜잭션 — 판이 맞을 때만 고친 줄·새 줄·지운 줄을 한꺼번에(전부 아니면 전무). 새 판번호.""" + pool = get_db_pool() + async with pool.acquire() as connection, connection.cursor() as cursor: + try: + await connection.begin() + current = await _lock_revision(cursor, kind) + if current != base_revision: + raise RevisionConflict(current) + for row_key, data in changed.items(): + await cursor.execute( + """UPDATE master_base_price + SET data = %s, updated_by = %s, updated_at = CURRENT_TIMESTAMP + WHERE kind = %s AND row_key = %s AND is_deleted = 0""", + (_dump(data), by, kind, row_key), + ) + for row_key, data in added.items(): + await cursor.execute( + """INSERT INTO master_base_price + (kind, row_key, data, seed, edition, is_deleted, updated_by, updated_at) + VALUES (%s, %s, %s, NULL, %s, 0, %s, CURRENT_TIMESTAMP)""", + (kind, row_key, _dump(data), str(data.get("effective_date") or ""), by), + ) + for row_key in deleted: + await cursor.execute( + """UPDATE master_base_price + SET is_deleted = 1, updated_by = %s, updated_at = CURRENT_TIMESTAMP + WHERE kind = %s AND row_key = %s""", + (by, kind, row_key), + ) + await _bump(cursor, kind) + await connection.commit() + return current + 1 + except Exception: + await connection.rollback() + raise + + +async def reset(kind: str, row_keys: list[str], by: int | None) -> int: + """고른 줄을 초기값으로 — 주입 줄은 data := seed(지운 줄도 되살림) · 추가 줄(seed NULL)은 지움. 새 판번호.""" + keys = list(dict.fromkeys(row_keys)) + marks = ", ".join(["%s"] * len(keys)) + pool = get_db_pool() + async with pool.acquire() as connection, connection.cursor() as cursor: + try: + await connection.begin() + current = await _lock_revision(cursor, kind) + await cursor.execute( + f"SELECT row_key FROM master_base_price WHERE kind = %s AND row_key IN ({marks})", + (kind, *keys), + ) + missing = set(keys) - {r[0] for r in await cursor.fetchall()} + if missing: + raise UnknownRows(sorted(missing)) + await cursor.execute( + f"""UPDATE master_base_price + SET data = seed, is_deleted = 0, updated_by = %s, updated_at = CURRENT_TIMESTAMP + WHERE kind = %s AND seed IS NOT NULL AND row_key IN ({marks})""", + (by, kind, *keys), + ) + await cursor.execute( + f"""UPDATE master_base_price + SET is_deleted = 1, updated_by = %s, updated_at = CURRENT_TIMESTAMP + WHERE kind = %s AND seed IS NULL AND row_key IN ({marks})""", + (by, kind, *keys), + ) + await _bump(cursor, kind) + await connection.commit() + return current + 1 + except Exception: + await connection.rollback() + raise diff --git a/Z01_MasterData/Z01_MasterData_Router.py b/Z01_MasterData/Z01_MasterData_Router.py index 294d1337..d47be921 100644 --- a/Z01_MasterData/Z01_MasterData_Router.py +++ b/Z01_MasterData/Z01_MasterData_Router.py @@ -1,32 +1,69 @@ -"""Z01 마스터 데이터 라우터 — 읽기 + 기초단가 고치기(2026-09-15 브레인). +"""Z01 마스터 데이터 라우터 — 읽기 + 기초단가 DB 정본 일괄 저장(2026-09-18 PLAN 1-0 계약). - GET /api/master-data/tree 갈래 → 파일 → 표 - GET /api/master-data/rows?file=&table=&page=&size=&q= 표 줄(쪽 나누기 · 검색) - GET /api/master-data/base-prices 표 목록(갈래·kind·이름·줄 수 · 공종 축 산림·건설 포함) - GET /api/master-data/base-prices/{kind}?page=&size=&q=&sort=&desc=[&거르기=값] 한 표(기초단가 다섯 · 품셈 기준 열 · 공종 축 둘) - PUT /api/master-data/base-prices/{kind}/{row_id} {values:{열:값}} → 덮개에만 씀 · null = 되돌리기 - GET /api/master-data/overrides 고친 것·원본 바뀐 것·주인 없는 것(서버 정렬) + GET /api/master-data/tree 갈래 → 파일 → 표 + GET /api/master-data/rows?file=&table=&page=&size=&q= 표 줄(쪽 나누기 · 검색) + GET /api/master-data/base-prices 표 목록(갈래·kind·이름·줄 수 · 공종 축 산림·건설 포함) + GET /api/master-data/base-prices/{kind}?page=&size=&q=&sort=&desc=[&거르기=값] 한 표 + revision · 줄마다 changed_columns · is_added + PUT /api/master-data/base-prices/{kind} {base_revision, edits, added, deleted} → 409 · 전부 아니면 전무 + POST /api/master-data/base-prices/{kind}/reset {row_keys} → 초기값으로(추가 줄은 지움) ⚠ 권한은 등록하는 쪽(`main.py` · 랩탑 서브)이 `dependencies=[verify_session, require_system_admin]` 로 붙임. +⚠ 자동저장 없음 — 화면은 [저장] 에서만 PUT(지침 5장). """ from __future__ import annotations -from typing import Any, Literal +import asyncio +from typing import Any from fastapi import APIRouter, Depends, HTTPException, Request from fastapi.responses import JSONResponse -from pydantic import BaseModel +from pydantic import BaseModel, ConfigDict, Field +from starlette.concurrency import run_in_threadpool from common_util.common_util_auth import verify_session from Z01_MasterData import Z01_MasterData_BasePrices as base_prices +from Z01_MasterData import Z01_MasterData_Repository_BasePrices 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"]) +_STALE = "그 사이 누가 이 표를 저장했습니다 — 표를 다시 불러와 고칠 것." -class BasePriceEdit(BaseModel): - values: dict[str, Any] + +class CellEdit(BaseModel): + model_config = ConfigDict(extra="forbid") + + row_key: str = Field(min_length=1, max_length=190) + column: str = Field(min_length=1, max_length=100) + value: Any = None + + +class AddedRow(BaseModel): + model_config = ConfigDict(extra="forbid") + + data: dict[str, Any] = Field(min_length=1) + + +class BasePriceSave(BaseModel): + """[저장] 한 번 — 읽어 간 판번호와 함께 고친 칸·새 줄·지운 줄.""" + + model_config = ConfigDict(extra="forbid") + + base_revision: int = Field(ge=0) + edits: list[CellEdit] = Field(default_factory=list, max_length=500) + added: list[AddedRow] = Field(default_factory=list, max_length=500) + deleted: list[str] = Field(default_factory=list, max_length=500) + + +class BasePriceReset(BaseModel): + model_config = ConfigDict(extra="forbid") + + row_keys: list[str] = Field(min_length=1, max_length=10000) + + +def _error(status: int, message: str) -> JSONResponse: + return JSONResponse(status_code=status, content={"status": "error", "message": message}) @router.get("/tree") @@ -50,14 +87,19 @@ def _kind(kind: str) -> str: return kind +def _db_kind(kind: str) -> str | None: + """고칠 수 있는 표(DB 정본) — 공종 축은 여기서 안 고침(`/work-items` 몫).""" + return kind if kind in base_prices.KINDS and kind not in work_items.KINDS else None + + @router.get("/base-prices") -def get_base_price_kinds() -> dict: +async def get_base_price_kinds() -> dict: """어떤 표가 있나 — 갈래·kind·한글 이름·줄 수. 화면이 목록을 제 코드에 들지 않게 서버가 냄.""" - return {"kinds": base_prices.kinds()} + return {"status": "success", "kinds": await base_prices.kinds()} @router.get("/base-prices/{kind}") -def get_base_prices( +async def get_base_prices( kind: str, request: Request, page: int = 1, @@ -66,39 +108,71 @@ def get_base_prices( sort: str = "", desc: int = 0, ) -> dict: + kind = _kind(kind) # 거르기 — 표가 준 `filters` 의 key 만 받음(공종 축 부문·줄 구실) picked = {k: v for k, v in request.query_params.items() if k in work_items.FILTER_KEYS} + if _db_kind(kind): + all_rows, revision = await asyncio.gather(base_prices.rows(kind), repo.revision(kind)) + else: + all_rows, revision = await base_prices.rows(kind), 0 try: - return base_prices.table( - _kind(kind), page=page, size=size, q=q, sort=sort, desc=bool(desc), filters=picked + table = await run_in_threadpool( + base_prices.table, + kind, + all_rows, + page=page, + size=size, + q=q, + sort=sort, + desc=bool(desc), + filters=picked, ) except base_prices.SortError as error: raise HTTPException(status_code=400, detail=str(error)) from error + return {"status": "success", "revision": revision, **table} -@router.put("/base-prices/{kind}/{row_id:path}") -def put_base_price( - kind: str, row_id: str, body: BasePriceEdit, session: dict = Depends(verify_session) +@router.put("/base-prices/{kind}") +async def put_base_prices( + kind: str, body: BasePriceSave, session: dict = Depends(verify_session) ) -> JSONResponse: - status, payload = base_prices.edit(_kind(kind), row_id, body.values, session.get("user_id")) - if status != 200: - raise HTTPException(status_code=status, detail=payload["message"]) - return JSONResponse(content=payload) + """고친 칸·새 줄·지운 줄을 한 번에 — 다 맞춰 본 뒤 한 트랜잭션(하나라도 틀리면 아무것도 안 씀).""" + if not _db_kind(kind): + return _error(404 if kind not in base_prices.KINDS else 400, f"고칠 수 없는 표: {kind}") + if await repo.revision(kind) != body.base_revision: + return _error(409, _STALE) + try: + changed, added, deleted = base_prices.plan_save( + kind, + await base_prices.rows(kind), + [edit.model_dump() for edit in body.edits], + [row.data for row in body.added], + body.deleted, + ) + except base_prices.SaveError as error: + return _error(error.status, str(error)) + if not (changed or added or deleted): + return JSONResponse( + content={"status": "success", "revision": body.base_revision, "added": []} + ) + try: + revision = await repo.save( + kind, body.base_revision, changed, added, deleted, session.get("user_id") + ) + except repo.RevisionConflict: + return _error(409, _STALE) + return JSONResponse(content={"status": "success", "revision": revision, "added": list(added)}) -@router.get("/overrides") -def get_overrides( - kind: str = "", state: str = "", page: int = 1, size: int = tables.DEFAULT_PAGE_SIZE -) -> dict: - return base_prices.override_items(kind and _kind(kind), state, page=page, size=size) - - -class OverrideClear(BaseModel): - kind: str - state: Literal["edited", "source_changed", "orphan"] - - -@router.post("/overrides/clear") -def clear_overrides(body: OverrideClear, _session: dict = Depends(verify_session)) -> dict: - """한 kind 의 그 state 덮개를 한꺼번에 뺌 — kind·state 둘 다 있어야 함(통째로 비우기 막음).""" - return {"removed": base_prices.clear_overrides(_kind(body.kind), body.state)} +@router.post("/base-prices/{kind}/reset") +async def reset_base_prices( + kind: str, body: BasePriceReset, session: dict = Depends(verify_session) +) -> JSONResponse: + """고른 줄을 초기값으로 — 주입 줄은 data := seed · 추가 줄은 지움. 판번호 +1.""" + if not _db_kind(kind): + return _error(404 if kind not in base_prices.KINDS else 400, f"고칠 수 없는 표: {kind}") + try: + revision = await repo.reset(kind, body.row_keys, session.get("user_id")) + except repo.UnknownRows as error: + return _error(404, f"없는 줄: {', '.join(error.args[0])}") + return JSONResponse(content={"status": "success", "revision": revision}) diff --git a/resources/tester/helper_z01_fake_repo.py b/resources/tester/helper_z01_fake_repo.py new file mode 100644 index 00000000..f20f5ddd --- /dev/null +++ b/resources/tester/helper_z01_fake_repo.py @@ -0,0 +1,92 @@ +"""Z01 기초단가 저장소 가짜 — DB 없이 API 시험(PLAN 1-2 · 2026-09-18). + +`install(monkeypatch)` 가 `Z01_MasterData_Repository_BasePrices` 의 함수 다섯을 바꿔 끼움. +주입 = 파일 원본 줄(`base_rows`) 그대로 · ⚠ '@' 칸(`@axis`·`@source`)은 **빼고** 담음 — +주입기가 안쪽 칸을 빼도 서버가 서는지 함께 잼. 판번호 규칙(다르면 409 · 쓰면 +1)은 진짜와 같게. +""" + +from __future__ import annotations + +from typing import Any + +import pytest + +from Z01_MasterData import Z01_MasterData_BasePrices as base_prices +from Z01_MasterData import Z01_MasterData_Repository_BasePrices as repo +from Z01_MasterData import Z01_MasterData_WorkItems as work_items + + +class FakeRepo: + def __init__(self) -> None: + self.tables: dict[str, list[dict[str, Any]]] = {} + self.revisions: dict[str, int] = {} + self.saved_by: list[Any] = [] + + def table(self, kind: str) -> list[dict[str, Any]]: + if kind not in self.tables: + self.tables[kind] = [] + for row in base_prices.base_rows(kind): + data = {k: v for k, v in row.items() if not k.startswith("@")} + self.tables[kind].append( + {"row_key": row["@id"], "data": data, "seed": dict(data), "deleted": False} + ) + return self.tables[kind] + + def by_key(self, kind: str) -> dict[str, dict[str, Any]]: + return {r["row_key"]: r for r in self.table(kind)} + + async def load(self, kind: str) -> list[dict[str, Any]]: + return [ + {"row_key": r["row_key"], "data": dict(r["data"]), "seed": r["seed"]} + for r in self.table(kind) + if not r["deleted"] + ] + + async def revision(self, kind: str) -> int: + return self.revisions.get(kind, 0) + + async def counts(self) -> dict[str, int]: + return { + kind: sum(not r["deleted"] for r in self.table(kind)) + for kind in base_prices.KINDS + if kind not in work_items.KINDS + } + + async def save(self, kind, base_revision, changed, added, deleted, by) -> int: + current = self.revisions.get(kind, 0) + if current != base_revision: + raise repo.RevisionConflict(current) + rows = self.by_key(kind) + for key, data in changed.items(): + if not rows[key]["deleted"]: + rows[key]["data"] = dict(data) + for key, data in added.items(): + self.table(kind).append( + {"row_key": key, "data": dict(data), "seed": None, "deleted": False} + ) + for key in deleted: + rows[key]["deleted"] = True + self.revisions[kind] = current + 1 + self.saved_by.append(by) + return current + 1 + + async def reset(self, kind, row_keys, by) -> int: + rows = self.by_key(kind) + missing = sorted(set(row_keys) - set(rows)) + if missing: + raise repo.UnknownRows(missing) + for key in row_keys: + row = rows[key] + if row["seed"] is None: + row["deleted"] = True + else: + row["data"], row["deleted"] = dict(row["seed"]), False + self.revisions[kind] = self.revisions.get(kind, 0) + 1 + return self.revisions[kind] + + +def install(monkeypatch: pytest.MonkeyPatch) -> FakeRepo: + fake = FakeRepo() + for name in ("load", "revision", "counts", "save", "reset"): + monkeypatch.setattr(repo, name, getattr(fake, name)) + return fake diff --git a/resources/tester/test_master_labels_base_prices.py b/resources/tester/test_master_labels_base_prices.py index 67ec9b59..a67fae7f 100644 --- a/resources/tester/test_master_labels_base_prices.py +++ b/resources/tester/test_master_labels_base_prices.py @@ -10,10 +10,12 @@ from __future__ import annotations +import asyncio import json from pathlib import Path import pytest +from helper_z01_fake_repo import install from Z01_MasterData import Z01_MasterData_BasePrices as base_prices @@ -28,11 +30,13 @@ def labels() -> dict: @pytest.fixture(scope="module") def shown() -> dict[str, list[dict]]: - """서버가 실제로 내보내는 열 — 이름표 파일이 아니라 **응답**을 잰다.""" + """서버가 실제로 내보내는 열 — 이름표 파일이 아니라 **응답**을 잰다(DB 는 파일을 주입한 가짜).""" out = {} - for kind in base_prices.KINDS: - columns = base_prices._columns(kind, base_prices.rows(kind)) - out[kind] = [base_prices.column_meta(kind, c) for c in columns] + with pytest.MonkeyPatch.context() as patch: + install(patch) + for kind in base_prices.KINDS: + columns = base_prices._columns(kind, asyncio.run(base_prices.rows(kind))) + out[kind] = [base_prices.column_meta(kind, c) for c in columns] return out @@ -99,8 +103,8 @@ def test_요율_축이_한_칸으로_안_합쳐졌다(shown): def test_요율_줄_열쇠에_축이_살아_있다(): - """`@id` 가 `{변수}[/{목록}/{축=값;…}]` 꼴 — 축이 죽으면 고친 값이 주인을 잃는다.""" - rows = base_prices.rows("rate") + """`@id` 가 `{변수}[/{목록}/{축=값;…}]` 꼴 — 축이 죽으면 주입 열쇠(row_key)가 흔들린다.""" + rows = base_prices.base_rows("rate") multi = [r for r in rows if "/" in str(r["@id"])] assert multi, "축이 든 줄 열쇠가 하나도 없다" assert any("=" in str(r["@id"]) for r in multi) @@ -109,7 +113,7 @@ def test_요율_줄_열쇠에_축이_살아_있다(): def test_기계_시간당_단가는_계산값으로_잠긴다(shown): """밑값만 고칠 수 있다 — 판정은 서버 `spec()` 한 곳(이름표가 따로 안 정한다).""" - meta = base_prices.table("machine", page=1, size=1) + meta = base_prices.table("machine", base_prices.base_rows("machine"), page=1, size=1) assert set(meta["formula"]) == { "hourly_loss_krw", "hourly_fuel_krw", @@ -155,15 +159,15 @@ def test_단위가_붙어야_할_열에_단위가_있다(shown): def test_줄_열쇠가_서버_축과_같다(labels): - """⭐ 덮개가 줄을 붙드는 끈 — 서버가 줄마다 `@axis` 로 내는 축이 곧 열쇠다. + """⭐ 줄을 붙드는 끈 — 파일 원본이 줄마다 `@axis` 로 내는 축이 곧 열쇠(주입 row_key)다. - 이름표가 축을 빠뜨리면 고친 값이 갱신 뒤 주인을 잃는다. + 이름표가 축을 빠뜨리면 새 판 주입 때 같은 줄을 못 알아본다. """ declared = {r["table"]: set(r["key"]) for r in labels["row_keys"]} bad = [] for kind in base_prices.KINDS: axes = set() - for row in base_prices.rows(kind): + for row in base_prices.base_rows(kind): axes |= set(row.get("@axis") or ()) if not axes: # 다섯 표는 `@id` 한 칸이라 축을 안 낸다 continue @@ -174,13 +178,13 @@ def test_줄_열쇠가_서버_축과_같다(labels): def test_구간_이름은_원문_표기_그대로다(): - """⚠ `~1.5`·`7이상` 을 예쁘게 고치면 줄 열쇠가 바뀌어 덮개가 주인을 잃는다.""" + """⚠ `~1.5`·`7이상` 을 예쁘게 고치면 줄 열쇠가 바뀌어 주입한 줄을 못 알아본다.""" for kind in ("masonry_slope", "masonry_back_length"): - brackets = {str(r.get("height_bracket")) for r in base_prices.rows(kind)} + brackets = {str(r.get("height_bracket")) for r in base_prices.base_rows(kind)} assert brackets == {"~1.5", "~3", "~5", "~7", "7이상"}, (kind, sorted(brackets)) classes = { str(r.get("class")) - for r in base_prices.rows("masonry_class") + for r in base_prices.base_rows("masonry_class") if r.get("table") == "boulder_diameter" } assert classes == {"40~60", "60~80", "80~100"}, sorted(classes) @@ -198,13 +202,13 @@ def test_축_칸은_못_고친다(): """축을 고치면 줄이 다른 줄이 된다 — 서버가 잠가야 한다.""" bad = [] for kind in base_prices.KINDS: - rows = base_prices.rows(kind) + rows = base_prices.base_rows(kind) axes = set() for row in rows: axes |= set(row.get("@axis") or ()) if not axes: continue - meta = base_prices.table(kind, page=1, size=1) + meta = base_prices.table(kind, rows, page=1, size=1) for axis in axes: if axis in meta["editable"] and f"{kind}/{axis}" not in AXIS_ALSO_VALUE: bad.append(f"{kind}/{axis} 가 고칠 수 있는 칸으로 섰다") @@ -228,10 +232,11 @@ def test_기초단가_상자_이름이_이름표에_있다(labels): assert named["pumsem_basis"]["name_ko"] == "품셈 기준" -def test_상자마다_kind_가_있다(): +def test_상자마다_kind_가_있다(monkeypatch): """빈 상자가 서면 화면에 이름만 뜨고 아무것도 안 열린다.""" + install(monkeypatch) counted: dict[str, int] = {} - for entry in base_prices.kinds(): + for entry in asyncio.run(base_prices.kinds()): counted[entry["group"]] = counted.get(entry["group"], 0) + 1 assert entry["label"] != entry["kind"], f"{entry['kind']} 이름이 영문 그대로다" assert entry["rows"] > 0, entry["kind"] diff --git a/resources/tester/test_z01_base_prices.py b/resources/tester/test_z01_base_prices.py index 911b90fc..1af960b9 100644 --- a/resources/tester/test_z01_base_prices.py +++ b/resources/tester/test_z01_base_prices.py @@ -1,23 +1,27 @@ -"""Z01 기초단가 다섯(노임·기계·자재·유가·요율) — 고칠 수 있게 · 2026-09-15 브레인 계약(사용자 지시). +"""Z01 기초단가 다섯(노임·기계·자재·유가·요율) — DB 정본 + [저장] 일괄 저장(2026-09-18 PLAN 1-0 · 1-2). -틀(브레인 못박음): 원본 `resources/data_*` 는 **읽기만** · 고친 값은 덮개 -`resources/data_master_override/{kind}.json`(주기가 달라 kind 마다 한 파일)에만 · 읽을 때 원본 위에 얹음 · -덮개 줄마다 original(원래 값)·source(원본 파일) — 원본이 바뀌면 「원본 바뀜」 · 줄이 사라지면 「주인 없음」(버리지 않음) · +틀: 정본 = DB `master_base_price`(data = 지금 값 · seed = 초기값) · 원본 파일 `resources/data_*` 는 **주입 재료로 읽기만** · +[저장] 한 번 = {base_revision, edits, added, deleted} → 판이 다르면 409 · 하나라도 틀리면 아무것도 안 씀 · +줄마다 changed_columns(초기값과 다른 칸)·is_added(관리자 추가 줄) · [초기값으로] = data := seed(추가 줄은 지움). 기계 시간당 단가는 계산값 — editable 에서 빼고 왜 못 고치나(locked) · 밑값(취득가·손료계수·노임·유가)을 고치면 따라 바뀜. +DB 는 `helper_z01_fake_repo` 가짜로 — 저장소 SQL 흐름(잠금·되돌림)은 맨 아래 가짜 커넥션으로 잼. """ from __future__ import annotations +import asyncio import hashlib import json from decimal import Decimal from pathlib import Path +from types import SimpleNamespace import pytest from fastapi import FastAPI from fastapi.testclient import TestClient +from helper_z01_fake_repo import FakeRepo, install -from Z01_MasterData import Z01_MasterData_Overrides as overrides +from Z01_MasterData import Z01_MasterData_Repository_BasePrices as repo from Z01_MasterData import Z01_MasterData_Router as router_module ROOT = Path(__file__).resolve().parents[2] @@ -36,14 +40,12 @@ SOURCES = [ @pytest.fixture -def store(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path: - folder = tmp_path / "data_master_override" - monkeypatch.setattr(overrides, "OVERRIDE_DIR", folder) - return folder +def fake(monkeypatch: pytest.MonkeyPatch) -> FakeRepo: + return install(monkeypatch) @pytest.fixture -def client(store: Path) -> TestClient: +def client(fake: FakeRepo) -> TestClient: from common_util.common_util_auth import verify_session app = FastAPI() @@ -63,8 +65,20 @@ def _row(client: TestClient, kind: str, row_id: str) -> dict: return next(r for r in rows if r["@id"] == row_id) -def _put(client: TestClient, kind: str, row_id: str, values: dict): - return client.put(f"/api/master-data/base-prices/{kind}/{row_id}", json={"values": values}) +def _save(client: TestClient, kind: str, edits=(), added=(), deleted=(), base=None): + if base is None: + base = _get(client, kind, size=1)["revision"] + body = { + "base_revision": base, + "edits": [{"row_key": k, "column": c, "value": v} for k, c, v in edits], + "added": [{"data": d} for d in added], + "deleted": list(deleted), + } + return client.put(f"/api/master-data/base-prices/{kind}", json=body) + + +def _edit(client: TestClient, kind: str, row_id: str, column: str, value): + return _save(client, kind, edits=[(row_id, column, value)]) def _sha(path: Path) -> str: @@ -81,13 +95,15 @@ def test_다섯_표는_제_열쇠와_고칠_칸을_냄(client: TestClient) -> No for kind, (total, editable) in expected.items(): table = _get(client, kind, size=500) assert table["total"] == total and table["editable"] == editable, kind + assert table["status"] == "success" and table["revision"] == 0, kind ids = [r["@id"] for r in _get(client, kind, size=500)["rows"]] assert len(ids) == len(set(ids)), kind - assert all("@overrides" not in r for r in table["rows"]), ( - kind - ) # 안 고친 줄엔 칸 자체가 없음 + # 안 고친 줄 — 고친 칸 없음 · 추가 줄 아님 · 안쪽 칸(@data 따위)은 안 나감 + assert all(r["changed_columns"] == [] and r["is_added"] is False for r in table["rows"]) + assert not {k for r in table["rows"] for k in r if k in ("@data", "@added", "@changed")} columns = {c["key"] for c in table["columns"]} assert not {k for k in columns if k.startswith("@")}, kind # 줄 수준 칸은 열 목록 밖 + assert not columns & {"changed_columns", "is_added"}, kind assert set(table["locked"]) <= columns and not set(table["locked"]) & set(editable), kind assert _row(client, "labor", "labor_const/1001")["daily_wage_krw"] == 215907 assert _row(client, "labor", "labor_mfg/1")["occupation_name"] == "CAD설계사(기계)" @@ -116,8 +132,8 @@ def test_기계는_시간당_단가를_계산값으로_잠그고_산출근거를 "hourly_total_krw" in row["@formula"] and f"{int(row['hourly_total_krw']):,}" in row["@formula"]["hourly_total_krw"] ) - res = _put(client, "machine", "0101-0007", {"hourly_total_krw": 1}) - assert res.status_code == 400 and table["locked"]["hourly_total_krw"] in res.text + res = _edit(client, "machine", "0101-0007", "hourly_total_krw", 1) + assert res.status_code == 400 and table["locked"]["hourly_total_krw"] in res.json()["message"] def test_기계_계산값은_B09_시간당_사용료와_같음(client: TestClient) -> None: @@ -146,38 +162,23 @@ def test_기계_계산값은_B09_시간당_사용료와_같음(client: TestClien assert checked >= 5 -def test_고친_값은_덮개에만_쓰고_원본은_안_건드림(client: TestClient, store: Path) -> None: +def test_저장은_DB_에만_쓰고_원본_파일은_안_건드림(client: TestClient, fake: FakeRepo) -> None: before = {p: _sha(p) for p in SOURCES} - res = _put(client, "labor", "labor_const/1001", {"daily_wage_krw": 230000}) + res = _save( + client, + "labor", + edits=[("labor_const/1001", "daily_wage_krw", 230000)], + base=0, + ) assert res.status_code == 200, res.text - row = res.json() - assert row["daily_wage_krw"] == 230000 - assert row["@overrides"] == { - "daily_wage_krw": { - "original": 215907, - "value": 230000, - "state": overrides.EDITED, - "current": 215907, - } - } - assert ( - _row(client, "labor", "labor_const/1001")["@overrides"]["daily_wage_krw"]["value"] == 230000 - ) - assert ( - _put(client, "oil", "national_average/oil_diesel", {"price_krw_per_l": 1900}).status_code - == 200 - ) - assert sorted(p.name for p in store.iterdir()) == [ - "labor.json", - "oil.json", - ] # kind 마다 한 파일 - saved = json.loads((store / "labor.json").read_text(encoding="utf-8")) - entry = saved["overrides"]["labor_const/1001"]["daily_wage_krw"] - assert ( - entry["original"] == 215907 - and entry["source"] == "labor_const_2026-01-01.json" - and entry["by"] == 42 - ) + assert res.json() == {"status": "success", "revision": 1, "added": []} + row = _row(client, "labor", "labor_const/1001") + assert row["daily_wage_krw"] == 230000 and row["changed_columns"] == ["daily_wage_krw"] + stored = fake.by_key("labor")["labor_const/1001"] + assert stored["data"]["daily_wage_krw"] == 230000 and stored["seed"]["daily_wage_krw"] == 215907 + assert fake.saved_by == [42] # 누가 고쳤나 + assert _get(client, "labor", size=1)["revision"] == 1 + assert _get(client, "oil", size=1)["revision"] == 0 # 판번호는 표마다 assert {p: _sha(p) for p in SOURCES} == before @@ -185,139 +186,113 @@ def test_밑값을_고치면_기계_계산값이_따라_바뀜(client: TestClien base = _row(client, "machine", "0101-0007") operator = base["operator_occupation_code"] wage = _row(client, "labor", f"labor_const/{operator}")["daily_wage_krw"] - _put(client, "labor", f"labor_const/{operator}", {"daily_wage_krw": wage + 80000}) - _put(client, "oil", "national_average/oil_diesel", {"price_krw_per_l": 2000}) - _put(client, "machine", "0101-0007", {"price_thousand_krw": base["price_thousand_krw"] * 2}) + assert _edit( + client, "labor", f"labor_const/{operator}", "daily_wage_krw", wage + 80000 + ).is_success + assert _edit(client, "oil", "national_average/oil_diesel", "price_krw_per_l", 2000).is_success + price = base["price_thousand_krw"] + assert _edit(client, "machine", "0101-0007", "price_thousand_krw", price * 2).is_success after = _row(client, "machine", "0101-0007") assert after["hourly_operator_krw"] > base["hourly_operator_krw"] assert after["hourly_fuel_krw"] > base["hourly_fuel_krw"] assert after["hourly_loss_krw"] == pytest.approx(base["hourly_loss_krw"] * 2) - price = base["price_thousand_krw"] - assert after["@overrides"] == { - "price_thousand_krw": { - "original": price, - "value": price * 2, - "state": overrides.EDITED, - "current": price, - } - } # 노임·유가 덮개는 그 표 줄에 서고 기계 줄은 계산만 바뀜 + # 노임·유가 고친 칸은 그 표 줄에 서고 기계 줄은 계산만 바뀜 + assert after["changed_columns"] == ["price_thousand_krw"] -def test_null_은_되돌리기_같은_값도_덮개를_안_남김(client: TestClient, store: Path) -> None: - _put(client, "material", "10023392", {"price_krw": 30}) - reverted = _put(client, "material", "10023392", {"price_krw": None}).json() - assert reverted["price_krw"] == 22 and "@overrides" not in reverted - assert _put(client, "material", "10023392", {"price_krw": 22}).status_code == 200 - saved = json.loads((store / "material.json").read_text(encoding="utf-8")) - assert saved["overrides"] == {} +def test_계산_칸은_DB_에_안_씀(client: TestClient, fake: FakeRepo) -> None: + """기계 줄을 고쳐도 저장 값은 입력 칸만 — 계산 칸이 굳어 들어가면 노임을 고쳐도 안 따라옴.""" + _edit(client, "machine", "0101-0007", "loss_coefficient_per_hour", 0.001) + data = fake.by_key("machine")["0101-0007"]["data"] + assert data["loss_coefficient_per_hour"] == 0.001 + assert not {"hourly_total_krw", "hourly_note", "fuel_price_krw_per_l"} & set(data) -def test_원본이_바뀐_덮개는_덮개_값을_쓰고_세게_드러냄_주인_없는_덮개도_남김( - client: TestClient, store: Path +def test_하나라도_틀리면_아무것도_안_씀(client: TestClient, fake: FakeRepo) -> None: + good = ("labor_const/1001", "daily_wage_krw", 230000) + for bad, status in ( + (("labor_const/1002", "occupation_name", "x"), 400), # 잠긴 칸 + (("labor_const/1002", "daily_wage_krw", -1), 400), # 나쁜 값 + (("labor_const/0000", "daily_wage_krw", 1), 404), # 없는 줄 + ): + res = _save(client, "labor", edits=[good, bad]) + assert res.status_code == status, (bad, res.text) + assert res.json()["status"] == "error" + res = _save(client, "labor", edits=[good], deleted=["labor_const/0000"]) + assert res.status_code == 404 + assert fake.revisions == {} and fake.saved_by == [] + assert _row(client, "labor", "labor_const/1001")["daily_wage_krw"] == 215907 + + +def test_낡은_판으로_저장하면_409( + client: TestClient, fake: FakeRepo, monkeypatch: pytest.MonkeyPatch ) -> None: - store.mkdir() - (store / "labor.json").write_text( - json.dumps( - { - "schema_version": "1.0", - "kind": "labor", - "overrides": { - "labor_const/1001": { - "daily_wage_krw": { - "value": 199000, - "original": 200000, - "source": "labor_const_2025-07-01.json", - } - }, - "labor_const/9999": { - "daily_wage_krw": { - "value": 1, - "original": 1, - "source": "labor_const_2025-07-01.json", - } - }, - }, - } - ), - encoding="utf-8", - ) - # 브레인 ③ — 자동으로 원본을 쓰면 금액이 말없이 바뀜 → 사람이 고친 값 유지 · 갈렸다는 것만 또렷이 - row = _row(client, "labor", "labor_const/1001") - assert row["daily_wage_krw"] == 199000 - mark = row["@overrides"]["daily_wage_krw"] - assert mark == { - "original": 200000, - "value": 199000, - "state": overrides.SOURCE_CHANGED, - "current": 215907, - } - _put(client, "labor", "labor_const/1002", {"daily_wage_krw": 170000}) - listed = client.get("/api/master-data/overrides").json() - assert [ - (i["row_id"], i["state"]) for i in listed["items"] - ] == [ # 서버가 정렬 — 원본 바뀜이 맨 위 - ("labor_const/1001", overrides.SOURCE_CHANGED), - ("labor_const/9999", overrides.ORPHAN), - ("labor_const/1002", overrides.EDITED), - ] - first = listed["items"][0] - assert set(first) == { - "kind", - "row_id", - "row_label", - "column", - "column_label", - "original", - "current", - "value", - "state", - "unit", - } - assert first["row_label"] == "작업반장" and ( - first["original"], - first["current"], - first["value"], - ) == (200000, 215907, 199000) - assert listed["items"][1]["current"] is None - saved = json.loads((store / "labor.json").read_text(encoding="utf-8"))["overrides"] - assert "labor_const/9999" in saved # 다른 줄을 고쳐도 주인 없는 덮개는 안 지움 - # 같은 값을 다시 넣으면 「지금 원본을 보고 고침」 으로 굳음 · null 이면 새 원본 값으로 받기 - kept = _put(client, "labor", "labor_const/1001", {"daily_wage_krw": 199000}).json() - assert kept["@overrides"]["daily_wage_krw"] == { - "original": 215907, - "value": 199000, - "state": overrides.EDITED, - "current": 215907, - } - taken = _put(client, "labor", "labor_const/1001", {"daily_wage_krw": None}).json() - assert taken["daily_wage_krw"] == 215907 and "@overrides" not in taken + assert _save( + client, "labor", edits=[("labor_const/1001", "daily_wage_krw", 1)], base=0 + ).is_success + stale = _save(client, "labor", edits=[("labor_const/1002", "daily_wage_krw", 2)], base=0) + assert stale.status_code == 409 and "다시 불러와" in stale.json()["message"] + assert _row(client, "labor", "labor_const/1002")["changed_columns"] == [] + # 앞 검사를 지난 뒤 트랜잭션 안에서 판이 갈려도 409(저장소가 잠그고 다시 봄) + + async def old_revision(kind: str) -> int: + return 0 + + monkeypatch.setattr(repo, "revision", old_revision) + raced = _save(client, "labor", edits=[("labor_const/1002", "daily_wage_krw", 2)], base=0) + assert raced.status_code == 409 and fake.revisions["labor"] == 1 -def test_빈_칸에_새로_넣은_값도_덮개_원본에_값이_생기면_원본_바뀜( - client: TestClient, store: Path -) -> None: - blank = next(r for r in _get(client, "labor", size=500)["rows"] if r["daily_wage_krw"] is None) - row = _put(client, "labor", blank["@id"], {"daily_wage_krw": 150000}).json() - assert row["@overrides"]["daily_wage_krw"] == { - "original": None, - "value": 150000, - "state": overrides.EDITED, - "current": None, - } - saved = json.loads((store / "labor.json").read_text(encoding="utf-8")) - saved["overrides"][blank["@id"]]["daily_wage_krw"]["original"] = ( - 140000 # 옛 판엔 값이 있었다고 치면 +def test_줄_추가_고치기_지우기(client: TestClient, fake: FakeRepo) -> None: + data = {"specification": "ZZ시험 규격, 특수", "unit": "개", "price_krw": 1234} + res = _save(client, "material", added=[data]) + assert res.status_code == 200, res.text + (key,) = res.json()["added"] + assert key.startswith("added/") and res.json()["revision"] == 1 + table = _get(client, "material", size=1, page=7000) + assert table["total"] == 7000 + row = table["rows"][0] # 추가 줄은 맨 끝(넣은 차례) + assert (row["@id"], row["is_added"], row["changed_columns"]) == (key, True, []) + assert row["price_krw"] == 1234 and row["specification"] == "ZZ시험 규격, 특수" + assert _get(client, "material", q="ZZ시험")["total"] == 1 + # 추가 줄은 글자 칸도 고침(사람이 넣은 이름) · 원래 줄의 글자 칸은 못 고침 + assert _edit(client, "material", key, "specification", "시험 규격, 고침").is_success + assert _edit(client, "material", "10023392", "specification", "x").status_code == 400 + for bad in ({"없는칸": 1}, {"price_krw": "비쌈"}, {"price_krw": -5}, {"unit": {"a": 1}}): + assert _save(client, "material", added=[bad]).status_code == 400, bad + assert _save(client, "material", added=[{}]).status_code == 422 # 빈 줄 + gone = _save(client, "material", deleted=[key, "10023392"]) + assert gone.status_code == 200 + ids = {r["@id"] for r in _get(client, "material", q="육각볼트", size=500)["rows"]} + assert "10023392" not in ids and _get(client, "material", size=1)["total"] == 6998 + assert fake.by_key("material")["10023392"]["deleted"] is True # 지운 표시만(되살릴 수 있게) + + +def test_초기값으로_되돌리기(client: TestClient, fake: FakeRepo) -> None: + _edit(client, "labor", "labor_const/1001", "daily_wage_krw", 230000) + added = _save(client, "labor", added=[{"occupation_name": "시험 직종", "daily_wage_krw": 1}]) + (key,) = added.json()["added"] + _save(client, "labor", deleted=["labor_const/1002"]) + res = client.post( + "/api/master-data/base-prices/labor/reset", + json={"row_keys": ["labor_const/1001", key, "labor_const/1002"]}, ) - (store / "labor.json").write_text(json.dumps(saved), encoding="utf-8") - assert ( - _row(client, "labor", blank["@id"])["@overrides"]["daily_wage_krw"]["state"] - == overrides.SOURCE_CHANGED + assert res.status_code == 200 and res.json() == {"status": "success", "revision": 4} + back = _row(client, "labor", "labor_const/1001") + assert back["daily_wage_krw"] == 215907 and back["changed_columns"] == [] + assert _row(client, "labor", "labor_const/1002")["daily_wage_krw"] # 지운 줄도 되살림 + assert _get(client, "labor", size=1)["total"] == 261 # 추가 줄은 지움 + missing = client.post( + "/api/master-data/base-prices/labor/reset", json={"row_keys": ["labor_const/0000"]} ) + assert missing.status_code == 404 and "labor_const/0000" in missing.json()["message"] + empty = client.post("/api/master-data/base-prices/labor/reset", json={"row_keys": []}) + assert empty.status_code == 422 # 통째로 되돌리기 막음 def test_요율_표는_법정값_경고를_실음(client: TestClient) -> None: rate = _get(client, "rate", size=1) - assert any("갱신" in line and "주인" in line for line in rate["notice"]) + assert any("법이 정한 값" in line and "고시" in line for line in rate["notice"]) for kind in ("labor", "machine", "material", "oil", "rate"): # 브레인 ③ — 문구는 서버가 한 벌로 notice = _get(client, kind, size=1)["notice"] assert any("새로 만드는 프로젝트" in line and "이미 만든" in line for line in notice), kind @@ -325,16 +300,18 @@ def test_요율_표는_법정값_경고를_실음(client: TestClient) -> None: def test_막는_자리(client: TestClient) -> None: - assert _put(client, "wage", "x", {"daily_wage_krw": 1}).status_code == 404 - assert _put(client, "labor", "labor_const/0000", {"daily_wage_krw": 1}).status_code == 404 - assert _put(client, "labor", "labor_const/1001", {"occupation_name": "x"}).status_code == 400 - locked = _put(client, "labor", "labor_const/1001", {"hours_per_day": 9}) # 숫자여도 잠긴 칸 - assert locked.status_code == 400 and "공표 원문" in locked.text + assert _save(client, "wage", base=0).status_code == 404 + assert _save(client, "work_item_forest", base=0).status_code == 400 # 공종은 /work-items 몫 + locked = _edit(client, "labor", "labor_const/1001", "hours_per_day", 9) # 숫자여도 잠긴 칸 + assert locked.status_code == 400 and "공표 원문" in locked.json()["message"] for bad in (True, -1, "220000", float("nan")): - body = json.dumps({"values": {"daily_wage_krw": bad}}).encode() + body = { + "base_revision": 0, + "edits": [{"row_key": "labor_const/1001", "column": "daily_wage_krw", "value": bad}], + } res = client.put( - "/api/master-data/base-prices/labor/labor_const/1001", - content=body, + "/api/master-data/base-prices/labor", + content=json.dumps(body).encode(), headers={"content-type": "application/json"}, ) assert res.status_code in (400, 422), bad @@ -344,9 +321,20 @@ def test_막는_자리(client: TestClient) -> None: if r["@id"].startswith("rate_performance_guarantee_fee/brackets/") ) assert ( - _put(client, "rate", fee["@id"], {"rate_percent": 1}).status_code == 400 - ) # 그 줄엔 요율 칸이 없음 + _edit(client, "rate", fee["@id"], "rate_percent", 1).status_code == 400 + ) # 그 줄엔 요율 칸 없음 assert Decimal(str(_row(client, "rate", "rate_sanjae")["rate_percent"])) == Decimal("3.56") + many = [("labor_const/1001", "daily_wage_krw", 1)] * 501 + assert _save(client, "labor", edits=many).status_code == 422 # 한 번에 500 칸까지 + extra = client.put( + "/api/master-data/base-prices/labor", json={"base_revision": 0, "values": {}} + ) + assert extra.status_code == 422 # 모르는 칸은 거절(옛 모양으로 보내면 조용히 무시 안 함) + old = client.put( + "/api/master-data/base-prices/labor/labor_const/1001", json={"values": {"x": 1}} + ) + assert old.status_code in (404, 405) # 옛 칸 단위 저장은 없어짐 + assert client.get("/api/master-data/overrides").status_code == 404 def test_합친_표는_줄마다_판_기준일을_실음(client: TestClient) -> None: @@ -364,61 +352,6 @@ def test_합친_표는_줄마다_판_기준일을_실음(client: TestClient) -> assert no_record["operating_effective_date"] is None and no_record["hourly_total_krw"] is None -def test_주인_없는_덮개는_모아서_한꺼번에_다룸(client: TestClient, store: Path) -> None: - """브레인 ② — 자재·기계는 품목 퇴출이 상시라 갱신 한 번에 여럿이 주인을 잃음.""" - store.mkdir() - gone = { - f"9999000{i}": {"price_krw": {"value": 1, "original": 1, "source": "old.json"}} - for i in range(3) - } - (store / "material.json").write_text( - json.dumps({"schema_version": "1.0", "kind": "material", "overrides": gone}), - encoding="utf-8", - ) - _put(client, "material", "10023392", {"price_krw": 30}) - _put(client, "labor", "labor_const/1001", {"daily_wage_krw": 230000}) - listed = client.get( - "/api/master-data/overrides", params={"state": overrides.ORPHAN, "kind": "material"} - ).json() - assert listed["total"] == 3 and {i["state"] for i in listed["items"]} == {overrides.ORPHAN} - assert listed["counts"] == { - "material": {overrides.ORPHAN: 3, overrides.EDITED: 1}, - "labor": {overrides.EDITED: 1}, - } - paged = client.get("/api/master-data/overrides", params={"size": 2, "page": 2}).json() - assert paged["total"] == 5 and len(paged["items"]) == 2 - everything = client.get("/api/master-data/overrides").json()[ - "items" - ] # labor 가 kind 차례로 앞이어도 - assert [i["state"] for i in everything] == [overrides.ORPHAN] * 3 + [overrides.EDITED] * 2 - res = client.post( - "/api/master-data/overrides/clear", json={"kind": "material", "state": overrides.ORPHAN} - ) - assert res.status_code == 200 and res.json()["removed"] == 3 - left = client.get("/api/master-data/overrides").json() - assert [(i["kind"], i["state"]) for i in left["items"]] == [ # 같은 state 안은 kind 차례 - ("labor", overrides.EDITED), - ("material", overrides.EDITED), - ] - assert ( - client.post("/api/master-data/overrides/clear", json={"kind": "material"}).status_code - == 422 - ) # 거름 없이 통째로 못 비움 - - -def test_줄은_있는데_칸이_사라진_덮개도_주인_없음(client: TestClient, store: Path) -> None: - rates = _get(client, "rate", size=500)["rows"] - fee = next(r for r in rates if r["@id"].startswith("rate_performance_guarantee_fee/brackets/")) - store.mkdir() - entry = {"rate_percent": {"value": 1, "original": 0.5, "source": "old.json"}} - (store / "rate.json").write_text( - json.dumps({"overrides": {fee["@id"]: entry}}), encoding="utf-8" - ) - items = client.get("/api/master-data/overrides", params={"kind": "rate"}).json()["items"] - assert [(i["row_id"], i["state"]) for i in items] == [(fee["@id"], overrides.ORPHAN)] - assert "@overrides" not in _row(client, "rate", fee["@id"]) # 없는 칸을 새로 만들지 않음 - - SOURCE_KEYS = { "source_id", "name", @@ -476,12 +409,6 @@ def test_우리_판은_출처표_글자가_아니라_파일에서_읽음( assert _get(client, "rate", size=1)["source"] == [] # 출처표가 없으면 빈 목록(모양은 그대로) -def test_모아_보기_항목에_단위(client: TestClient) -> None: - _put(client, "labor", "labor_const/1001", {"daily_wage_krw": 230000}) - (item,) = client.get("/api/master-data/overrides").json()["items"] - assert item["unit"] == "원" and item["column_label"] == "일 노임" - - def test_열_이름은_이름표에서만_옴(client: TestClient) -> None: """브레인 ② — 이름·판정은 한 곳에만. 합친 표 열 이름은 이름표 `merged_tables` 가 정본.""" labels = json.loads( @@ -540,8 +467,8 @@ def test_빈_칸은_어느_쪽으로_세워도_끝(client: TestClient) -> None: assert hourly[0] is not None and hourly[-1] is None # 계산 열도 세워짐 -def test_덮개_얹은_값으로_세움(client: TestClient) -> None: - _put(client, "labor", "labor_const/1003", {"daily_wage_krw": 9_000_000}) +def test_고친_값으로_세움(client: TestClient) -> None: + _edit(client, "labor", "labor_const/1003", "daily_wage_krw", 9_000_000) top = _get(client, "labor", sort="daily_wage_krw", desc=1, size=1)["rows"][0] assert top["@id"] == "labor_const/1003" and top["daily_wage_krw"] == 9_000_000 @@ -551,54 +478,107 @@ def test_sort_가_없으면_원본_차례(client: TestClient) -> None: assert _get(client, "machine", size=1)["rows"][0]["@id"] == "0101-0007" -def test_줄의_뜻이_바뀌면_넷째_낱말(client: TestClient, store: Path) -> None: - """브레인 — 같은 코드에 다른 기준이 붙음(노임 3001~ 문화재→국가유산 · 자재 KS 개정 규격). - - 값만 보면 안 잡힘 → 덮개를 쓸 때 **값이 아닌 글자 칸**(이름·규격·단위…)을 함께 적어 두고 견줌. - """ - saved = _put(client, "labor", "labor_const/1001", {"daily_wage_krw": 230000}).json() - assert saved["@overrides"]["daily_wage_krw"]["state"] == overrides.EDITED - entry = json.loads((store / "labor.json").read_text(encoding="utf-8"))["overrides"] - meaning = entry["labor_const/1001"]["daily_wage_krw"]["meaning"] - assert meaning["occupation_name"] == "작업반장" # 이름·규격 같은 글자 칸만 - assert "daily_wage_krw" not in meaning and "hours_per_day" not in meaning # 값 칸은 안 적음 - assert "effective_date" not in meaning # 판 날짜는 갱신마다 바뀜 — 뜻 바뀜과 구별이 안 됨 - _put(client, "oil", "national_average/oil_diesel", {"price_krw_per_l": 1900}) - oil = json.loads((store / "oil.json").read_text(encoding="utf-8"))["overrides"] - assert "date" not in oil["national_average/oil_diesel"]["price_krw_per_l"]["meaning"] - entry["labor_const/1001"]["daily_wage_krw"]["meaning"]["occupation_name"] = "옛 직종 이름" - (store / "labor.json").write_text( - json.dumps({"schema_version": "1.0", "kind": "labor", "overrides": entry}), encoding="utf-8" - ) - mark = _row(client, "labor", "labor_const/1001")["@overrides"]["daily_wage_krw"] - assert mark["state"] == overrides.MEANING_CHANGED # 값은 그대로인데 줄의 뜻이 바뀜 - # 이름 칸이 여럿인 자료가 있어 **서버가 한 줄로 합쳐** 냄(화면이 합치면 판정이 두 벌) - assert mark["was"] == "labor_const · 1001 · 옛 직종 이름 · KRW/day · published" - assert mark["now"] == "labor_const · 1001 · 작업반장 · KRW/day · published" - listed = client.get("/api/master-data/overrides").json() - top = listed["items"][0] - assert top["state"] == overrides.MEANING_CHANGED # 값 틀림보다 더 세게 - assert (top["was"], top["now"]) == (mark["was"], mark["now"]) # 모아 보기에도 같은 둘 - assert overrides.STATE_ORDER[0] == overrides.MEANING_CHANGED - assert all(i.get("was") for i in listed["items"] if i["state"] == overrides.MEANING_CHANGED) +def test_추가한_기계_줄도_셈이_섬(client: TestClient) -> None: + """카탈로그에 없는 기종 — 제원은 줄 값에서 · 운전경비 원단위가 없으니 손료만 서고 합계는 안 냄.""" + data = { + "machine_name": "시험 굴삭기", + "price_thousand_krw": 100000, + "loss_coefficient_per_hour": 0.0001, + } + (key,) = _save(client, "machine", added=[data]).json()["added"] + row = _get(client, "machine", size=1, page=614)["rows"][0] + assert row["@id"] == key and row["hourly_loss_krw"] == pytest.approx(10000) + assert row["hourly_total_krw"] is None and row["hourly_note"] + assert _save(client, "machine", added=[{"hourly_total_krw": 1}]).status_code == 400 # 계산 칸 -def test_뜻_칸은_이름표가_가림_옛_덮개는_그대로(client: TestClient, store: Path) -> None: - _put(client, "machine", "0101-0007", {"price_thousand_krw": 80000}) - meaning = json.loads((store / "machine.json").read_text(encoding="utf-8"))["overrides"][ - "0101-0007" - ]["price_thousand_krw"]["meaning"] - assert meaning["machine_name"] == "불도저(무한궤도)" and meaning["specification"] == "7" - assert not [k for k in meaning if k in ("price_thousand_krw", "loss_coefficient_per_hour")] - store.mkdir( - exist_ok=True - ) # 뜻 칸이 없던 옛 덮개 — 그대로 edited 로 둠(없는 것을 지어내지 않음) - (store / "material.json").write_text( - json.dumps( - {"overrides": {"10023392": {"price_krw": {"value": 30, "original": 22, "source": "x"}}}} - ), - encoding="utf-8", - ) - assert ( - _row(client, "material", "10023392")["@overrides"]["price_krw"]["state"] == overrides.EDITED - ) +# ── 저장소 SQL 흐름 — 가짜 커넥션(DB 없이) ────────────────────────────── + + +class _Cursor: + def __init__(self, db: SimpleNamespace) -> None: + self.db = db + + async def __aenter__(self): + return self + + async def __aexit__(self, *_): + return False + + async def execute(self, sql: str, params=()) -> None: + self.db.sql.append(" ".join(sql.split())) + + async def fetchone(self): + return (self.db.revision,) + + async def fetchall(self): + return [(key,) for key in self.db.keys] + + +class _Connection: + def __init__(self, db: SimpleNamespace) -> None: + self.db = db + + def cursor(self, *_): + return _Cursor(self.db) + + async def begin(self) -> None: + self.db.log.append("begin") + + async def commit(self) -> None: + self.db.log.append("commit") + + async def rollback(self) -> None: + self.db.log.append("rollback") + + +class _Pool: + def __init__(self, db: SimpleNamespace) -> None: + self.db = db + + def acquire(self): + return self + + async def __aenter__(self): + return _Connection(self.db) + + async def __aexit__(self, *_): + return False + + +def _writes(db: SimpleNamespace) -> list[str]: + return [ + s + for s in db.sql + if s.startswith(("UPDATE master_base_price", "INSERT INTO master_base_price")) + ] + + +def test_저장소는_판을_잠그고_보며_틀리면_되돌림(monkeypatch: pytest.MonkeyPatch) -> None: + db = SimpleNamespace(sql=[], log=[], revision=4, keys=[]) + monkeypatch.setattr(repo, "get_db_pool", lambda: _Pool(db)) + with pytest.raises(repo.RevisionConflict): + asyncio.run(repo.save("labor", 3, {"a": {"x": 1}}, {}, ["b"], 7)) + assert db.log == ["begin", "rollback"] and not _writes(db) + db.sql.clear() + db.log.clear() + assert asyncio.run(repo.save("labor", 4, {"a": {"x": 1}}, {"added/1": {"y": 2}}, ["b"], 7)) == 5 + assert db.log == ["begin", "commit"] + lock = next(i for i, s in enumerate(db.sql) if s.endswith("FOR UPDATE")) + bump = next(i for i, s in enumerate(db.sql) if s.startswith("UPDATE master_revision")) + writes = [db.sql.index(s) for s in _writes(db)] + assert len(writes) == 3 and lock < min(writes) and max(writes) < bump == len(db.sql) - 1 + assert "SET is_deleted = 1" in _writes(db)[-1] # 지우기는 표시만 + + +def test_저장소_되돌리기는_없는_줄이면_아무것도_안_씀(monkeypatch: pytest.MonkeyPatch) -> None: + db = SimpleNamespace(sql=[], log=[], revision=2, keys=["a"]) + monkeypatch.setattr(repo, "get_db_pool", lambda: _Pool(db)) + with pytest.raises(repo.UnknownRows): + asyncio.run(repo.reset("labor", ["a", "없음"], 7)) + assert db.log == ["begin", "rollback"] and not _writes(db) + db.sql.clear() + db.log.clear() + assert asyncio.run(repo.reset("labor", ["a", "a"], 7)) == 3 + assert db.log == ["begin", "commit"] + assert [s.split(" SET ")[1][:13] for s in _writes(db)] == ["data = seed, ", "is_deleted = "] diff --git a/resources/tester/test_z01_base_tables.py b/resources/tester/test_z01_base_tables.py index 88b17725..e0bf710b 100644 --- a/resources/tester/test_z01_base_tables.py +++ b/resources/tester/test_z01_base_tables.py @@ -19,9 +19,9 @@ ROOT = Path(__file__).resolve().parents[2] import pytest from fastapi import FastAPI from fastapi.testclient import TestClient +from helper_z01_fake_repo import install from Z01_MasterData import Z01_MasterData_BasePrices as base_prices -from Z01_MasterData import Z01_MasterData_Overrides as overrides from Z01_MasterData import Z01_MasterData_Router as router_module from Z01_MasterData import Z01_MasterData_WorkItems as work_items @@ -40,10 +40,10 @@ NEW_KINDS = ( @pytest.fixture -def client(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> TestClient: +def client(monkeypatch: pytest.MonkeyPatch) -> TestClient: from common_util.common_util_auth import verify_session - monkeypatch.setattr(overrides, "OVERRIDE_DIR", tmp_path / "data_master_override") + install(monkeypatch) # DB 가짜 — 파일 원본을 초기값으로 주입한 모양 app = FastAPI() app.include_router(router_module.router) app.dependency_overrides[verify_session] = lambda: {"user_id": 7, "role": "SYSTEM_ADMIN"} @@ -69,13 +69,13 @@ def test_아홉이_기초단가와_같은_길로_섬(client: TestClient) -> None table = _get(client, kind, size=500) assert table["total"] == len(table["rows"]) > 0, kind ids = _ids(table) - assert len(ids) == len(set(ids)), kind # ⭐ 열쇠가 겹치면 덮개가 엉뚱한 줄에 붙음 + assert len(ids) == len(set(ids)), kind # ⭐ 열쇠가 겹치면 (kind, row_key) 한 줄이 둘을 덮음 assert set(table["editable"]) <= {c["key"] for c in table["columns"]}, kind assert table["editable"] or kind == "rebar_complexity", kind # 철근은 값이 다 계산·글 assert set(table["locked"]) | set(table["editable"]) == set(table["sortable"]), kind axis = {a for r in base_prices.base_rows(kind) for a in r.get("@axis", ())} assert {table["locked"][c] for c in axis if c in table["locked"]} == { - "자료 열쇠 — 바꾸면 덮개·다른 표가 이 줄을 못 찾음" + "자료 열쇠 — 바꾸면 다른 표가 이 줄을 못 찾음" }, kind # 축 칸은 「왜 못 고치나」 가 원문 칸과 달라야 함 assert any("새로 만드는 프로젝트" in line for line in table["notice"]), kind assert table["source"] == [], kind # 출처표엔 아직 이 아홉이 없음 — 빈 목록(모양은 그대로) @@ -165,35 +165,39 @@ def test_아홉의_고칠_칸은_값_칸(client: TestClient) -> None: ) -def test_고치기도_같은_길(client: TestClient, tmp_path: Path) -> None: - res = client.put( - "/api/master-data/base-prices/stone_kind/wedge_stone_m3_per_m2/깬잡석/25", - json={"values": {"m3_per_m2": 0.1}}, +def _save(client: TestClient, kind: str, row_key: str, column: str, value): + revision = _get(client, kind, size=1)["revision"] + return client.put( + f"/api/master-data/base-prices/{kind}", + json={ + "base_revision": revision, + "edits": [{"row_key": row_key, "column": column, "value": value}], + }, ) + + +def test_고치기도_같은_길(client: TestClient) -> None: + res = _save(client, "stone_kind", "wedge_stone_m3_per_m2/깬잡석/25", "m3_per_m2", 0.1) assert res.status_code == 200, res.text - row = res.json() - assert row["m3_per_m2"] == 0.1 - assert row["@overrides"]["m3_per_m2"]["original"] == 0.09 - assert (tmp_path / "data_master_override" / "stone_kind.json").is_file() - listed = client.get("/api/master-data/overrides", params={"kind": "stone_kind"}).json() - assert [i["row_id"] for i in listed["items"]] == ["wedge_stone_m3_per_m2/깬잡석/25"] - blocked = client.put( - "/api/master-data/base-prices/stone_kind/wedge_stone_m3_per_m2/깬잡석/25", - json={"values": {"stone_kind": "딴 돌"}}, - ) + rows = {r["@id"]: r for r in _get(client, "stone_kind", size=500)["rows"]} + row = rows["wedge_stone_m3_per_m2/깬잡석/25"] + assert row["m3_per_m2"] == 0.1 and row["changed_columns"] == ["m3_per_m2"] + assert [k for k, r in rows.items() if r["changed_columns"]] == [ + "wedge_stone_m3_per_m2/깬잡석/25" + ] + blocked = _save(client, "stone_kind", "wedge_stone_m3_per_m2/깬잡석/25", "stone_kind", "딴 돌") assert blocked.status_code == 400 and "자료 열쇠" in blocked.text # 축 칸은 못 고침 - assert "@axis" not in row and "@source" not in row # 안쪽 칸은 안 내보냄 + assert not {"@axis", "@source", "@data"} & set(row) # 안쪽 칸은 안 내보냄 # 같은 칸이 어느 줄에선 값이고 어느 줄에선 축임(사용횟수) — 축인 줄에선 막혀야 함 - axis_edit = client.put( - "/api/master-data/base-prices/formwork_reuse/reuse_ratio_pct/plywood_material_pct/2", - json={"values": {"reuse_count": 3}}, + # ⚠ 주입 data 에 `@axis` 가 없어도 막힘(축은 파일 원본에서 읽음 · 가짜 주입은 '@' 칸을 뺌) + axis_edit = _save( + client, "formwork_reuse", "reuse_ratio_pct/plywood_material_pct/2", "reuse_count", 3 ) assert axis_edit.status_code == 400 and "자료 열쇠" in axis_edit.text - value_edit = client.put( - "/api/master-data/base-prices/formwork_reuse/reuse_by_class/복잡한 구조", - json={"values": {"reuse_count": 3}}, - ) - assert value_edit.status_code == 200 and value_edit.json()["reuse_count"] == 3 + value_edit = _save(client, "formwork_reuse", "reuse_by_class/복잡한 구조", "reuse_count", 3) + assert value_edit.status_code == 200 + reuse = {r["@id"]: r for r in _get(client, "formwork_reuse", size=500)["rows"]} + assert reuse["reuse_by_class/복잡한 구조"]["reuse_count"] == 3 def test_코드에_박혔던_기계_밑값이_마스터로_나옴(client: TestClient) -> None: diff --git a/resources/tester/test_z01_work_items.py b/resources/tester/test_z01_work_items.py index 5bce9ff6..ec12ea20 100644 --- a/resources/tester/test_z01_work_items.py +++ b/resources/tester/test_z01_work_items.py @@ -12,17 +12,17 @@ from pathlib import Path import pytest from fastapi import FastAPI from fastapi.testclient import TestClient +from helper_z01_fake_repo import install -from Z01_MasterData import Z01_MasterData_Overrides as overrides from Z01_MasterData import Z01_MasterData_Router as router_module from Z01_MasterData import Z01_MasterData_WorkItems as work_items @pytest.fixture -def client(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> TestClient: +def client(monkeypatch: pytest.MonkeyPatch) -> TestClient: from common_util.common_util_auth import verify_session - monkeypatch.setattr(overrides, "OVERRIDE_DIR", tmp_path / "data_master_override") + install(monkeypatch) # 기초단가 표 목록·줄 수가 DB 를 읽음 — 가짜로 app = FastAPI() app.include_router(router_module.router) app.dependency_overrides[verify_session] = lambda: {"user_id": 7, "role": "SYSTEM_ADMIN"} @@ -131,10 +131,15 @@ def test_불변_열쇠가_오면_id_로_씀_자료_파일은_판_id_로_가림( assert _get(client, "work_item_forest")["total"] == 0 # 산림 파일은 이 폴더에 없음 res = client.put( - "/api/master-data/base-prices/work_item_const/CW-00002%2301", - json={"values": {"name": "바꿈"}}, + "/api/master-data/base-prices/work_item_const", + json={ + "base_revision": 0, + "edits": [{"row_key": "CW-00002#01", "column": "name", "value": "바꿈"}], + }, ) - assert res.status_code == 400 and "공종 축 뼈대" in res.json()["detail"] + assert ( + res.status_code == 400 and "고칠 수 없는 표" in res.json()["message"] + ) # 공종은 /work-items 몫 def test_거르기는_자료에_있는_가름만_서버가_판정(client: TestClient) -> None: