83 lines
3.1 KiB
Python
83 lines
3.1 KiB
Python
"""Unit tests for the SQLite web cache (backend.tools.webcache, #92)."""
|
|
|
|
import pytest
|
|
|
|
from backend.tools import webcache
|
|
|
|
|
|
@pytest.fixture
|
|
def cache_enabled(tmp_path, monkeypatch):
|
|
"""Enable the cache against an isolated file with a short TTL."""
|
|
monkeypatch.setenv("OBSIGATE_WEB_CACHE_PATH", str(tmp_path / "cache.sqlite3"))
|
|
monkeypatch.setenv("OBSIGATE_WEB_CACHE_TTL", "60")
|
|
webcache._schema_ready = False
|
|
yield
|
|
webcache._schema_ready = False
|
|
|
|
|
|
class TestCacheKey:
|
|
def test_deterministic_and_payload_sensitive(self):
|
|
a = webcache.cache_key("search", {"q": "pizza", "page": 1})
|
|
b = webcache.cache_key("search", {"page": 1, "q": "pizza"})
|
|
c = webcache.cache_key("search", {"q": "pasta", "page": 1})
|
|
d = webcache.cache_key("fetch", {"q": "pizza", "page": 1})
|
|
assert a == b
|
|
assert a != c
|
|
assert a != d
|
|
|
|
|
|
class TestCacheRoundTrip:
|
|
def test_set_get_roundtrip(self, cache_enabled):
|
|
key = webcache.cache_key("search", {"q": "x"})
|
|
webcache.cache_set(key, {"results": [1, 2, 3], "provider": "tavily"})
|
|
assert webcache.cache_get(key) == {"results": [1, 2, 3], "provider": "tavily"}
|
|
|
|
def test_miss_returns_none(self, cache_enabled):
|
|
assert webcache.cache_get("search:unknown") is None
|
|
|
|
def test_overwrite_updates_value(self, cache_enabled):
|
|
key = webcache.cache_key("search", {"q": "x"})
|
|
webcache.cache_set(key, {"v": 1})
|
|
webcache.cache_set(key, {"v": 2})
|
|
assert webcache.cache_get(key) == {"v": 2}
|
|
|
|
def test_ttl_expiry(self, tmp_path, monkeypatch):
|
|
monkeypatch.setenv("OBSIGATE_WEB_CACHE_PATH", str(tmp_path / "cache.sqlite3"))
|
|
monkeypatch.setenv("OBSIGATE_WEB_CACHE_TTL", "0.05")
|
|
webcache._schema_ready = False
|
|
key = webcache.cache_key("search", {"q": "x"})
|
|
webcache.cache_set(key, {"v": 1})
|
|
assert webcache.cache_get(key) == {"v": 1}
|
|
import time
|
|
|
|
time.sleep(0.15)
|
|
assert webcache.cache_get(key) is None
|
|
|
|
def test_disabled_when_ttl_zero(self, cache_enabled, monkeypatch):
|
|
monkeypatch.setenv("OBSIGATE_WEB_CACHE_TTL", "0")
|
|
key = webcache.cache_key("search", {"q": "x"})
|
|
webcache.cache_set(key, {"v": 1})
|
|
assert webcache.cache_get(key) is None
|
|
|
|
def test_purge_expired(self, cache_enabled, monkeypatch):
|
|
key = webcache.cache_key("search", {"q": "x"})
|
|
webcache.cache_set(key, {"v": 1})
|
|
monkeypatch.setenv("OBSIGATE_WEB_CACHE_TTL", "0.05")
|
|
import time
|
|
|
|
time.sleep(0.15)
|
|
assert webcache.purge_expired() >= 1
|
|
assert webcache.cache_get(key) is None
|
|
|
|
def test_corrupt_db_degrades_silently(self, tmp_path, monkeypatch):
|
|
# A directory as cache file breaks sqlite3.connect: the cache must
|
|
# disable itself instead of breaking the tools.
|
|
monkeypatch.setenv("OBSIGATE_WEB_CACHE_PATH", str(tmp_path))
|
|
monkeypatch.setenv("OBSIGATE_WEB_CACHE_TTL", "60")
|
|
webcache._schema_ready = False
|
|
try:
|
|
assert webcache.cache_get("search:x") is None
|
|
webcache.cache_set("search:x", {"v": 1})
|
|
finally:
|
|
webcache._schema_ready = False
|