eomsangdonandClaude Opus 5
6fe4056d98
- 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) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01V1MKKZKpUHTPKb513FneU8
585 lines
28 KiB
Python
585 lines
28 KiB
Python
"""Z01 기초단가 다섯(노임·기계·자재·유가·요율) — DB 정본 + [저장] 일괄 저장(2026-09-18 PLAN 1-0 · 1-2).
|
|
|
|
틀: 정본 = 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_Repository_BasePrices as repo
|
|
from Z01_MasterData import Z01_MasterData_Router as router_module
|
|
|
|
ROOT = Path(__file__).resolve().parents[2]
|
|
SOURCES = [
|
|
ROOT / "resources/data_cost_input_value" / name
|
|
for name in (
|
|
"labor_const_2026-01-01.json",
|
|
"labor_mfg_2026-07-01.json",
|
|
"mach_base_2026.json",
|
|
"mat_price_public_2026-08-14.json",
|
|
"oil_2026-08-14.json",
|
|
"oil_regional_2026-09-09.json",
|
|
"rates_2026.json",
|
|
)
|
|
]
|
|
|
|
|
|
@pytest.fixture
|
|
def fake(monkeypatch: pytest.MonkeyPatch) -> FakeRepo:
|
|
return install(monkeypatch)
|
|
|
|
|
|
@pytest.fixture
|
|
def client(fake: FakeRepo) -> TestClient:
|
|
from common_util.common_util_auth import verify_session
|
|
|
|
app = FastAPI()
|
|
app.include_router(router_module.router)
|
|
app.dependency_overrides[verify_session] = lambda: {"user_id": 42, "role": "SYSTEM_ADMIN"}
|
|
return TestClient(app)
|
|
|
|
|
|
def _get(client: TestClient, kind: str, **params) -> dict:
|
|
res = client.get(f"/api/master-data/base-prices/{kind}", params=params)
|
|
assert res.status_code == 200, res.text
|
|
return res.json()
|
|
|
|
|
|
def _row(client: TestClient, kind: str, row_id: str) -> dict:
|
|
rows = _get(client, kind, q=row_id.rsplit("/", 1)[-1], size=500)["rows"]
|
|
return next(r for r in rows if r["@id"] == row_id)
|
|
|
|
|
|
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:
|
|
return hashlib.sha256(path.read_bytes()).hexdigest()
|
|
|
|
|
|
def test_다섯_표는_제_열쇠와_고칠_칸을_냄(client: TestClient) -> None:
|
|
expected = {
|
|
"labor": (261, ["daily_wage_krw"]),
|
|
"material": (6999, ["price_krw"]),
|
|
"oil": (36, ["price_krw_per_l"]),
|
|
"machine": (613, ["price_thousand_krw", "loss_coefficient_per_hour"]),
|
|
}
|
|
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
|
|
# 안 고친 줄 — 고친 칸 없음 · 추가 줄 아님 · 안쪽 칸(@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설계사(기계)"
|
|
assert _row(client, "oil", "national_average/oil_diesel")["price_krw_per_l"] == 1846.39
|
|
assert _row(client, "oil", "regional/oil_diesel/01")["region_name"] == "서울"
|
|
rate = _get(client, "rate", size=500)
|
|
# 2026-09-18 218 → 220 — 전문공사 일반관리비 300억~1000억·1000억 이상 두 구간이 빠져 있던 것을 채움
|
|
assert rate["total"] == 220 and "rate_percent" in rate["editable"]
|
|
by_id = {r["@id"]: r for r in rate["rows"]}
|
|
assert by_id["rate_sanjae"]["rate_percent"] == 3.56
|
|
goyong = (
|
|
"rate_goyong/brackets/estimated_amount_bracket=gte_140_billion;grade=1" # 구간 칸 이름 차례
|
|
)
|
|
assert by_id[goyong]["rate_percent"] == 1.57 and by_id[goyong]["base"] == "total_labor_cost"
|
|
material = _get(client, "material", q="육각볼트", size=5)
|
|
assert 0 < material["total"] < 6999 and len(material["rows"]) == 5
|
|
|
|
|
|
def test_기계는_시간당_단가를_계산값으로_잠그고_산출근거를_실음(client: TestClient) -> None:
|
|
table = _get(client, "machine", q="0101-0007", size=5)
|
|
assert "hourly_total_krw" in table["locked"] and "hourly_total_krw" in table["formula"]
|
|
row = next(r for r in table["rows"] if r["@id"] == "0101-0007")
|
|
parts = row["hourly_loss_krw"] + row["hourly_fuel_krw"] + row["hourly_operator_krw"]
|
|
assert row["hourly_total_krw"] == pytest.approx(parts)
|
|
assert (
|
|
"hourly_total_krw" in row["@formula"]
|
|
and f"{int(row['hourly_total_krw']):,}" in row["@formula"]["hourly_total_krw"]
|
|
)
|
|
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:
|
|
from B09_Estimation.B09_Estimation_MachineOperating import (
|
|
hourly_cost_of,
|
|
load_operating_records,
|
|
)
|
|
|
|
codes = sorted(r.machine_code for r in load_operating_records().records)[::9]
|
|
rows = {r["@id"]: r for r in _get(client, "machine", size=500, page=1)["rows"]}
|
|
rows |= {r["@id"]: r for r in _get(client, "machine", size=500, page=2)["rows"]}
|
|
checked = 0
|
|
from B09_Estimation.B09_Estimation_MachineCost import MachineCostError
|
|
|
|
for code in codes:
|
|
try:
|
|
cost = hourly_cost_of(code)
|
|
except MachineCostError: # 손료계수 없는 기종 — B09 도 못 셈
|
|
assert rows[code]["hourly_total_krw"] is None and rows[code]["hourly_note"], code
|
|
continue
|
|
if cost.gaps:
|
|
assert rows[code]["hourly_total_krw"] is None, code # 반만 선 값은 안 냄
|
|
continue
|
|
assert rows[code]["hourly_total_krw"] == pytest.approx(float(cost.money.total)), code
|
|
checked += 1
|
|
assert checked >= 5
|
|
|
|
|
|
def test_저장은_DB_에만_쓰고_원본_파일은_안_건드림(client: TestClient, fake: FakeRepo) -> None:
|
|
before = {p: _sha(p) for p in SOURCES}
|
|
res = _save(
|
|
client,
|
|
"labor",
|
|
edits=[("labor_const/1001", "daily_wage_krw", 230000)],
|
|
base=0,
|
|
)
|
|
assert res.status_code == 200, res.text
|
|
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
|
|
|
|
|
|
def test_밑값을_고치면_기계_계산값이_따라_바뀜(client: TestClient) -> None:
|
|
base = _row(client, "machine", "0101-0007")
|
|
operator = base["operator_occupation_code"]
|
|
wage = _row(client, "labor", f"labor_const/{operator}")["daily_wage_krw"]
|
|
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)
|
|
# 노임·유가 고친 칸은 그 표 줄에 서고 기계 줄은 계산만 바뀜
|
|
assert after["changed_columns"] == ["price_thousand_krw"]
|
|
|
|
|
|
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, 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:
|
|
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, 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"]},
|
|
)
|
|
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"])
|
|
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
|
|
assert len(rate["notice"]) == 2 and len(_get(client, "material", size=1)["notice"]) == 1
|
|
|
|
|
|
def test_막는_자리(client: TestClient) -> None:
|
|
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 = {
|
|
"base_revision": 0,
|
|
"edits": [{"row_key": "labor_const/1001", "column": "daily_wage_krw", "value": bad}],
|
|
}
|
|
res = client.put(
|
|
"/api/master-data/base-prices/labor",
|
|
content=json.dumps(body).encode(),
|
|
headers={"content-type": "application/json"},
|
|
)
|
|
assert res.status_code in (400, 422), bad
|
|
fee = next(
|
|
r
|
|
for r in _get(client, "rate", size=500)["rows"]
|
|
if r["@id"].startswith("rate_performance_guarantee_fee/brackets/")
|
|
)
|
|
assert (
|
|
_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:
|
|
"""브레인 ① — 노임 두 판(건설 01-01 · 제조 07-01)·유가 두 판(전국 08-14 · 지역 09-09)이 한 표에 섞임."""
|
|
assert _row(client, "labor", "labor_const/1001")["effective_date"] == "2026-01-01"
|
|
assert _row(client, "labor", "labor_mfg/1")["effective_date"] == "2026-07-01"
|
|
assert _row(client, "oil", "national_average/oil_diesel")["effective_date"] == "2026-08-14"
|
|
assert _row(client, "oil", "regional/oil_diesel/01")["effective_date"] == "2026-09-09"
|
|
assert _row(client, "material", "10023392")["effective_date"] == "2026-08-14"
|
|
assert _row(client, "rate", "rate_sanjae")["effective_date"] == "2026-04-13"
|
|
m = _row(client, "machine", "0101-0007")
|
|
assert (m["effective_date"], m["operating_effective_date"]) == ("2026-01-01", "2026-01-01")
|
|
assert (m["fuel_price_date"], m["operator_wage_date"]) == ("2026-08-14", "2026-01-01")
|
|
no_record = next(r for r in _get(client, "machine", size=500)["rows"] if r["fuel_type"] is None)
|
|
assert no_record["operating_effective_date"] is None and no_record["hourly_total_krw"] is None
|
|
|
|
|
|
SOURCE_KEYS = {
|
|
"source_id",
|
|
"name",
|
|
"publisher",
|
|
"where",
|
|
"cycle",
|
|
"latest_published",
|
|
"our_edition",
|
|
"checked_at",
|
|
"outdated",
|
|
}
|
|
|
|
|
|
def test_표마다_출처와_뒤처짐을_서버가_판정(client: TestClient) -> None:
|
|
"""브레인 — 건설 노임이 한 판 뒤처진 것을 반 년간 아무도 몰랐음 → 표 수준에 출처 · outdated 는 서버 판정."""
|
|
labor = {s["source_id"]: s for s in _get(client, "labor", size=1)["source"]}
|
|
const = labor["labor_const"]
|
|
assert set(const) == SOURCE_KEYS
|
|
assert (const["our_edition"], const["latest_published"], const["outdated"]) == (
|
|
"2026-01-01",
|
|
"2026-09-01",
|
|
True,
|
|
)
|
|
assert "cak.or.kr" in const["where"] and const["publisher"] == "대한건설협회"
|
|
assert const["checked_at"] == "2026-09-15"
|
|
assert labor["labor_mfg"]["outdated"] is None # 최신 공표일 모름
|
|
machine = {s["source_id"]: s for s in _get(client, "machine", size=1)["source"]}
|
|
assert (
|
|
machine["machine_operating"]["our_edition"] == "2026-01-01"
|
|
) # derived_from.effective_date
|
|
assert machine["mach_base"]["outdated"] is False
|
|
assert [s["source_id"] for s in _get(client, "oil", size=1)["source"]] == [
|
|
"oil",
|
|
"oil_regional",
|
|
]
|
|
|
|
|
|
def test_우리_판은_출처표_글자가_아니라_파일에서_읽음(
|
|
client: TestClient, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
from Z01_MasterData import Z01_MasterData_BasePrices as base_prices
|
|
|
|
real = sorted((ROOT / "resources/data_master_sources").glob("sources_*.json"))[-1]
|
|
doc = json.loads(real.read_text(encoding="utf-8"))
|
|
for s in doc["sources"]:
|
|
if s["source_id"] == "rates": # 표 글자가 굳어 틀려도
|
|
s.update(our_edition="1999-01-01", latest_published="2027-01-01", status="current")
|
|
folder = tmp_path / "sources"
|
|
folder.mkdir()
|
|
(folder / "sources_2099-01-01.json").write_text(json.dumps(doc), encoding="utf-8")
|
|
monkeypatch.setattr(base_prices, "SOURCES_DIR", folder)
|
|
rate = _get(client, "rate", size=1)["source"]
|
|
assert [(s["our_edition"], s["outdated"]) for s in rate] == [("2026-04-13", True)]
|
|
monkeypatch.setattr(base_prices, "SOURCES_DIR", tmp_path / "없음")
|
|
assert _get(client, "rate", size=1)["source"] == [] # 출처표가 없으면 빈 목록(모양은 그대로)
|
|
|
|
|
|
def test_열_이름은_이름표에서만_옴(client: TestClient) -> None:
|
|
"""브레인 ② — 이름·판정은 한 곳에만. 합친 표 열 이름은 이름표 `merged_tables` 가 정본."""
|
|
labels = json.loads(
|
|
(ROOT / "resources/data_master_labels/labels_2026-01-01.json").read_text(encoding="utf-8")
|
|
)
|
|
merged = {t["key"]: t for t in labels["merged_tables"]}
|
|
for kind in ("labor", "material", "oil"):
|
|
named = {c["key"]: c for c in merged[kind]["columns"]}
|
|
for column in _get(client, kind, size=1)["columns"]:
|
|
if column["key"] in named:
|
|
assert column["label"] == named[column["key"]]["name_ko"], (kind, column["key"])
|
|
assert column["unit"] == named[column["key"]]["unit"]
|
|
assert not [c for c in _get(client, kind, size=1)["columns"] if c["label"] == c["key"]], (
|
|
kind
|
|
)
|
|
|
|
|
|
def _sorted_rows(client: TestClient, kind: str, column: str, desc: int = 0, **params) -> list:
|
|
table = _get(client, kind, sort=column, desc=desc, size=500, **params)
|
|
return [r.get(column) for r in table["rows"]]
|
|
|
|
|
|
def test_차례_세우기는_서버가_전체_줄로(client: TestClient) -> None:
|
|
"""브레인 — 자재 140 쪽이라 화면이 받은 50 줄만 세우면 1 쪽 안에서만 맞는 가짜 차례."""
|
|
table = _get(client, "material", size=1)
|
|
assert set(table["sortable"]) == {c["key"] for c in table["columns"]}
|
|
first = _get(client, "material", sort="price_krw", size=1)
|
|
last = _get(client, "material", sort="price_krw", desc=1, size=1)
|
|
assert first["total"] == last["total"] == 6999
|
|
assert first["rows"][0]["price_krw"] < last["rows"][0]["price_krw"]
|
|
page2 = _get(client, "material", sort="price_krw", size=50, page=2)["rows"]
|
|
assert first["rows"][0]["price_krw"] <= page2[0]["price_krw"] # 쪽을 넘어도 이어짐
|
|
values = _sorted_rows(client, "material", "price_krw")
|
|
assert values == sorted(values)
|
|
|
|
|
|
def test_거른_뒤에_세움_모르는_열은_거절(client: TestClient) -> None:
|
|
hits = _get(client, "material", q="육각볼트", sort="price_krw", desc=1, size=500)
|
|
values = [r["price_krw"] for r in hits["rows"]]
|
|
assert values == sorted(values, reverse=True) and hits["total"] == len(values)
|
|
assert all("육각볼트" in json.dumps(r, ensure_ascii=False) for r in hits["rows"])
|
|
bad = client.get("/api/master-data/base-prices/material", params={"sort": "없는열"})
|
|
assert bad.status_code == 400 and "없는열" in bad.text # 조용히 원본 차례로 안 돌아감
|
|
|
|
|
|
def test_빈_칸은_어느_쪽으로_세워도_끝(client: TestClient) -> None:
|
|
"""미공표 노임 14 · 손료계수 없는 기계 226 이 첫 쪽을 채우면 표를 못 씀."""
|
|
for desc in (0, 1):
|
|
wages = _sorted_rows(client, "labor", "daily_wage_krw", desc)
|
|
filled = [v for v in wages if v is not None]
|
|
assert wages[: len(filled)] == sorted(filled, reverse=bool(desc))
|
|
assert wages[len(filled) :] == [None] * (len(wages) - len(filled)) and len(filled) < len(
|
|
wages
|
|
)
|
|
hourly = _sorted_rows(client, "machine", "hourly_total_krw", 1)
|
|
assert hourly[0] is not None and hourly[-1] is None # 계산 열도 세워짐
|
|
|
|
|
|
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
|
|
|
|
|
|
def test_sort_가_없으면_원본_차례(client: TestClient) -> None:
|
|
assert _get(client, "rate", size=1)["rows"][0]["@id"] == "rate_sanjae"
|
|
assert _get(client, "machine", size=1)["rows"][0]["@id"] == "0101-0007"
|
|
|
|
|
|
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 # 계산 칸
|
|
|
|
|
|
# ── 저장소 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 = "]
|