thermograph/tests/test_store.py
Emi Griffith 4ac5323375 Add backend test suite; gate direct pushes; serialize LAN deploys (#41)
- backend/tests: 74 hermetic tests (no network, no repo data//logs/ writes)
  covering grid snapping/round-trips, grading percentiles/bands/windows/
  dry streaks, the places index (norm, one-edit matchers, search,
  corrections), the derived store (token validity, cache=False, degraded
  mode), and route-level API tests over a faked climate layer — routing,
  validation, ETag/304 revalidation, store replay, the /cell bundle, and
  the v1/v2 aliases. The API tests would have caught the /place
  AttributeError regression.
- requirements-dev.txt + make test (venv prefers uv-pinned 3.12, matching
  deploy-dev.sh — pyarrow wheels stop at 3.12 and some pyenv builds lack
  sqlite).
- CI: extract the build job into a reusable build.yml, add the test run
  and an API health probe (page-only curl can't catch route wiring
  faults); deploy-dev.yml now runs the same build gate before deploying
  direct pushes, which previously deployed with no CI at all.
- Deploys serialize under one dev-lan-deploy concurrency group across
  both workflows (previously per-PR groups could interleave two deploys
  to the same checkout), and are never cancelled mid-restart.
- deploy-dev.sh health check also probes /api/v2/place — best-effort
  externals mean a failure there is a genuine server bug.
2026-07-11 19:37:49 +00:00

65 lines
2.8 KiB
Python

import json
import sqlite3
import time
def test_payload_round_trip(tmp_store):
payload = {"cell": "1_2", "days": [1, 2, 3]}
body = tmp_store.put_payload("grade", "1_2", "k", "tok", payload)
assert json.loads(body) == payload
assert tmp_store.get_payload("grade", "1_2", "k", "tok") == body
assert tmp_store.get_json("grade", "1_2", "k", "tok") == payload
def test_token_mismatch_is_a_miss(tmp_store):
tmp_store.put_payload("grade", "1_2", "k", "tok-a", {"v": 1})
assert tmp_store.get_payload("grade", "1_2", "k", "tok-b") is None
# A rewrite under the new token replaces the row (same primary key).
tmp_store.put_payload("grade", "1_2", "k", "tok-b", {"v": 2})
assert tmp_store.get_json("grade", "1_2", "k", "tok-b") == {"v": 2}
assert tmp_store.get_payload("grade", "1_2", "k", "tok-a") is None
def test_cache_false_serves_without_persisting(tmp_store):
body = tmp_store.put_payload("grade", "1_2", "k", "tok", {"v": 1}, cache=False)
assert json.loads(body) == {"v": 1}
assert tmp_store.get_payload("grade", "1_2", "k", "tok") is None
def test_put_payload_rejects_nan(tmp_store):
import pytest
with pytest.raises(ValueError):
tmp_store.put_payload("grade", "1_2", "k", "tok", {"v": float("nan")})
def test_revgeo_round_trip(tmp_store):
key = tmp_store.revgeo_key(47.60623, -122.33305)
assert key == "47.606,-122.333"
assert tmp_store.get_revgeo(key) == (False, None)
tmp_store.put_revgeo(key, "Belltown, Seattle, Washington")
assert tmp_store.get_revgeo(key) == (True, "Belltown, Seattle, Washington")
def test_revgeo_cached_miss_retries_after_ttl(tmp_store):
key = tmp_store.revgeo_key(1.0, 2.0)
tmp_store.put_revgeo(key, None)
assert tmp_store.get_revgeo(key) == (True, None) # fresh miss: don't re-ask yet
# Age the row past the TTL directly in SQLite.
conn = sqlite3.connect(tmp_store.DB_PATH)
conn.execute("UPDATE revgeo SET updated_at=? WHERE key=?",
(time.time() - tmp_store.REVGEO_MISS_TTL - 1, key))
conn.commit()
conn.close()
assert tmp_store.get_revgeo(key) == (False, None) # stale miss: retryable
def test_store_is_a_pure_accelerator_when_db_unavailable(tmp_store, monkeypatch):
# Point the store somewhere unwritable: every helper degrades, none raises.
monkeypatch.setattr(tmp_store, "DB_PATH", "/proc/nope/store.sqlite")
import threading
monkeypatch.setattr(tmp_store, "_local", threading.local())
assert tmp_store.get_payload("grade", "c", "k", "t") is None
body = tmp_store.put_payload("grade", "c", "k", "t", {"v": 1})
assert json.loads(body) == {"v": 1} # still serves the encoded payload
assert tmp_store.get_revgeo("x") == (False, None)
tmp_store.put_revgeo("x", "label") # no-op, no exception