Aislo/resources/tester/test_343_3_browser_cache.py

"""343-3 B04 다시 열기 받는 양 — 보관함 저장 실패 고침 · 점군 고정 표본 이진 · 켤 때 받기 · ETag.

`helper_343_3_asset_cache.cjs` = 진짜 TS(b_asset_cache · decodePointPreview) + 가짜 IndexedDB.
"""

from __future__ import annotations

import json
import subprocess
from pathlib import Path

import numpy as np
import pytest

import B04_PreProcess.B04_PreProcess_Engine_PointPreview as preview

ROOT = Path(__file__).resolve().parents[2]
HERE = Path(__file__).resolve().parent


def _npz(path: Path, count: int, rgb: bool = True) -> tuple[np.ndarray, np.ndarray]:
    rng = np.random.default_rng(3)
    xyz = np.column_stack(
        (
            rng.uniform(200000, 200100, count),
            rng.uniform(500000, 500100, count),
            rng.uniform(90, 120, count),
        )
    ).astype(np.float32)
    colors = rng.integers(0, 65535, (count, 3)).astype(np.uint16)
    bounds = np.array([[200000, 200100], [500000, 500100], [90, 120]], dtype=np.float64)
    arrays = {"xyz": xyz, "bounds": bounds}
    if rgb:
        arrays["rgb"] = colors
    np.savez_compressed(path, **arrays)
    return xyz, colors


