106 lines
3.3 KiB
Python
106 lines
3.3 KiB
Python
import time
|
|
|
|
from db import (
|
|
cache_key,
|
|
feed_cache_key,
|
|
get_feed_cache,
|
|
get_translation,
|
|
prune_translations,
|
|
set_feed_cache,
|
|
set_translation,
|
|
)
|
|
|
|
|
|
# ── cache_key / feed_cache_key ────────────────────────────────────────────────
|
|
|
|
def test_cache_key_is_deterministic():
|
|
assert cache_key("hello", "sk") == cache_key("hello", "sk")
|
|
|
|
|
|
def test_cache_key_differs_by_text():
|
|
assert cache_key("hello", "sk") != cache_key("world", "sk")
|
|
|
|
|
|
def test_cache_key_differs_by_lang():
|
|
assert cache_key("hello", "sk") != cache_key("hello", "cs")
|
|
|
|
|
|
def test_feed_cache_key_is_deterministic():
|
|
assert feed_cache_key("sk", "some/path") == feed_cache_key("sk", "some/path")
|
|
|
|
|
|
def test_feed_cache_key_differs_by_path():
|
|
assert feed_cache_key("sk", "path/a") != feed_cache_key("sk", "path/b")
|
|
|
|
|
|
def test_feed_cache_key_differs_by_lang():
|
|
assert feed_cache_key("sk", "path") != feed_cache_key("cs", "path")
|
|
|
|
|
|
# ── Translation cache ─────────────────────────────────────────────────────────
|
|
|
|
def test_get_translation_miss(db_conn):
|
|
assert get_translation(db_conn, "nonexistent", ttl=60) is None
|
|
|
|
|
|
def test_get_translation_fresh(db_conn):
|
|
set_translation(db_conn, "k1", "translated")
|
|
assert get_translation(db_conn, "k1", ttl=3600) == "translated"
|
|
|
|
|
|
def test_get_translation_expired(db_conn):
|
|
db_conn.execute(
|
|
"INSERT INTO cache (key, value, created_at) VALUES (?,?,?)",
|
|
("k2", "stale", 0.0),
|
|
)
|
|
db_conn.commit()
|
|
assert get_translation(db_conn, "k2", ttl=60) is None
|
|
|
|
|
|
def test_set_translation_overwrites(db_conn):
|
|
set_translation(db_conn, "k3", "first")
|
|
set_translation(db_conn, "k3", "second")
|
|
assert get_translation(db_conn, "k3", ttl=3600) == "second"
|
|
|
|
|
|
def test_prune_translations_removes_old_keeps_fresh(db_conn):
|
|
db_conn.execute(
|
|
"INSERT INTO cache (key, value, created_at) VALUES (?,?,?)",
|
|
("old", "value", 0.0),
|
|
)
|
|
db_conn.commit()
|
|
set_translation(db_conn, "fresh", "value")
|
|
|
|
prune_translations(db_conn, max_age=60)
|
|
|
|
assert get_translation(db_conn, "old", ttl=999999) is None
|
|
assert get_translation(db_conn, "fresh", ttl=3600) == "value"
|
|
|
|
|
|
# ── Feed cache ────────────────────────────────────────────────────────────────
|
|
|
|
def test_get_feed_cache_miss(db_conn):
|
|
assert get_feed_cache(db_conn, "nokey", ttl=300) is None
|
|
|
|
|
|
def test_get_feed_cache_fresh(db_conn):
|
|
now = time.time()
|
|
content = b"<rss/>"
|
|
set_feed_cache(db_conn, "fk1", content, now)
|
|
result = get_feed_cache(db_conn, "fk1", ttl=300)
|
|
assert result is not None
|
|
assert result[0] == content
|
|
assert abs(result[1] - now) < 0.01
|
|
|
|
|
|
def test_get_feed_cache_expired(db_conn):
|
|
set_feed_cache(db_conn, "fk2", b"old", fetched_at=0.0)
|
|
assert get_feed_cache(db_conn, "fk2", ttl=300) is None
|
|
|
|
|
|
def test_set_feed_cache_overwrites(db_conn):
|
|
now = time.time()
|
|
set_feed_cache(db_conn, "fk3", b"first", now - 10)
|
|
set_feed_cache(db_conn, "fk3", b"second", now)
|
|
result = get_feed_cache(db_conn, "fk3", ttl=300)
|
|
assert result[0] == b"second"
|