Merge remote-tracking branch 'origin/dev' into sub_laptop_1

This commit is contained in:
2026-09-15 19:09:39 +09:00
5 changed files with 7173 additions and 0 deletions
+29
View File
@@ -0,0 +1,29 @@
"""Z01 마스터 데이터 라우터 — 읽기 전용(고치기는 다음 차례 · 2026-09-15 브레인).
GET /api/master-data/tree 갈래 → 파일 → 표
GET /api/master-data/rows?file=&table=&page=&size=&q= 표 줄(쪽 나누기 · 검색)
⚠ 권한은 등록하는 쪽(`main.py` · 랩탑 서브)이 `dependencies=[verify_session, require_system_admin]` 로 붙임.
"""
from __future__ import annotations
from fastapi import APIRouter, HTTPException
from Z01_MasterData import Z01_MasterData_Tables as tables
router = APIRouter(prefix="/api/master-data", tags=["Z01 MasterData"])
@router.get("/tree")
def get_tree() -> dict:
return tables.tree()
@router.get("/rows")
def get_rows(
file: str, table: str, page: int = 1, size: int = tables.DEFAULT_PAGE_SIZE, q: str = ""
) -> dict:
result = tables.rows(file, table, page=page, size=size, q=q)
if result is None:
raise HTTPException(status_code=404, detail="없는 파일이나 표입니다.")
return result
+280
View File
@@ -0,0 +1,280 @@
"""Z01 마스터 데이터 — `resources/` 마스터 JSON 을 **갈래 → 파일 → 표 → 줄** 로 가름(읽기 전용).
마스터 = 로직 + 기초값(2026-09-15 사용자 방향 · bottom-up). 가르는 잣대 = 「값이 바뀌는 것 = 기초값 · 규칙·표 = 로직」(브레인).
표 가르기(브레인 승인):
· `variables` 칸은 변수 하나 = 표 하나 — 변수 안에 목록이 둘 이상이면 한 표에 잇고 `@part` 열에 목록 이름
· 나머지는 줄 모음마다 표 하나 — 딕셔너리 목록 · 같은 길이 **수** 열(글 목록은 길이가 같아도 짝짓지 않음) ·
딕셔너리의 딕셔너리(`@key` 열) · 값 딕셔너리(`@key`·`@value` 줄)
· 설명(note·source·policy …)은 표가 아님 — 줄 안의 설명 칸은 그대로 둠
한글 이름·단위는 **이름표 파일**(`resources/data_master_labels/*.json` · 데스크탑 서브 몫)에서만 — 없으면 영문 key.
"""
from __future__ import annotations
import json
import re
from functools import lru_cache
from pathlib import Path
from typing import Any
RESOURCES = Path(__file__).resolve().parent.parent / "resources"
LABEL_DIR = RESOURCES / "data_master_labels"
DEFAULT_PAGE_SIZE = 50
MAX_PAGE_SIZE = 500
# 갈래 나눔표(브레인 2026-09-15) — 차례가 화면 차례. 여기 없는 파일(강우 IDF 캐시 등)은 안 보임.
GROUPS: tuple[tuple[str, tuple[str, ...]], ...] = (
("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
# 파일 머리 — 표가 아님
_HEAD_KEYS = {
"source", "sources", "policy", "derived_from", "source_dataset_version", "source_master_file",
"dataset_version", "stats", "master", "parse_audit", "variables",
} # fmt: skip
_NOTE_KEYS = {"note", "notes", "source", "sources", "quote", "why", "policy"}
_HIDDEN_KEYS = {"sha256", "generated_at", "raw_row_index", "sort_order"}
_DATE_SUFFIX = re.compile(r"_\d{4}(?:-\d{2}-\d{2})?$")
def _is_note(key: str) -> bool:
return key in _NOTE_KEYS or key.endswith("_note")
def _scalar(v: Any) -> bool:
return not isinstance(v, (dict, list))
def _flat(d: dict, prefix: str = "") -> dict[str, Any]:
"""줄 하나 — 안쪽 딕셔너리는 `a.b` 열로 펼치고 목록은 값 그대로."""
out: dict[str, Any] = {}
for k, v in d.items():
if isinstance(v, dict) and v:
out.update(_flat(v, f"{prefix}{k}."))
else:
out[f"{prefix}{k}"] = v
return out
def _list_rows(items: list) -> list[dict[str, Any]]:
return [_flat(x) if isinstance(x, dict) else {"@value": x} for x in items]
def _numeric(v: Any) -> bool:
if isinstance(v, list):
return all(_numeric(x) and not isinstance(x, list) for x in v)
return v is None or (isinstance(v, (int, float)) and not isinstance(v, bool))
def _columnar(d: dict) -> dict[str, list] | None:
"""같은 길이 수 열(2개 이상) — 안쪽 한 겹 딕셔너리(`columns`)는 끌어올림."""
cols: dict[str, list] = {}
for k, v in d.items():
if _is_note(k):
continue
if isinstance(v, dict):
for ck, cv in v.items():
if _is_note(ck):
continue
if ck in cols or not isinstance(cv, list):
return None
cols[ck] = cv
elif isinstance(v, list) and k not in cols:
cols[k] = v
else:
return None
lengths = {len(v) for v in cols.values()}
if len(cols) < 2 or len(lengths) != 1 or 0 in lengths:
return None
if not all(_numeric(x) for v in cols.values() for x in v):
return None
return cols
def _tableish(d: dict) -> bool:
"""제 표로 떼어 낼 딕셔너리 — 같은 길이 수 열이거나 그런 것을 품음. 나머지는 한 줄로 펼침."""
return _columnar(d) is not None or any(
isinstance(v, dict) and _tableish(v) for k, v in d.items() if not _is_note(k)
)
def _dict_rows(d: dict) -> list[dict[str, Any]]:
"""목록 없는 딕셔너리 — 안쪽 딕셔너리는 키마다 한 줄(`@key` + 펼친 칸), 값은 키·값 줄."""
body = {k: v for k, v in d.items() if not _is_note(k)}
kv = [{"@key": k, "@value": v} for k, v in body.items() if not isinstance(v, dict)]
return kv + [{"@key": k, **_flat(v)} for k, v in body.items() if isinstance(v, dict)]
def _variable_rows(v: Any) -> list[dict[str, Any]]:
if isinstance(v, list):
return _list_rows(v)
lists = [(k, x) for k, x in v.items() if isinstance(x, list) and not _is_note(k)]
if not lists:
cols = _columnar(v)
return [dict(zip(cols, r)) for r in zip(*cols.values())] if cols else _dict_rows(v)
consts = _flat(
{k: x for k, x in v.items() if not _is_note(k) and (_scalar(x) or isinstance(x, dict))}
)
# `key` = 어느 열이 키인지 적은 머리 — 줄 값이 아님
consts = {k: x for k, x in consts.items() if k != "key" and not _is_note(k.rsplit(".", 1)[-1])}
return [
{**consts, **({"@part": k} if len(lists) > 1 else {}), **row}
for k, items in lists
for row in _list_rows(items)
]
def _collect(node: Any, path: str, out: list[tuple[str, list[dict[str, Any]]]]) -> None:
if isinstance(node, list):
out.append((path, _list_rows(node)))
return
cols = _columnar(node)
if cols:
out.append((path, [dict(zip(cols, r)) for r in zip(*cols.values())]))
return
body = {k: v for k, v in node.items() if not _is_note(k)}
lists = {k: v for k, v in body.items() if isinstance(v, list)}
if lists: # 목록마다 표 하나 · 곁의 값은 줄마다 붙임
consts = {k: v for k, v in body.items() if _scalar(v)}
for k, v in lists.items():
out.append((f"{path}/{k}", [{**consts, **r} for r in _list_rows(v)]))
for k, v in body.items():
if isinstance(v, dict):
_collect(v, f"{path}/{k}", out)
return
split = [k for k, v in body.items() if isinstance(v, dict) and _tableish(v)]
rest = {k: v for k, v in body.items() if k not in split}
if rest:
out.append((path, _dict_rows(rest)))
for k in split:
_collect(body[k], f"{path}/{k}", out)
def file_id(path: Path) -> str:
return _DATE_SUFFIX.sub("", path.stem)
def master_files() -> dict[str, Path]:
"""갈래표에 든 id → 파일(같은 id 가 여러 판이면 이름 차례로 끝 판)."""
wanted = {i for _, ids in GROUPS for i in ids}
found: dict[str, Path] = {}
for p in sorted(
[*RESOURCES.glob("data_*/*.json"), *RESOURCES.glob("library_structure/*.json")]
):
if file_id(p) in wanted and p.parent != LABEL_DIR:
found[file_id(p)] = p
return found
@lru_cache(maxsize=64)
def _tables_of(path: str, _mtime_ns: int) -> tuple[tuple[str, tuple[dict[str, Any], ...]], ...]:
data = json.loads(Path(path).read_text(encoding="utf-8"))
out: list[tuple[str, list[dict[str, Any]]]] = []
for name, v in (data.get("variables") or {}).items():
out.append((f"variables/{name}", _variable_rows(v)))
for k, v in data.items():
if k not in _HEAD_KEYS and not _is_note(k) and not _scalar(v):
_collect(v, k, out)
return tuple((tid, tuple(rows)) for tid, rows in out)
def tables_of(path: Path) -> dict[str, tuple[dict[str, Any], ...]]:
return dict(_tables_of(str(path), path.stat().st_mtime_ns))
def load_labels() -> dict[str, dict[str, Any]]:
"""이름표 파일 모두를 한 벌로 — 부를 때마다 읽음(작은 파일 · 고치면 바로 보임)."""
merged: dict[str, dict[str, Any]] = {"groups": {}, "files": {}, "tables": {}, "columns": {}}
for p in sorted(LABEL_DIR.glob("*.json")) if LABEL_DIR.is_dir() else []:
data = json.loads(p.read_text(encoding="utf-8"))
for section, entries in merged.items():
entries.update(data.get(section) or {})
return merged
def _text(entry: Any, key: str = "label") -> str:
if isinstance(entry, dict):
return str(entry.get(key) or "")
return str(entry or "") if key == "label" else ""
def tree() -> dict[str, Any]:
labels = load_labels()
files = master_files()
groups = []
for gkey, ids in GROUPS:
entries = []
for fid in ids:
if fid not in files:
continue
entries.append(
{
"id": fid,
"label": _text(labels["files"].get(fid)) or fid,
"key": files[fid].name,
"tables": [
{
"id": tid,
"label": _text(labels["tables"].get(f"{fid}/{tid}"))
or tid.rsplit("/", 1)[-1],
"key": tid.rsplit("/", 1)[-1],
"row_count": len(rows),
}
for tid, rows in tables_of(files[fid]).items()
],
}
)
groups.append(
{"key": gkey, "label": _text(labels["groups"].get(gkey)) or gkey, "files": entries}
)
return {"groups": groups}
def rows(
fid: str, tid: str, page: int = 1, size: int = DEFAULT_PAGE_SIZE, q: str = ""
) -> dict[str, Any] | None:
"""표 줄 한 쪽 — 없는 파일·표는 None. 검색은 줄 값 글자에 든 것(대소문자 무시)."""
path = master_files().get(fid)
table = tables_of(path).get(tid) if path else None
if table is None:
return None
columns: dict[str, None] = {}
for r in table:
columns.update(dict.fromkeys(r))
needle = q.strip().lower()
hits = (
[r for r in table if needle in json.dumps(r, ensure_ascii=False).lower()]
if needle
else list(table)
)
size = max(1, min(size, MAX_PAGE_SIZE))
start = (max(page, 1) - 1) * size
col_labels = load_labels()["columns"]
out_cols = []
for key in columns:
entry = col_labels.get(f"{fid}/{key}") or col_labels.get(key)
hidden = entry.get("hidden") if isinstance(entry, dict) and "hidden" in entry else None
out_cols.append(
{
"key": key,
"label": _text(entry) or key,
"unit": _text(entry, "unit"),
"hidden": bool(hidden)
if hidden is not None
else key.rsplit(".", 1)[-1] in _HIDDEN_KEYS,
}
)
return {"columns": out_cols, "rows": hits[start : start + size], "total": len(hits)}
File diff suppressed because it is too large Load Diff
@@ -0,0 +1,283 @@
"""마스터 이름표(`resources/data_master_labels/labels_2026-01-01.json`) 덮임 시험.
이름표는 **값을 안 담고 이름만 담는다** — 그래서 마스터가 늘거나 열이 바뀌면
이름표가 조용히 뒤처진다. 이 시험이 그 어긋남을 잡는다.
- 마스터에 있는데 이름표에 없는 표·열 → 빨강
- 이름표에 있는데 마스터에 없는 표·열(묵은 이름표) → 빨강
- 갈래 나눔은 2026-09-15 브레인 갈래표(로직 17 · 기초값 9 · 부산물 4 · 씨앗 1)와 대조
"""
from __future__ import annotations
import json
from pathlib import Path
import pytest
ROOT = Path(__file__).resolve().parents[2]
LABELS_PATH = ROOT / "resources" / "data_master_labels" / "labels_2026-01-01.json"
#: 브레인 갈래 나눔표(2026-09-15). 폴더 장부(_manifest)는 그 31 밖이라 따로 센다.
BRAIN_KIND_COUNTS = {"logic": 17, "base_value": 9, "byproduct": 4, "seed": 1}
MANIFEST_PREFIX = "manifest_"
META_KEYS = {
"schema_version",
"dataset_id",
"effective_date",
"generated_at",
"publication_date",
"survey_month",
"pum_edition",
"dataset_version",
"source_master_file",
"source_dataset_version",
}
@pytest.fixture(scope="module")
def labels() -> dict:
return json.loads(LABELS_PATH.read_text(encoding="utf-8"))
def _is_table(node) -> str | None:
"""이름표 생성기와 **같은 잣대** — 화면이 표로 그릴 마디인가."""
if isinstance(node, list) and node and all(isinstance(r, dict) for r in node):
return "list"
if isinstance(node, dict):
values = list(node.values())
dicts = [v for v in values if isinstance(v, dict)]
if (
len(values) >= 2
and len(dicts) == len(values)
and not any(
isinstance(v.get("records"), list) or isinstance(v.get("rows"), list) for v in dicts
)
):
keysets = [frozenset(v.keys()) for v in dicts]
if len(set(keysets)) <= max(2, len(keysets) // 3 + 1):
return "map"
return None
def _has_table(node, depth: int = 0) -> bool:
if _is_table(node):
return True
if depth > 5 or not isinstance(node, dict):
return False
return any(_has_table(v, depth + 1) for v in node.values())
def _collect(node, path, tables, values, depth: int = 0) -> None:
shape = _is_table(node)
if shape:
tables.append(("/".join(path), shape, node))
return
if depth <= 5 and isinstance(node, dict) and any(_has_table(v) for v in node.values()):
for key, value in node.items():
_collect(value, path + [key], tables, values, depth + 1)
return
values.append("/".join(path))
def _scan(path: Path) -> tuple[list, list]:
doc = json.loads(path.read_text(encoding="utf-8"))
tables, values = [], []
for key, value in doc.items():
if key in META_KEYS:
continue
if key == "variables" and isinstance(value, dict) and not _is_table(value):
for sub_key, sub in value.items():
_collect(sub, ["variables", sub_key], tables, values)
else:
_collect(value, [key], tables, values)
return tables, values
def _columns_of(node, shape: str) -> list[str]:
records = node if shape == "list" else list(node.values())
seen: list[str] = []
for record in records:
if not isinstance(record, dict):
continue
for key in record:
if key not in seen:
seen.append(key)
return seen
def test_이름표_파일이_읽힌다(labels):
assert labels["dataset_id"] == "data_master_labels"
assert labels["files"], "이름표에 파일이 하나도 없다"
def test_이름표가_가리키는_파일이_다_있다(labels):
missing = [f["path"] for f in labels["files"] if not (ROOT / f["path"]).is_file()]
assert not missing, f"이름표가 없는 파일을 가리킨다: {missing}"
def test_갈래_나눔이_브레인_표와_같다(labels):
counted: dict[str, int] = {}
for entry in labels["files"]:
if entry["file_id"].startswith(MANIFEST_PREFIX):
continue
counted[entry["kind"]] = counted.get(entry["kind"], 0) + 1
assert counted == BRAIN_KIND_COUNTS
assert sum(counted.values()) == 31
def test_갈래_이름이_다_풀려_있다(labels):
known = set(labels["kinds"])
used = {f["kind"] for f in labels["files"]}
assert used <= known, f"뜻을 안 적은 갈래: {sorted(used - known)}"
def test_폴더_지문은_부산물이_아니다(labels):
"""`_manifest` 는 계산이 남긴 기록이 아니라 **어느 판으로 셈했는지 못박는 표**다 —
「정본 아님」 딱지가 붙으면 안 된다(2026-09-15 브레인 ②)."""
manifests = [f for f in labels["files"] if f["file_id"].startswith(MANIFEST_PREFIX)]
assert manifests, "폴더 지문이 이름표에 하나도 없다"
for entry in manifests:
assert entry["kind"] == "fingerprint", f"{entry['file_id']} 갈래가 {entry['kind']}"
assert "정본" not in labels["kinds"]["fingerprint"]["summary"]
def test_마스터의_표가_이름표에_다_있다(labels):
missing = []
for entry in labels["files"]:
tables, _ = _scan(ROOT / entry["path"])
labelled = {t["key"] for t in entry["tables"]}
for key, _shape, _node in tables:
if key not in labelled:
missing.append(f"{entry['file_id']}::{key}")
assert not missing, f"이름표에 없는 표: {missing}"
def test_이름표의_표가_마스터에_다_있다(labels):
stale = []
for entry in labels["files"]:
tables, _ = _scan(ROOT / entry["path"])
actual = {key for key, _s, _n in tables}
for table in entry["tables"]:
if table["key"] not in actual:
stale.append(f"{entry['file_id']}::{table['key']}")
assert not stale, f"마스터에 없는 묵은 이름표: {stale}"
def test_마스터의_값_묶음이_이름표에_다_있다(labels):
missing = []
for entry in labels["files"]:
_tables, values = _scan(ROOT / entry["path"])
labelled = {v["key"] for v in entry["value_groups"]}
for key in values:
if key not in labelled:
missing.append(f"{entry['file_id']}::{key}")
assert not missing, f"이름표에 없는 값 묶음: {missing}"
def test_표_이름이_비지_않았다(labels):
blank = [
f"{entry['file_id']}::{table['key']}"
for entry in labels["files"]
for table in entry["tables"]
if not table["name_ko"].strip()
]
assert not blank, f"한글 이름이 빈 표: {blank}"
def test_값_묶음_이름이_비지_않았다(labels):
blank = [
f"{entry['file_id']}::{group['key']}"
for entry in labels["files"]
for group in entry["value_groups"]
if not group["name_ko"].strip()
]
assert not blank, f"한글 이름이 빈 값 묶음: {blank}"
def test_마스터의_열이_이름표에_다_있다(labels):
missing = []
for entry in labels["files"]:
tables, _ = _scan(ROOT / entry["path"])
by_key = {t["key"]: t for t in entry["tables"]}
for key, shape, node in tables:
table = by_key.get(key)
if table is None:
continue
labelled = {c["key"] for c in table["columns"]}
for column in _columns_of(node, shape):
if column not in labelled:
missing.append(f"{entry['file_id']}::{key}/{column}")
assert not missing, f"이름표에 없는 열: {missing}"
def test_이름표의_열이_마스터에_다_있다(labels):
stale = []
for entry in labels["files"]:
tables, _ = _scan(ROOT / entry["path"])
actual = {key: _columns_of(node, shape) for key, shape, node in tables}
for table in entry["tables"]:
known = set(actual.get(table["key"], ()))
for column in table["columns"]:
if column["key"] not in known:
stale.append(f"{entry['file_id']}::{table['key']}/{column['key']}")
assert not stale, f"마스터에 없는 묵은 열 이름표: {stale}"
def test_열_이름은_붙었거나_사유가_있다(labels):
bad = []
for entry in labels["files"]:
for table in entry["tables"]:
for column in table["columns"]:
if column["name_ko"].strip():
continue
if not column.get("unknown_reason", "").strip():
bad.append(f"{entry['file_id']}::{table['key']}/{column['key']}")
assert not bad, f"이름도 사유도 없는 열: {bad}"
def test_이름_없는_열은_unknown_에도_적혀_있다(labels):
listed = {u["where"] for u in labels["unknown"]}
for entry in labels["files"]:
for table in entry["tables"]:
for column in table["columns"]:
if column["name_ko"].strip():
continue
where = f"{entry['file_id']}::{table['key']}/{column['key']}"
assert where in listed, f"unknown 에 안 적힌 빈 열: {where}"
def test_찾는_차례가_적혀_있다(labels):
order = labels["lookup_order"]["column"]
assert order[0].startswith("column_overrides")
assert order[1].startswith("columns")
def test_덮어쓰기_이름표가_다_쓰인다(labels):
"""`column_overrides` 는 「파일id/열key」 꼴이고 그 파일에 실제로 그 열이 있어야 한다."""
by_file = {}
for entry in labels["files"]:
cols = set()
for table in entry["tables"]:
cols |= {c["key"] for c in table["columns"]}
by_file[entry["file_id"]] = cols
dangling = []
for key in labels["column_overrides"]:
file_id, _, column = key.partition("/")
if column not in by_file.get(file_id, set()):
dangling.append(key)
assert not dangling, f"쓰이지 않는 열 덮어쓰기: {dangling}"
def test_세어_둔_수가_실제와_같다(labels):
counts = labels["counts"]
assert counts["files"] == len(labels["files"])
assert counts["tables"] == sum(len(f["tables"]) for f in labels["files"])
assert counts["columns"] == sum(len(t["columns"]) for f in labels["files"] for t in f["tables"])
assert counts["value_groups"] == sum(len(f["value_groups"]) for f in labels["files"])
assert counts["unknown"] == len(labels["unknown"])
def test_강우_IDF_캐시는_안_담았다(labels):
assert not [f for f in labels["files"] if "rainfall" in f["path"]]
+173
View File
@@ -0,0 +1,173 @@
"""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