Aislo/resources/tester/test_343_4_derived.py

"""343 덧 — 파생 저장본 도우미 · 유토 배분 저장본 · B07 읽기 안 지움 · 묶음 낡음 검사 기억."""

from __future__ import annotations

import asyncio
import json
import os
import time
from pathlib import Path
from uuid import uuid4

import pytest

from B06_Section import B06_Section_Router_HaulPlan as haul
from common_util import common_util_derived_store as store
from common_util import common_util_node_bundle as nb

# ── 도우미 ──────────────────────────────────────────────────────────────────────


def test_열쇠_같은_입력_같은_값() -> None:
    assert store.derived_key({"a": 1, "b": 2}, 3) == store.derived_key({"b": 2, "a": 1}, 3)
    assert store.derived_key({"a": 1}) != store.derived_key({"a": 2})


def test_읽기_쓰기_null_도_맞음(tmp_path: Path) -> None:
    assert store.read_derived(tmp_path, "haul", "k") is None
    store.write_derived(tmp_path, "haul", "k", None)
    assert store.read_derived(tmp_path, "haul", "k") == {"key": "k", "value": None}
    assert (tmp_path / "_derived" / "haul" / "k.json").is_file()


def test_종류마다_최근_두_벌_꺼내면_최근(tmp_path: Path) -> None:
    store.write_derived(tmp_path, "haul", "a", 1)
    time.sleep(0.02)
    store.write_derived(tmp_path, "haul", "b", 2)
    store.write_derived(tmp_path, "other", "x", 0)
    time.sleep(0.02)
    assert store.read_derived(tmp_path, "haul", "a")["value"] == 1
    time.sleep(0.02)
    store.write_derived(tmp_path, "haul", "c", 3)
    names = sorted(path.stem for path in (tmp_path / "_derived" / "haul").glob("*.json"))
    assert names == ["a", "c"]
    assert (tmp_path / "_derived" / "other" / "x.json").is_file()


def test_결과_옆_열쇠와_파일_지문(tmp_path: Path) -> None:
    path = tmp_path / "m.key.json"
    assert not store.key_matches(path, {"a": 1})
    path.write_text(json.dumps({"a": 1}), encoding="utf-8")
    assert store.key_matches(path, {"a": 1}) and not store.key_matches(path, {"a": 2})
    assert store.file_stamp(tmp_path / "없음") is None and store.file_stamp(path)


# ── 유토 배분 디스크 저장본 ─────────────────────────────────────────────────────────


def _wire(monkeypatch: pytest.MonkeyPatch, root: Path, outputs: list) -> list:
    calls: list = []

    async def _none(_project_id):  # type: ignore[no-untyped-def]
        return {}

    async def _stored(_fn, _project_id):  # type: ignore[no-untyped-def]
        return "p"

    def _run(_bundle, _script, payload):  # type: ignore[no-untyped-def]
        calls.append(payload)
        return outputs.pop(0)

    monkeypatch.setattr(haul, "haul_inputs_for", _none)
    monkeypatch.setattr(haul, "conversion_factors_for", _none)
    monkeypatch.setattr(haul, "haul_limits_for", _none)
    monkeypatch.setattr(haul, "_mass_haul_context", lambda *_: {"limits": []})
    monkeypatch.setattr(haul, "run_bundle_json", _run)
    monkeypatch.setattr(haul, "run_with_connection", _stored)
    monkeypatch.setattr(haul, "resolve_stored_project_path", lambda _p: str(root))
    haul._KEPT.clear()
    return calls


def _post(result: dict) -> dict:
    response = asyncio.run(haul.compute_haul_plan(uuid4(), 7, {"result": result}))
    return json.loads(response.body)


