Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01BW6Jdsh18WPtYUR6THZJqn
264 lines
13 KiB
Python
264 lines
13 KiB
Python
"""Z01 마스터 데이터 — 읽기 API(갈래 → 파일 → 표 · 표 줄 쪽 나누기·검색) · 2026-09-15 브레인 새 판.
|
|
|
|
약속(브레인): 파일 목록·갈래·한글 이름·단위·보임은 **이름표 파일**(`resources/data_master_labels/`)에서만 ·
|
|
표 = 화면이 표로 그릴 마디(이름표 91 이 정본) · ⭐ 트리 표 id 와 이름표 표 id 를 **양쪽으로** 세어 둘 다 0 ·
|
|
열 이름 찾는 차례 「파일id/열key」 → 「열key」 → 영문 key · 줄 한 줄 = {열key: 값} · file·table 은 트리의 id.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
from fastapi import FastAPI
|
|
from fastapi.testclient import TestClient
|
|
|
|
from Z01_MasterData import Z01_MasterData_Router as router_module
|
|
from Z01_MasterData import Z01_MasterData_Tables as tables
|
|
|
|
LABELS = json.loads(
|
|
(tables.LABEL_DIR / "labels_2026-01-01.json").read_text(encoding="utf-8")
|
|
) # 서브 이름표 판
|
|
|
|
|
|
@pytest.fixture
|
|
def client() -> TestClient:
|
|
app = FastAPI()
|
|
app.include_router(router_module.router)
|
|
return TestClient(app)
|
|
|
|
|
|
def _tree(client: TestClient) -> dict:
|
|
res = client.get("/api/master-data/tree")
|
|
assert res.status_code == 200, res.text
|
|
return res.json()
|
|
|
|
|
|
def _rows(client: TestClient, **params) -> dict:
|
|
res = client.get("/api/master-data/rows", params=params)
|
|
assert res.status_code == 200, res.text
|
|
return res.json()
|
|
|
|
|
|
def _files(client: TestClient) -> dict[str, dict]:
|
|
return {f["id"]: f for g in _tree(client)["groups"] for f in g["files"]}
|
|
|
|
|
|
def test_갈래는_이름표_차례_파일_34_강우_캐시는_없음(client: TestClient) -> None:
|
|
groups = _tree(client)["groups"]
|
|
assert [g["key"] for g in groups] == [k for k in LABELS["kinds"]]
|
|
counts = {g["key"]: len(g["files"]) for g in groups}
|
|
assert counts == {"logic": 17, "base_value": 9, "byproduct": 4, "fingerprint": 3, "seed": 1}
|
|
assert {g["key"]: g["label"] for g in groups}["base_value"] == "기초값"
|
|
by_id = {f["file_id"]: f for f in LABELS["files"]}
|
|
for f in (f for g in groups for f in g["files"]):
|
|
assert (
|
|
f["label"] == by_id[f["id"]]["name_ko"]
|
|
and f["key"] == Path(by_id[f["id"]]["path"]).name
|
|
)
|
|
assert not [
|
|
f for f in _files(client).values() if "rainfall" in f["key"] or f["key"][:3] == "002"
|
|
]
|
|
|
|
|
|
def test_트리_표_id_와_이름표_표_id_는_양쪽으로_같음(client: TestClient) -> None:
|
|
files = { # 낱값 표 `@values` 는 예약어 — 이름표 표 목록 밖(아래 시험이 따로 셈)
|
|
fid: [t for t in f["tables"] if t["id"] != tables.VALUES_TABLE]
|
|
for fid, f in _files(client).items()
|
|
}
|
|
not_labelled, label_only = [], []
|
|
for entry in LABELS["files"]:
|
|
tree_ids = {t["id"] for t in files[entry["file_id"]]}
|
|
label_ids = {t["key"] for t in entry["tables"]}
|
|
not_labelled += [f"{entry['file_id']}::{i}" for i in sorted(tree_ids - label_ids)]
|
|
label_only += [f"{entry['file_id']}::{i}" for i in sorted(label_ids - tree_ids)]
|
|
assert not_labelled == [], not_labelled
|
|
assert label_only == [], label_only
|
|
assert sum(len(t) for t in files.values()) == LABELS["counts"]["tables"] == 91
|
|
for entry in LABELS["files"]: # 이름이 실제로 붙음 · 줄 수도 이름표 셈과 같음
|
|
named = {t["key"]: t for t in entry["tables"]}
|
|
for t in files[entry["file_id"]]:
|
|
assert t["label"] == named[t["id"]]["name_ko"], (entry["file_id"], t["id"])
|
|
assert t["row_count"] == named[t["id"]]["rows"], (entry["file_id"], t["id"])
|
|
|
|
|
|
def test_자재_단가는_쪽으로_나눠_주고_검색으로_좁힘(client: TestClient) -> None:
|
|
table = "variables/mat_price/records"
|
|
first = _rows(client, file="mat_price_public", table=table, page=1, size=50)
|
|
second = _rows(client, file="mat_price_public", table=table, page=2, size=50)
|
|
assert first["total"] == 6999 and len(first["rows"]) == 50 and first["rows"] != second["rows"]
|
|
raw = json.loads(
|
|
Path(
|
|
tables.ROOT / "resources/data_cost_input_value/mat_price_public_2026-08-14.json"
|
|
).read_text(encoding="utf-8")
|
|
)["variables"]["mat_price"]["records"]
|
|
assert first["rows"][0] == raw[0] and second["rows"][0] == raw[50] # 한 줄 = {열key: 값} 그대로
|
|
found = _rows(client, file="mat_price_public", table=table, q="육각볼트", size=500)
|
|
assert 0 < found["total"] < 6999
|
|
assert all("육각볼트" in json.dumps(r, ensure_ascii=False) for r in found["rows"])
|
|
capped = _rows(client, file="mat_price_public", table=table, size=100000)
|
|
assert len(capped["rows"]) == tables.MAX_PAGE_SIZE
|
|
|
|
|
|
def test_묶음_표는_줄마다_이름_열(client: TestClient) -> None:
|
|
slope = _rows(client, file="masonry_slope", table="table")
|
|
assert [r["@key"] for r in slope["rows"]] == ["메쌓기", "찰쌓기"]
|
|
assert slope["rows"][0]["성토"] == [0.3, 0.35, 0.4, 0.45, 0.5]
|
|
cols = {c["key"]: c for c in slope["columns"]}
|
|
assert cols["성토"]["label"] == "성토부 경사" and cols["성토"]["unit"] == "1:n"
|
|
assert cols["@key"]["label"] == "쌓기 방식" # 줄 이름 열은 표마다 제 이름(이름표 표 열 목록)
|
|
fx = {c["key"]: c for c in _rows(client, file="fx", table="variables")["columns"]}
|
|
assert fx["value"]["label"] == "환율" and fx["value"]["unit"] == "원" # 「fx/value」 덮어쓰기
|
|
assert fx["@key"]["label"] == "통화"
|
|
|
|
|
|
def test_숨김은_이름표의_보임에서(client: TestClient) -> None:
|
|
cols = {c["key"]: c for c in _rows(client, file="pum_forest", table="sources")["columns"]}
|
|
assert cols["sha256"]["hidden"] is True and cols["path"]["hidden"] is False
|
|
hit = _rows(client, file="pum_forest", table="variables/pum/tables", q="F0155")
|
|
assert [r["table_id"] for r in hit["rows"]] == ["F0155"]
|
|
|
|
|
|
def test_찾는_차례와_없는_이름은_영문_key(
|
|
client: TestClient, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
(tmp_path / "labels_2026-01-01.json").write_text(
|
|
json.dumps(
|
|
{
|
|
"kinds": {"base_value": {"name_ko": "기초값"}},
|
|
"files": [
|
|
{
|
|
"file_id": "rates",
|
|
"path": "resources/data_cost_input_value/rates_2026.json",
|
|
"kind": "base_value",
|
|
"name_ko": "",
|
|
"tables": [],
|
|
},
|
|
{"file_id": "outside", "path": "config/config_db.py", "kind": "base_value"},
|
|
],
|
|
"columns": {
|
|
"rate_percent": {"name_ko": "딴 이름", "unit": "딴 단위", "visible": True},
|
|
"grade": {"name_ko": "등급", "unit": "", "visible": False},
|
|
},
|
|
"column_overrides": {"rates/rate_percent": {"name_ko": "요율", "unit": "%"}},
|
|
},
|
|
ensure_ascii=False,
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
monkeypatch.setattr(tables, "LABEL_DIR", tmp_path)
|
|
groups = _tree(client)["groups"]
|
|
assert [f["id"] for g in groups for f in g["files"]] == ["rates"] # resources 밖은 안 읽음
|
|
rates = groups[0]["files"][0]
|
|
assert rates["label"] == "rates"
|
|
goyong = next(t for t in rates["tables"] if t["id"] == "variables/rate_goyong/brackets")
|
|
assert goyong["label"] == goyong["id"]
|
|
cols = {c["key"]: c for c in _rows(client, file="rates", table=goyong["id"])["columns"]}
|
|
assert cols["rate_percent"]["label"] == "요율" and cols["rate_percent"]["unit"] == "%"
|
|
assert cols["grade"]["label"] == "등급" and cols["grade"]["hidden"] is True
|
|
assert cols["estimated_amount_bracket"] == {
|
|
"key": "estimated_amount_bracket",
|
|
"label": "estimated_amount_bracket",
|
|
"unit": "",
|
|
"hidden": False,
|
|
}
|
|
assert (
|
|
client.get("/api/master-data/rows", params={"file": "outside", "table": "x"}).status_code
|
|
== 404
|
|
)
|
|
|
|
|
|
def test_낱값은_파일마다_표_하나_설명은_뺌(client: TestClient) -> None:
|
|
"""브레인 ④ — 표 아닌 마디(값 묶음)도 트리에 `@values` 한 마디로 · 값/설명 가름은 이름표 `kind`."""
|
|
files = _files(client)
|
|
rates = files["rates"]["tables"]
|
|
assert rates[-1]["id"] == "@values" and rates[-1]["label"] == "낱값" # 이름표 synthetic_tables
|
|
values = _rows(client, file="rates", table="@values", size=500)
|
|
assert values["total"] == rates[-1]["row_count"]
|
|
by_path = {r["@path"]: r for r in values["rows"]}
|
|
sanjae = by_path["variables/rate_sanjae"]
|
|
assert sanjae["@value"] == {"rate_percent": 3.56, "base": "total_labor_cost"}
|
|
assert sanjae["@name"] == "산재보험료"
|
|
assert {c["key"]: (c["label"], c["hidden"]) for c in values["columns"]} == {
|
|
"@name": ("항목", False),
|
|
"@value": ("값", False),
|
|
"@path": ("원문 자리", True),
|
|
}
|
|
assert "@values" not in {
|
|
t["id"] for t in files["work_item_master"]["tables"]
|
|
} # 방침·집계 = 설명
|
|
assert "@values" not in {t["id"] for t in files["fx"]["tables"]} # 값 묶음 없음 — 줄이 곧 값
|
|
for entry in LABELS[
|
|
"files"
|
|
]: # 보이는 것 = 이름표가 「값」 이라 한 것 그대로(수로 박지 않음 — 이름표가 늘어도 맞음)
|
|
want = [g["key"] for g in entry["value_groups"] if g["kind"] == "value"]
|
|
got = (
|
|
_rows(client, file=entry["file_id"], table="@values", size=500)["rows"]
|
|
if "@values" in {t["id"] for t in files[entry["file_id"]]["tables"]}
|
|
else []
|
|
)
|
|
assert sorted(r["@path"] for r in got) == sorted(want), entry["file_id"]
|
|
hit = _rows(client, file="rates", table="@values", q="산재")
|
|
assert [r["@path"] for r in hit["rows"]] == ["variables/rate_sanjae"]
|
|
|
|
|
|
def test_이름표가_값_설명을_안_가른_마디는_미판정으로_보임(
|
|
client: TestClient, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
"""브레인 — 조용히 값으로 보이면 설명이 섞이고, 조용히 빼면 원가에 드는 값이 사라짐 → 값으로 보이되 드러나게."""
|
|
entry = next(f for f in LABELS["files"] if f["file_id"] == "rates")
|
|
lagging = [dict(g) for g in entry["value_groups"]]
|
|
for g in lagging:
|
|
if g["key"] in ("variables/rate_vat", "variables/rate_goyong/base"):
|
|
del g["kind"] # 이름표가 뒤처짐
|
|
elif g["key"] == "variables/rate_care":
|
|
g["kind"] = "formula" # 코드가 모르는 낱말도 미판정
|
|
(tmp_path / "labels_2026-01-01.json").write_text(
|
|
json.dumps({**LABELS, "files": [{**entry, "value_groups": lagging}]}, ensure_ascii=False),
|
|
encoding="utf-8",
|
|
)
|
|
monkeypatch.setattr(tables, "LABEL_DIR", tmp_path)
|
|
values = _rows(client, file="rates", table="@values", size=500)
|
|
marks = {r["@path"]: r.get(tables.UNDECIDED_COLUMN) for r in values["rows"]}
|
|
undecided = {"variables/rate_vat", "variables/rate_goyong/base", "variables/rate_care"}
|
|
assert {p for p, m in marks.items() if m} == undecided
|
|
assert {marks[p] for p in undecided} == {tables.UNDECIDED}
|
|
assert tables.UNDECIDED_COLUMN in {c["key"] for c in values["columns"]}
|
|
assert not next(c for c in values["columns"] if c["key"] == tables.UNDECIDED_COLUMN)["hidden"]
|
|
|
|
|
|
def test_진짜_이름표엔_미판정이_없음() -> None:
|
|
undecided = [
|
|
f"{entry['file_id']}::{r['@path']}"
|
|
for entry in LABELS["files"]
|
|
for r in tables._value_rows(entry, tables.ROOT / entry["path"], LABELS)
|
|
if r.get(tables.UNDECIDED_COLUMN)
|
|
]
|
|
assert undecided == [], undecided
|
|
|
|
|
|
def test_값_묶음_자리는_이름표와_양쪽으로_같음() -> None:
|
|
not_labelled, label_only = [], []
|
|
for entry in LABELS["files"]:
|
|
ours = set(tables.values_of(tables.ROOT / entry["path"]))
|
|
theirs = {g["key"] for g in entry["value_groups"]}
|
|
not_labelled += [f"{entry['file_id']}::{k}" for k in sorted(ours - theirs)]
|
|
label_only += [f"{entry['file_id']}::{k}" for k in sorted(theirs - ours)]
|
|
assert not_labelled == [] and label_only == [], (not_labelled, label_only)
|
|
for entry in LABELS["files"]: # 이름도 이름표가 푼 그대로(찾는 차례 「파일id::key」 → 끝 조각)
|
|
named = {g["key"]: g["name_ko"] for g in entry["value_groups"]}
|
|
rows = tables._value_rows(entry, tables.ROOT / entry["path"], LABELS)
|
|
assert [r["@name"] for r in rows] == [named[r["@path"]] for r in rows], entry["file_id"]
|
|
|
|
|
|
def test_없는_파일_표는_404(client: TestClient) -> None:
|
|
for params in (
|
|
{"file": "nope", "table": "rows"},
|
|
{"file": "rates", "table": "nope"},
|
|
{"file": "rates", "table": "variables/rate_sanjae"}, # 값 묶음 — 표 아님
|
|
{"file": "../../config/config_db", "table": "rows"},
|
|
{"file": "002yr_01hr", "table": "features"}, # 강우 IDF 캐시 — 이름표 밖
|
|
):
|
|
assert client.get("/api/master-data/rows", params=params).status_code == 404, params
|