eomsangdonandClaude Opus 5
3a4823b6da
- 인력 한 줄 저장 7.0초 → 0.9초 · 자재품목 7.2초 → 2.0초 · 소요량 표 1.7초 · 자재품목 저장 뒤 목록 2.5초 → 0.45초 · 켠 뒤 첫 목록 3.4초 → 1.6초 - 로직 줄을 셀 때 `Master.get` 을 덧옷(`_Watch`)으로 감싸 그 줄이 본 자리를 모음 — 키 · 「ID:원문번호」 · 없는 키를 물은 것까지 · 재료 고르기 줄은 자재품목 묶음 통째 - 저장은 그 지도로 흔들리는 줄만 검사(고친 키 · 고치기 전후 원문번호 · 지운 줄) · 요소를 고친 저장은 연결·준용 검사를 그대로 돌림 - 저장 뒤 목록은 흔들린 자리를 알려 받아(`stale(names, marks)`) 그 줄만 다시 셈 · 다른 창이 고친 파일이면 예전처럼 통째로 다시 셈 - 자재품목을 「구분」 으로 미리 갈라 둠(`master_material.by_class`) — 고르기 후보를 3만 줄에서 훑지 않음 · 가른 차례는 파일 차례 그대로라 고르는 줄은 안 바뀜 - 전수 확인 — 요소·표 파일 90 개마다 줄을 통째로 비우고 로직 1,354 줄을 다시 검사, 달라진 줄이 모두 지도 안(지도 밖 0) - 시험 `test_m01_cache.py` 에 셋 더함(열둘) — 쓰이는 요소를 지우면 저장이 막힘 · 요소를 고쳐 저장한 뒤 막힘 표시가 줄마다 검사한 것과 같음 · 인력 줄을 비운 전수 대조 Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01JgUhN55Z3BCYhUJTjwTNfj
250 lines
10 KiB
Python
250 lines
10 KiB
Python
"""M01 목록이 마스터를 한 번만 읽는지 — `M01_MasterData_Store_Cache`.
|
|
|
|
목록(`/logics` · `/combos`)은 요청마다 마스터를 새로 읽지 않고 막힘도 미리 세어 둔 값을 씀.
|
|
여기서 보는 것 = ① 세어 둔 값이 줄마다 검사한 것과 같음 ② 파일이 바뀌면 바로 다시 읽음
|
|
③ 저장 직후 목록에 바로 비침 ④ 다른 파일의 로직이 바뀌면 그것을 부르던 줄도 다시 셈
|
|
⑤ 저장 검사를 고친 줄 둘레로 좁혀도 흠(없는 키 · 없는 요소 · 돌고 도는 참조)을 그대로 잡음.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import shutil
|
|
import time
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
from fastapi import FastAPI
|
|
from fastapi.testclient import TestClient
|
|
|
|
from M01_MasterData import M01_MasterData_Router as router_module
|
|
from M01_MasterData import M01_MasterData_Store as store
|
|
|
|
REAL = store.FOLDER
|
|
CALLER = "GC000994" # 로직_건설품셈_01장_공통.json — 아래 로직을 부름
|
|
CALLEE = "GC000276" # 로직_건설품셈_08장_건설기계.json — 다른 파일에 있음
|
|
|
|
|
|
@pytest.fixture
|
|
def client(tmp_path: Path, monkeypatch) -> TestClient:
|
|
for path in REAL.glob("*.json"):
|
|
if not path.name.startswith("_") or path.name == store.mk.BOOK.name:
|
|
shutil.copy(path, tmp_path / path.name)
|
|
monkeypatch.setattr(store, "FOLDER", tmp_path)
|
|
app = FastAPI()
|
|
app.include_router(router_module.router)
|
|
return TestClient(app)
|
|
|
|
|
|
def _logics(client: TestClient, **params) -> list[dict]:
|
|
res = client.get("/api/m01/logics", params=params)
|
|
assert res.status_code == 200, res.text
|
|
return res.json()["logics"]
|
|
|
|
|
|
def _write(file: str, change) -> None:
|
|
"""정본 파일을 손으로 고침(저장 길을 거치지 않음) — 바뀐 것을 알아채는지 보려고."""
|
|
path = store.FOLDER / file
|
|
data = json.loads(path.read_text(encoding="utf-8"))
|
|
change(data)
|
|
path.write_text(json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8", newline="\n")
|
|
|
|
|
|
def test_세어_둔_막힘이_줄마다_검사한_것과_같다(client: TestClient) -> None:
|
|
listed = _logics(client)
|
|
files, whole = store.loaded()
|
|
rows = {
|
|
str(r.get("키")): r for _, d in files.items() if d.get("그룹") == "로직" for r in d["줄"]
|
|
}
|
|
assert len(listed) == len(rows)
|
|
for one in listed:
|
|
reasons = store.cm.mf.check_logic(whole, rows[one["키"]])[0]
|
|
assert one["reasons"] == reasons and one["blocked"] is bool(reasons)
|
|
|
|
|
|
def test_거름이_그대로다(client: TestClient) -> None:
|
|
whole = _logics(client)
|
|
sub, detail = whole[0]["구분"], whole[0]["상세구분"]
|
|
by_sub = _logics(client, sub=sub, detail=detail)
|
|
assert by_sub == [x for x in whole if (x["구분"], x["상세구분"]) == (sub, detail)]
|
|
assert _logics(client, blocked=1) == [x for x in whole if x["blocked"]]
|
|
assert _logics(client, blocked=0) == [x for x in whole if not x["blocked"]]
|
|
assert _logics(client, q=CALLER) == [x for x in whole if x["키"] == CALLER]
|
|
|
|
|
|
def test_안_바뀌면_다시_읽지_않고_바뀌면_바로_읽는다(client: TestClient) -> None:
|
|
files, whole = store.loaded()
|
|
assert store.loaded() == (files, whole) # 같은 것을 그대로 씀
|
|
_write("로직_건설품셈_01장_공통.json", lambda d: d["줄"][0].update(비고="자국 바뀜"))
|
|
fresh, _ = store.loaded()
|
|
assert fresh is not files
|
|
assert fresh["로직_건설품셈_01장_공통.json"]["줄"][0]["비고"] == "자국 바뀜"
|
|
|
|
|
|
def test_로직을_저장하면_목록에_바로_비친다(client: TestClient) -> None:
|
|
one = client.get("/api/m01/logic", params={"key": CALLER}).json()
|
|
assert [x for x in _logics(client, q=CALLER)][0]["이름"] == one["logic"]["이름"]
|
|
row = {**one["logic"], "이름": "저장 뒤 이름"}
|
|
res = client.post(
|
|
"/api/m01/save",
|
|
json={
|
|
"files": [
|
|
{
|
|
"file": one["file"],
|
|
"version": one["version"],
|
|
"changes": [{"op": "edit", "key": CALLER, "row": row}],
|
|
}
|
|
]
|
|
},
|
|
)
|
|
assert res.status_code == 200, res.text
|
|
listed = [x for x in _logics(client) if x["키"] == CALLER]
|
|
assert listed and listed[0]["이름"] == "저장 뒤 이름"
|
|
|
|
|
|
def test_부르는_로직이_바뀌면_다른_파일의_줄도_다시_센다(client: TestClient) -> None:
|
|
assert [x for x in _logics(client) if x["키"] == CALLER][0]["blocked"] is False
|
|
|
|
def add_input(data: dict) -> None:
|
|
row = next(r for r in data["줄"] if r["키"] == CALLEE)
|
|
row["입력"] = [*row.get("입력", []), {"이름": "새입력", "종류": "수"}]
|
|
|
|
_write("로직_건설품셈_08장_건설기계.json", add_input) # 부르는 쪽 파일은 그대로
|
|
caller = [x for x in _logics(client) if x["키"] == CALLER][0]
|
|
assert caller["blocked"] is True and any("새입력" in x for x in caller["reasons"])
|
|
|
|
|
|
def _save(client: TestClient, key: str, spoil) -> object:
|
|
"""그 로직 줄을 고쳐 저장 길로 보냄 — 저장 검사가 흠을 잡는지 보려고."""
|
|
one = client.get("/api/m01/logic", params={"key": key}).json()
|
|
row = json.loads(json.dumps(one["logic"], ensure_ascii=False))
|
|
spoil(row)
|
|
return client.post(
|
|
"/api/m01/save",
|
|
json={
|
|
"files": [
|
|
{
|
|
"file": one["file"],
|
|
"version": one["version"],
|
|
"changes": [{"op": "edit", "key": key, "row": row}],
|
|
}
|
|
]
|
|
},
|
|
)
|
|
|
|
|
|
def _errors(res) -> list[str]:
|
|
assert res.status_code == 422, res.text
|
|
return res.json()["detail"]["errors"]
|
|
|
|
|
|
def test_저장_검사를_좁혀도_흠을_잡는다(client: TestClient) -> None:
|
|
"""고친 줄과 그 줄을 부르던 줄만 검사하지만 없는 키 · 없는 요소는 그대로 걸림."""
|
|
_logics(client) # 미리 세어 둔 값이 선 평소 모양
|
|
|
|
def no_logic(row: dict) -> None:
|
|
at = next(i for i, h in enumerate(row["호표"]) if h["종류"] == "로직")
|
|
row["호표"][at]["요소"] = "로직(GC999999, 기계='5401-0017', 지역=지역)"
|
|
|
|
def no_element(row: dict) -> None:
|
|
at = next(i for i, h in enumerate(row["호표"]) if h["종류"] == "인력")
|
|
row["호표"][at]["요소"] = "LB999999"
|
|
|
|
for spoil, word in ((no_logic, "GC999999"), (no_element, "LB999999")):
|
|
assert any(word in x for x in _errors(_save(client, CALLER, spoil)))
|
|
|
|
|
|
def test_돌고_도는_참조도_저장이_막는다(client: TestClient) -> None:
|
|
_logics(client)
|
|
|
|
def call_back(row: dict) -> None:
|
|
row.setdefault("호표", []).append(
|
|
{
|
|
"종류": "로직",
|
|
"요소": f"로직({CALLER}, 지반구분='토사', 크레인='2104-0010', 지역=지역)",
|
|
"이름": "되부르기",
|
|
"단위": "시간",
|
|
"수량": "1",
|
|
}
|
|
)
|
|
|
|
assert any("돌고 도는" in x for x in _errors(_save(client, CALLEE, call_back)))
|
|
|
|
|
|
def test_고친_줄을_부르던_줄이_깨지면_저장이_막힌다(client: TestClient) -> None:
|
|
"""좁힌 검사가 「그 줄을 부르는 줄」 을 빠뜨리지 않는지 — 다른 파일에 있는 줄."""
|
|
_logics(client)
|
|
|
|
def add_input(row: dict) -> None:
|
|
row["입력"] = [*row.get("입력", []), {"이름": "새입력", "종류": "수"}]
|
|
|
|
assert any("새입력" in x for x in _errors(_save(client, CALLEE, add_input)))
|
|
|
|
|
|
def test_미리_읽어_두면_첫_목록이_기다리지_않는다(client: TestClient) -> None:
|
|
store.cache.forget()
|
|
thread = store.cache.warm(store.FOLDER)
|
|
thread.join(120)
|
|
assert not thread.is_alive()
|
|
start = time.perf_counter()
|
|
assert _logics(client)
|
|
assert time.perf_counter() - start < 1.0 # 미리 안 읽으면 2~3초
|
|
|
|
|
|
def _plain(file: str) -> dict:
|
|
"""정본 파일 그대로(수는 글이 아닌 그대로) — 저장에 실어 보낼 줄을 꺼내려고."""
|
|
return json.loads((store.FOLDER / file).read_text(encoding="utf-8"))
|
|
|
|
|
|
def _save_file(client: TestClient, file: str, changes: list[dict]):
|
|
"""정본 파일을 저장 길로 고침 — 요소·표 줄도 같은 길."""
|
|
return client.post(
|
|
"/api/m01/save",
|
|
json={"files": [{"file": file, "version": store.read(file)[1], "changes": changes}]},
|
|
)
|
|
|
|
|
|
def test_쓰이는_요소를_지우면_저장이_막힌다(client: TestClient) -> None:
|
|
"""요소 저장도 좁혀 검사하지만 그 요소를 쓰던 로직은 그대로 걸림."""
|
|
_logics(client) # 미리 세어 둔 값이 선 평소 모양
|
|
res = _save_file(client, "인력.json", [{"op": "delete", "key": "LB000014"}])
|
|
assert any("LB000014" in x for x in _errors(res))
|
|
|
|
|
|
def test_요소를_고쳐_저장해도_막힘_표시가_전수와_같다(client: TestClient) -> None:
|
|
"""요소를 고친 뒤 다시 센 막힘이 줄마다 검사한 것과 같은지 — 좁혀 세는 길."""
|
|
was = _logics(client)
|
|
raw = _plain("인력.json")
|
|
row = {**next(r for r in raw["줄"] if str(r.get("키")) == "LB000014"), "비고": "저장 검사 시험"}
|
|
res = _save_file(client, "인력.json", [{"op": "edit", "key": "LB000014", "row": row}])
|
|
assert res.status_code == 200, res.text
|
|
listed = _logics(client)
|
|
files, whole = store.loaded()
|
|
rows = {
|
|
str(r.get("키")): r for _, d in files.items() if d.get("그룹") == "로직" for r in d["줄"]
|
|
}
|
|
assert [x["키"] for x in listed] == [x["키"] for x in was]
|
|
for one in listed:
|
|
assert one["reasons"] == store.cm.mf.check_logic(whole, rows[one["키"]])[0]
|
|
|
|
|
|
def test_요소_지도가_전수_검사와_같다(client: TestClient) -> None:
|
|
"""「어느 로직이 어느 요소를 보는지」 지도가 놓치는 자리가 없는지 —
|
|
인력 줄을 통째로 비우고 1,354 줄을 다 검사해, 달라진 줄이 지도 안에 드는지 봄."""
|
|
counts = store.cache.counted(store.FOLDER)
|
|
files, _ = store.loaded()
|
|
marks = set()
|
|
for row in files["인력.json"]["줄"]:
|
|
key = str(row.get("키"))
|
|
marks |= {key, "#" + key[:2]}
|
|
if row.get("원문번호"):
|
|
marks.add(f"{key[:2]}:{row['원문번호']}")
|
|
marked = {key for key, got in counts.items() if got[2] & marks}
|
|
empty = {**files, "인력.json": {**files["인력.json"], "줄": []}}
|
|
other = store.cm.mf.Master(empty)
|
|
rows = {
|
|
str(r.get("키")): r for _, d in files.items() if d.get("그룹") == "로직" for r in d["줄"]
|
|
}
|
|
changed = {k for k, r in rows.items() if store.cm.mf.check_logic(other, r)[0] != counts[k][0]}
|
|
assert changed and changed <= marked
|