Aislo/resources/tester/test_m01_cache.py

"""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초