Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01BW6Jdsh18WPtYUR6THZJqn
174 lines
8.4 KiB
Python
174 lines
8.4 KiB
Python
"""Z01 마스터 데이터 — 읽기 API(갈래 → 파일 → 표 · 표 줄 쪽 나누기·검색) · 2026-09-15 브레인 새 판.
|
|
|
|
약속(브레인 승인): 갈래 넷 로직 17 · 기초값 9 · 부산물 4 · 씨앗 1 = 31 파일 · 강우 IDF 캐시는 뺌 ·
|
|
variables 칸은 변수 하나 = 표 하나 · 나머지는 줄 모음(목록·같은 길이 수 열)마다 표 하나 · 설명(note·source·policy)은 표 아님 ·
|
|
한글 이름은 이름표 파일(`resources/data_master_labels/`)에서만 — 찾는 차례 「파일id/열key」 → 「열key」 → 영문 key ·
|
|
unit 도 이름표에서만(자료에서 지어내지 않음) · 숨김 = 내부 id · sha · 생성시각.
|
|
"""
|
|
|
|
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
|
|
|
|
GROUPS = {
|
|
"logic": {
|
|
"pum_forest", "pum_const", "work_item_master", "coef", "material_surcharge", "formwork_reuse",
|
|
"rebar_complexity", "timber_structure_class", "masonry_class", "masonry_slope", "masonry_back_length",
|
|
"stone_kind", "revetment_sabang", "structure_unit_observed", "work_item_mapping", "aliases",
|
|
"resource_catalog_ext",
|
|
},
|
|
"base": {
|
|
"labor_const", "labor_mfg", "mach_base", "machine_operating", "mat_price_public", "oil", "oil_regional",
|
|
"fx", "rates",
|
|
},
|
|
"byproduct": {"basis_missing", "form_undetermined", "resource_axis", "unmatched"},
|
|
"seed": {"masonry_wet"},
|
|
} # fmt: skip
|
|
|
|
|
|
@pytest.fixture
|
|
def client(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> TestClient:
|
|
monkeypatch.setattr(tables, "LABEL_DIR", tmp_path / "labels") # 이름표 없음 — 영문 key 그대로
|
|
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 test_갈래_넷에_파일_31_강우_캐시는_없음(client: TestClient) -> None:
|
|
groups = _tree(client)["groups"]
|
|
assert [g["key"] for g in groups] == ["logic", "base", "byproduct", "seed"]
|
|
for g in groups:
|
|
assert {f["id"] for f in g["files"]} == GROUPS[g["key"]], g["key"]
|
|
assert g["label"] == g["key"] # 이름표가 없으면 영문 key
|
|
for f in g["files"]:
|
|
assert f["label"] == f["id"] and f["key"].endswith(".json") and f["tables"], f["id"]
|
|
ids = [f["id"] for g in groups for f in g["files"]]
|
|
assert len(ids) == 31 and not any("idf" in i or "rainfall" in i for i in ids)
|
|
|
|
|
|
def test_variables_는_변수_하나가_표_하나(client: TestClient) -> None:
|
|
files = {f["id"]: f for g in _tree(client)["groups"] for f in g["files"]}
|
|
rates = [t for t in files["rates"]["tables"] if t["id"].startswith("variables/")]
|
|
assert len(rates) == 19, [t["id"] for t in rates]
|
|
vat = next(t for t in rates if t["id"] == "variables/rate_vat")
|
|
assert (
|
|
vat["key"] == "rate_vat" and vat["label"] == "rate_vat" and vat["row_count"] == 2
|
|
) # rate_percent · base
|
|
assert len([t for t in files["coef"]["tables"] if t["id"].startswith("variables/")]) == 5
|
|
assert "processing_rules" in {t["id"] for t in files["rates"]["tables"]} # 변수 밖 규칙도 표
|
|
# 한 변수에 목록이 둘이면 한 표에 잇고 어느 목록인지 열로 가름
|
|
fuel = _rows(client, file="mach_base", table="variables/mach_fuel_rate", size=500)
|
|
assert fuel["total"] == 214 + 21
|
|
assert {r["@part"] for r in fuel["rows"]} >= {"parsed_records"}
|
|
|
|
|
|
def test_자재_단가는_쪽으로_나눠_주고_검색으로_좁힘(client: TestClient) -> None:
|
|
first = _rows(client, file="mat_price_public", table="variables/mat_price", page=1, size=50)
|
|
second = _rows(client, file="mat_price_public", table="variables/mat_price", page=2, size=50)
|
|
assert first["total"] == 6999 and len(first["rows"]) == 50 and first["rows"] != second["rows"]
|
|
assert {c["key"] for c in first["columns"]} >= {"item_code", "price_krw", "specification"}
|
|
assert "key" not in {
|
|
c["key"] for c in first["columns"]
|
|
} # 「어느 열이 키인지」 머리는 줄 값이 아님
|
|
found = _rows(
|
|
client, file="mat_price_public", table="variables/mat_price", 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="variables/mat_price", size=100000)
|
|
assert len(capped["rows"]) == tables.MAX_PAGE_SIZE
|
|
|
|
|
|
def test_같은_길이_수_열은_줄로_세움(client: TestClient) -> None:
|
|
slope = _rows(client, file="masonry_slope", table="table/메쌓기")
|
|
assert slope["total"] == 5 and slope["rows"][0] == {"성토": 0.3, "절토": 0.25}
|
|
wedge = _rows(client, file="masonry_wet", table="tables/고임돌표")
|
|
assert wedge["total"] == 7 and wedge["rows"][0]["keys"] == 25
|
|
# 글 목록 둘은 길이가 같아도 짝짓지 않음(자재 이름 ↔ 뒤진 자리)
|
|
files = {f["id"]: f for g in _tree(client)["groups"] for f in g["files"]}
|
|
ids = {t["id"] for t in files["material_surcharge"]["tables"]}
|
|
assert {"not_found/materials", "not_found/checked"} <= ids
|
|
|
|
|
|
def test_원문_표는_번호로_찾음(client: TestClient) -> None:
|
|
hit = _rows(client, file="pum_forest", table="variables/pum", q="F0155")
|
|
assert [r["table_id"] for r in hit["rows"]] == ["F0155"]
|
|
|
|
|
|
def test_내부_id_sha_생성시각은_숨김(client: TestClient) -> None:
|
|
axis = _rows(client, file="resource_axis", table="rows", size=1)
|
|
hidden = {c["key"]: c["hidden"] for c in axis["columns"]}
|
|
assert hidden["raw_row_index"] is True and hidden["resource_name"] is False
|
|
master = _rows(client, file="work_item_master", table="work_items", size=1)
|
|
assert {c["key"]: c["hidden"] for c in master["columns"]}["sort_order"] is True
|
|
assert all(c["unit"] == "" and c["label"] == c["key"] for c in axis["columns"]) # 이름표 없음
|
|
|
|
|
|
def test_이름표_파일이_있으면_label_unit_을_붙임(client: TestClient, tmp_path: Path) -> None:
|
|
labels = tmp_path / "labels"
|
|
labels.mkdir()
|
|
(labels / "labels.json").write_text(
|
|
json.dumps(
|
|
{
|
|
"groups": {"base": "기초값"},
|
|
"files": {"rates": "요율"},
|
|
"tables": {"rates/variables/rate_vat": "부가가치세"},
|
|
"columns": {
|
|
"base": {"label": "밑수"},
|
|
"rates/rate_percent": {"label": "요율", "unit": "%"},
|
|
"rate_percent": {"label": "딴 이름", "unit": "딴 단위"},
|
|
},
|
|
},
|
|
ensure_ascii=False,
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
base = next(g for g in _tree(client)["groups"] if g["key"] == "base")
|
|
rates = next(f for f in base["files"] if f["id"] == "rates")
|
|
assert base["label"] == "기초값" and rates["label"] == "요율"
|
|
assert (
|
|
next(t for t in rates["tables"] if t["id"] == "variables/rate_vat")["label"] == "부가가치세"
|
|
)
|
|
cols = {c["key"]: c for c in _rows(client, file="rates", table="variables/rate_vat")["columns"]}
|
|
assert cols["@key"]["label"] == "@key" # 없는 것은 영문 key
|
|
vat = _rows(client, file="rates", table="variables/rate_sanjae")
|
|
assert {r["@key"] for r in vat["rows"]} == {"rate_percent", "base"}
|
|
goyong = {
|
|
c["key"]: c for c in _rows(client, file="rates", table="variables/rate_goyong")["columns"]
|
|
}
|
|
assert (
|
|
goyong["rate_percent"]["label"] == "요율" and goyong["rate_percent"]["unit"] == "%"
|
|
) # 파일 것이 이김
|
|
assert goyong["base"]["label"] == "밑수" and goyong["base"]["unit"] == ""
|
|
|
|
|
|
def test_없는_파일_표는_404_경로_넘기도_404(client: TestClient) -> None:
|
|
for params in (
|
|
{"file": "nope", "table": "rows"},
|
|
{"file": "rates", "table": "nope"},
|
|
{"file": "../../config/config_db", "table": "rows"},
|
|
{"file": "002yr_01hr", "table": "features"}, # 강우 IDF 캐시 — 갈래표 밖
|
|
{"file": "_manifest", "table": "files"},
|
|
):
|
|
assert client.get("/api/master-data/rows", params=params).status_code == 404, params
|