def _decode(path: Path) -> tuple[dict, np.ndarray, np.ndarray | None]:
    raw = path.read_bytes()
    length = int.from_bytes(raw[:4], "little")
    head = json.loads(raw[4 : 4 + length])
    offset = -(-(4 + length) // 4) * 4
    count = head["sampled_count"]
    xyz = np.frombuffer(raw, "<f4", count * 3, offset).reshape(-1, 3)
    rgb = None
    if head["has_rgb"]:
        rgb = np.frombuffer(raw, "<u2", count * 3, offset + count * 12).reshape(-1, 3)
    return head, xyz, rgb


def test_고정_표본은_JSON_길과_같은_점(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
    monkeypatch.setattr(preview, "POINT_CLOUD_SAMPLE_LIMIT", 50)
    source = tmp_path / "structured.npz"
    xyz, colors = _npz(source, 200)
    head, sample, rgb = _decode(preview.ensure_point_preview(source, None))
    # 옛 JSON 길과 같은 규칙(같은 씨앗 · 같은 float32)
    indexes = np.random.default_rng(20260710).choice(200, 50, replace=False)
    assert head["point_count"] == 200 and head["sampled_count"] == 50
    assert np.array_equal(sample, xyz[indexes]) and np.array_equal(rgb, colors[indexes])
    assert head["bounds"]["x_min"] == 200000 and head["bounds"]["z_max"] == 120


def test_원천이_그대로면_다시_안_만듦(tmp_path: Path) -> None:
    source = tmp_path / "ground_points_csf.npz"
    _npz(source, 30, rgb=False)
    target = preview.ensure_point_preview(source, "csf")
    assert target.name == "preview_points_csf.bin"
    stamp = target.stat().st_mtime_ns
    assert preview.ensure_point_preview(source, "csf").stat().st_mtime_ns == stamp
    _npz(source, 40, rgb=False)
    head, _, rgb = _decode(preview.ensure_point_preview(source, "csf"))
    assert head["sampled_count"] == 40 and rgb is None


@pytest.fixture(scope="module")
def out(tmp_path_factory: pytest.TempPathFactory) -> dict:
    folder = tmp_path_factory.mktemp("p343")
    source = folder / "structured.npz"
    xyz, colors = _npz(source, 25)
    target = preview.ensure_point_preview(source, None)
    proc = subprocess.run(
        ["node", str(HERE / "helper_343_3_asset_cache.cjs"), str(target)],
        capture_output=True,
        text=True,
        encoding="utf-8",
        timeout=60,
    )
    assert proc.returncode == 0, proc.stderr
    data = json.loads(proc.stdout)
    data["expected"] = {
        "first": xyz[0].tolist(),
        "last": xyz[-1].tolist(),
        "rgb": colors[0].tolist(),
    }
    return data


def test_보관함_저장_뒤_두번째는_보관본(out: dict) -> None:
    store = out["store"]
    assert store["stored"] and store["first"] == store["second"] == "body:http://x/a"
    assert store["warnings"] == []


def test_용량_초과면_한_줄_경고_다음엔_다시_받음(out: dict) -> None:
    quota = out["quota"]
    assert quota["stored"] == 0 and quota["refetched"] == 1
    assert len(quota["warnings"]) == 1  # 같은 까닭은 한 번만
    assert "저장" in quota["warnings"][0] and "QuotaExceededError" in quota["warnings"][0]


def test_열기_실패를_끝까지_기억하지_않음(out: dict) -> None:
    retry = out["retry"]
    assert retry["missed"] is None and retry["hit"] == "v" and retry["opens"] == 2
    assert any("열기" in line for line in retry["warnings"])


def test_다른_탭이_막아도_멈추지_않음(out: dict) -> None:
    blocked = out["blocked"]
    assert blocked["body"] == "body:http://x/a"
    assert any("막음" in line for line in blocked["warnings"])


def test_지우기는_색인_열쇠만(out: dict) -> None:
    purge = out["purge"]
    assert purge["keys"] == ["P|http://x/keep"] and purge["valueReads"] == 0


def test_등고선_주소_한_벌(out: dict) -> None:
    assert out["contour"] == [
        "http://x/api/projects/P/surface/models/7/contour?interval=1&smooth=false&recalculate=false",
        "http://x/api/projects/P/surface/models/7/contour?interval=2&smooth=true&recalculate=true",
    ]
    source = (ROOT / "B04_PreProcess/B04_PreProcess_UI_TerrainViewer_Contours.ts").read_text(
        encoding="utf-8"
    )
    assert "contourAssetUrl(" in source and "/contour?interval" not in source


def test_TS_가_파이썬_이진을_같은_값으로_풂(out: dict) -> None:
    decoded, expected = out["preview"], out["expected"]
    assert decoded["point_count"] == decoded["sampled_count"] == 25
    assert decoded["first"] == expected["first"] and decoded["last"] == expected["last"]
    assert decoded["rgbFirst"] == expected["rgb"]
    assert decoded["bounds"]["y_max"] == 500100


def test_꺼진_층은_켤_때_받음() -> None:
    viewer = (ROOT / "B04_PreProcess/B04_PreProcess_UI_MapViewer.ts").read_text(encoding="utf-8")
    assert "GIS_LAYERS.filter((layer) => activeGisLayers.has(layer))" in viewer
    assert "activeGisLayers.forEach((layer) => void loadLayerOnDemand(layer))" in viewer
    assert len(viewer.splitlines()) <= 700


def test_남은_층은_파일_그대로_ETag() -> None:
    source = (ROOT / "B04_PreProcess/B04_PreProcess_Router_GIS.py").read_text(encoding="utf-8")
    start = source.index("async def get_project_geojson")
    body = source[start : source.index("@tiles_router", start)]
    assert "json.loads(filepath" not in body and "json.loads(simplified" not in body
    assert body.rstrip().count("cached_file_response(") >= 2


def test_화면은_이진_길로_점군을_받음() -> None:
    api = (ROOT / "B04_PreProcess/B04_PreProcess_Api_Fetch.ts").read_text(encoding="utf-8")
    assert "/surface/point-cloud/binary" in api and "fetchCachedBytes(" in api


def test_전처리가_미리보기_이진을_한_번_만듦(
    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
    import laspy

    import B04_PreProcess.B04_PreProcess_Engine as engine
    import B04_PreProcess.B04_PreProcess_Engine_SheetSurface as sheet

    rng = np.random.default_rng(1)
    header = laspy.LasHeader(point_format=3, version="1.2")
    header.offsets = [200000, 500000, 0]
    header.scales = [0.01, 0.01, 0.01]
    las_data = laspy.LasData(header)
    las_data.x = rng.uniform(200000, 200060, 3000)
    las_data.y = rng.uniform(500000, 500060, 3000)
    las_data.z = rng.uniform(100, 101, 3000)
    las_data.classification = np.full(3000, 2, np.uint8)
    las_path = tmp_path / "B03_FileInput" / "input" / "t.las"
    las_path.parent.mkdir(parents=True)
    las_data.write(las_path)
    monkeypatch.setattr(engine, "download_geodata", lambda *a, **k: None)
    monkeypatch.setattr(sheet, "build_sheet_surface_from_route", lambda *a, **k: [])
    result = engine.run_surface_analysis(tmp_path, las_path, source_filters=None, methods=["dtm"])
    processed = tmp_path / "B04_PreProcess" / "processed"
    made = {path.name: path.stat().st_mtime_ns for path in processed.glob("preview_points_*.bin")}
    assert "preview_points_raw.bin" in made
    assert {f"preview_points_{key}.bin" for key in result["source_filters"]} <= set(made)
    engine.run_surface_analysis(tmp_path, las_path, source_filters=None, methods=["dtm"])
    again = {path.name: path.stat().st_mtime_ns for path in processed.glob("preview_points_*.bin")}
    assert again == made