21f7ef4d19
Move the flat backend/tests/*.py into domain subfolders so the suite mirrors the code's concerns: data/ climate, grading, scoring, grid, places, store web/ api, content, homepage, views notifications/ notify, digest, discord (+ dm/interactions/link) accounts/ api_accounts core/ metrics, singleton, dashboard conftest.py stays at the tests/ root, so its sys.path setup and shared fixtures still apply to every subfolder. The four tests that derive repo paths from __file__ get their depth bumped one level to match their new location. Pytest discovers the subfolders recursively; the CI command (python -m pytest backend/tests) is unchanged. Claude-Session: https://claude.ai/code/session_01XXxmNFy9cZ6Gh8Y9thZn62
65 lines
2.8 KiB
Python
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 == "v2:47.606,-122.333" # v2: label format now carries country
|
|
assert tmp_store.get_revgeo(key) == (False, None)
|
|
tmp_store.put_revgeo(key, "Belltown, Seattle, Washington, United States")
|
|
assert tmp_store.get_revgeo(key) == (True, "Belltown, Seattle, Washington, United States")
|
|
|
|
|
|
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
|