tests/test_tile_cache.py

"""The optional R2 LiDAR tile cache (A3) + its L1/L2/origin fall-through in _stream_download.

All fakes — no boto3, no network (the autouse guard would fail a real socket). The point is
that the cache is a transparent no-op when R2 env is unset (byte-identical to before) and a
correct L2 layer when it's set.
"""

from __future__ import annotations

from lawn_estimator.sources import lidar
from lawn_estimator.sources.tile_cache import TileCache

_R2_ENV = ["R2_BUCKET", "R2_ENDPOINT", "R2_ACCOUNT_ID", "R2_ACCESS_KEY_ID", "R2_SECRET_ACCESS_KEY"]


def _clear_r2(monkeypatch):
    for k in _R2_ENV:
        monkeypatch.delenv(k, raising=False)


# --- env gating ------------------------------------------------------------

def test_cache_disabled_without_env(monkeypatch):
    _clear_r2(monkeypatch)
    assert TileCache().enabled is False


def test_cache_enabled_and_derives_endpoint_from_account(monkeypatch):
    _clear_r2(monkeypatch)
    monkeypatch.setenv("R2_BUCKET", "tiles")
    monkeypatch.setenv("R2_ACCESS_KEY_ID", "key")
    monkeypatch.setenv("R2_SECRET_ACCESS_KEY", "secret")
    monkeypatch.setenv("R2_ACCOUNT_ID", "acct123")
    tc = TileCache()
    assert tc.enabled is True
    assert tc._endpoint == "https://acct123.r2.cloudflarestorage.com"
    assert tc._obj_key("douglas/x.las") == "lidar/douglas/x.las"   # default prefix


def test_disabled_cache_methods_are_noops_and_never_touch_boto3(monkeypatch, tmp_path):
    # A disabled cache must not import boto3 or write anything — fetch is a miss, store a no-op.
    _clear_r2(monkeypatch)
    tc = TileCache()
    dest = tmp_path / "x.las"
    assert tc.fetch("douglas/x.las", dest) is False
    assert not dest.exists()
    tc.store("douglas/x.las", tmp_path)   # no source needed; returns without error


# --- L1 / L2 / origin fall-through in _stream_download ----------------------

class _FakeResp:
    def __init__(self, data: bytes):
        self._data = data
        self.headers = {"content-length": str(len(data))}

    def __enter__(self):
        return self

    def __exit__(self, *_):
        return False

    def raise_for_status(self):
        pass

    def iter_content(self, chunk_size=1):
        for i in range(0, len(self._data), chunk_size):
            yield self._data[i : i + chunk_size]


class _FakeSession:
    def __init__(self, data: bytes = b"ORIGIN-TILE-BYTES"):
        self.data = data
        self.calls = 0

    def get(self, url, stream=False, timeout=None):
        self.calls += 1
        return _FakeResp(self.data)


class _FakeCache:
    def __init__(self, hit: bool = False):
        self.hit = hit
        self.fetched: list[str] = []
        self.stored: list[str] = []

    def fetch(self, key, dest):
        self.fetched.append(key)
        if self.hit:
            dest.write_bytes(b"FROM-R2")
            return True
        return False

    def store(self, key, src):
        self.stored.append(key)


def test_l1_hit_uses_local_disk_and_skips_cache_and_origin(monkeypatch, tmp_path):
    fake = _FakeCache(hit=True)
    monkeypatch.setattr(lidar, "TILE_CACHE", fake)
    session = _FakeSession()
    dest = tmp_path / "t.las"
    dest.write_bytes(b"ALREADY-LOCAL")
    out = lidar._stream_download(session, "http://x/t.las", dest, "test", cache_key="douglas/t.las")
    assert out == dest and dest.read_bytes() == b"ALREADY-LOCAL"
    assert fake.fetched == [] and session.calls == 0   # neither R2 nor origin touched


def test_l2_hit_skips_origin_download(monkeypatch, tmp_path):
    fake = _FakeCache(hit=True)
    monkeypatch.setattr(lidar, "TILE_CACHE", fake)
    session = _FakeSession()
    dest = tmp_path / "t.las"
    out = lidar._stream_download(session, "http://x/t.las", dest, "test", cache_key="douglas/t.las")
    assert out == dest and dest.read_bytes() == b"FROM-R2"
    assert fake.fetched == ["douglas/t.las"]
    assert session.calls == 0          # origin never hit — the whole point of A3


def test_origin_download_populates_the_cache(monkeypatch, tmp_path):
    fake = _FakeCache(hit=False)       # R2 miss
    monkeypatch.setattr(lidar, "TILE_CACHE", fake)
    session = _FakeSession(b"ORIGIN-TILE-BYTES")
    dest = tmp_path / "t.las"
    out = lidar._stream_download(session, "http://x/t.las", dest, "test", cache_key="douglas/t.las")
    assert out == dest and dest.read_bytes() == b"ORIGIN-TILE-BYTES"
    assert fake.fetched == ["douglas/t.las"]   # checked R2 first (miss)
    assert session.calls == 1                  # then fell through to origin
    assert fake.stored == ["douglas/t.las"]    # and populated R2 for next cold start


def test_no_cache_key_means_local_only(monkeypatch, tmp_path):
    # The tile-index zip passes no cache_key -> R2 is never consulted or written.
    fake = _FakeCache(hit=True)
    monkeypatch.setattr(lidar, "TILE_CACHE", fake)
    session = _FakeSession(b"IDX")
    dest = tmp_path / "index.zip"
    lidar._stream_download(session, "http://x/index.zip", dest, "test")
    assert session.calls == 1 and fake.fetched == [] and fake.stored == []