def test_유토_배분_재시작_뒤에도_셈_없음(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
    calls = _wire(monkeypatch, tmp_path, [{"haul_plan": {"tag": "a"}}, {"haul_plan": {"tag": "b"}}])
    a = {"points": [{"cumulative_volume_m3": 1.0}]}
    assert _post(a)["haul_plan"] == {"tag": "a"} and len(calls) == 1
    key = haul._plan_key(a, {"limits": []})  # 화면 · 서버 메모리와 같은 열쇠
    assert (tmp_path / "_derived" / "haul" / f"{key}.json").is_file()
    haul._KEPT.clear()  # 재시작
    assert _post(a)["haul_plan"] == {"tag": "a"} and len(calls) == 1
    b = {"points": [{"cumulative_volume_m3": 2.0}]}
    assert _post(b)["haul_plan"] == {"tag": "b"} and len(calls) == 2


def test_유토_배분_저장소_못_찾아도_셈(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
    calls = _wire(monkeypatch, tmp_path, [{"haul_plan": {"tag": "a"}}])

    async def _missing(_fn, _project_id):  # type: ignore[no-untyped-def]
        raise LookupError("없음")

    monkeypatch.setattr(haul, "run_with_connection", _missing)
    assert _post({"points": [{"cumulative_volume_m3": 1.0}]})["haul_plan"] == {"tag": "a"}
    assert len(calls) == 1 and not (tmp_path / "_derived").exists()


# ── B07 읽기는 측점 파일을 안 지움 ─────────────────────────────────────────────────


def test_B07_읽기는_잔재를_걸러만_읽음(tmp_path: Path) -> None:
    from B07_DesignDetail.B07_DesignDetail_Router_Support_Io import _cross_files

    cross_dir = tmp_path / "B05_Profile" / "cross_sections"
    cross_dir.mkdir(parents=True)
    for name in ("cross_00000m.json", "cross_00020m.json", "cross_00040m.json"):
        (cross_dir / name).write_text("{}", encoding="utf-8")
    longitudinal_path = tmp_path / "B05_Profile" / "longitudinal" / "longitudinal.json"
    files = _cross_files(longitudinal_path, {"stations": [{"chainage_m": 0}, {"chainage_m": 20}]})
    assert [path.name for path in files] == ["cross_00000m.json", "cross_00020m.json"]
    assert (cross_dir / "cross_00040m.json").is_file()


# ── 묶음 낡음 검사 기억 ─────────────────────────────────────────────────────────


@pytest.fixture
def bundle(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path:
    monkeypatch.setattr(nb, "ROOT", tmp_path)
    monkeypatch.setattr(nb, "_FRESH", {})
    path = tmp_path / "dist" / "bundle.mjs"
    path.parent.mkdir()
    path.write_text("", encoding="utf-8")
    os.utime(path, (1_000_000_000, 1_000_000_000))
    return path


def _counting(monkeypatch: pytest.MonkeyPatch) -> list:
    scans: list = []
    real = nb._sources_newer

    def counting(built_at: float) -> bool:
        scans.append(built_at)
        return real(built_at)

    monkeypatch.setattr(nb, "_sources_newer", counting)
    return scans


def test_낡지_않음은_잠시_기억(bundle: Path, monkeypatch: pytest.MonkeyPatch) -> None:
    scans = _counting(monkeypatch)
    assert not nb.bundle_stale(bundle) and not nb.bundle_stale(bundle)
    assert len(scans) == 1
    # 묶음을 새로 지으면(mtime 바뀜) 다시 잼
    os.utime(bundle, (1_000_000_100, 1_000_000_100))
    assert not nb.bundle_stale(bundle) and len(scans) == 2
    # 기억 시간이 지나면 다시 잼
    monkeypatch.setattr(nb, "RECHECK_S", 0.0)
    assert not nb.bundle_stale(bundle) and len(scans) == 3


def test_낡음은_기억하지_않음(bundle: Path, monkeypatch: pytest.MonkeyPatch) -> None:
    source = nb.ROOT / "B05_Profile" / "x.ts"
    source.parent.mkdir()
    source.write_text("", encoding="utf-8")
    scans = _counting(monkeypatch)
    assert nb.bundle_stale(bundle) and nb.bundle_stale(bundle)
    assert len(scans) == 2 and str(bundle) not in nb._FRESH