Spaces:
Running
Running
File size: 5,052 Bytes
1f279fb | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 | """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 == []
|