sepehr
dd1e005c70
All checks were successful
Deploy to Production / Build and Deploy (push) Successful in 3m13s
532 lines
18 KiB
Python
532 lines
18 KiB
Python
"""
|
|
Tests for Track C2 — Redis translation cache.
|
|
|
|
Covers:
|
|
- make_cache_key: stable across calls, distinct inputs → distinct keys
|
|
- hash_prompt / hash_glossary_terms
|
|
- _LRUBackend: get/set, eviction order, stats
|
|
- _RedisBackend: connection probe, fallback to LRU on error
|
|
- TranslationCache: auto backend selection, user_id namespacing,
|
|
prompt/glossary isolation
|
|
- get_cache: singleton, reset_cache_for_tests
|
|
"""
|
|
import os
|
|
import sys
|
|
import threading
|
|
import time
|
|
from pathlib import Path
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
from services.translation_cache import (
|
|
make_cache_key,
|
|
hash_prompt,
|
|
hash_glossary_terms,
|
|
_LRUBackend,
|
|
_RedisBackend,
|
|
TranslationCache,
|
|
get_cache,
|
|
reset_cache_for_tests,
|
|
)
|
|
|
|
|
|
# ============================================================================
|
|
# make_cache_key
|
|
# ============================================================================
|
|
|
|
class TestMakeCacheKey:
|
|
def test_stable_across_calls(self):
|
|
k1 = make_cache_key("Hello", "fr", "en", "google")
|
|
k2 = make_cache_key("Hello", "fr", "en", "google")
|
|
assert k1 == k2
|
|
|
|
def test_different_target_different_key(self):
|
|
k1 = make_cache_key("Hello", "fr", "en", "google")
|
|
k2 = make_cache_key("Hello", "es", "en", "google")
|
|
assert k1 != k2
|
|
|
|
def test_different_provider_different_key(self):
|
|
k1 = make_cache_key("Hello", "fr", "en", "google")
|
|
k2 = make_cache_key("Hello", "fr", "en", "deepl")
|
|
assert k1 != k2
|
|
|
|
def test_different_text_different_key(self):
|
|
k1 = make_cache_key("Hello", "fr", "en", "google")
|
|
k2 = make_cache_key("Goodbye", "fr", "en", "google")
|
|
assert k1 != k2
|
|
|
|
def test_user_namespacing(self):
|
|
k1 = make_cache_key("Hello", "fr", "en", "google", user_id="alice")
|
|
k2 = make_cache_key("Hello", "fr", "en", "google", user_id="bob")
|
|
assert k1 != k2, "Different users should not share cache entries"
|
|
|
|
def test_same_user_same_key(self):
|
|
k1 = make_cache_key("Hello", "fr", "en", "google", user_id="alice")
|
|
k2 = make_cache_key("Hello", "fr", "en", "google", user_id="alice")
|
|
assert k1 == k2
|
|
|
|
def test_anon_default(self):
|
|
k1 = make_cache_key("Hello", "fr", "en", "google")
|
|
k2 = make_cache_key("Hello", "fr", "en", "google", user_id=None)
|
|
assert k1 == k2, "user_id=None should equal the default anonymous user"
|
|
|
|
def test_prompt_isolation(self):
|
|
k1 = make_cache_key("Hello", "fr", "en", "google", custom_prompt_hash="abc")
|
|
k2 = make_cache_key("Hello", "fr", "en", "google", custom_prompt_hash="xyz")
|
|
assert k1 != k2, "Different prompts should not share cache entries"
|
|
|
|
def test_glossary_isolation(self):
|
|
k1 = make_cache_key("Hello", "fr", "en", "google", glossary_hash="g1")
|
|
k2 = make_cache_key("Hello", "fr", "en", "google", glossary_hash="g2")
|
|
assert k1 != k2
|
|
|
|
def test_returns_hex_string(self):
|
|
k = make_cache_key("Hello", "fr", "en", "google")
|
|
assert isinstance(k, str)
|
|
assert len(k) == 64 # sha256 hex digest length
|
|
assert all(c in "0123456789abcdef" for c in k)
|
|
|
|
|
|
# ============================================================================
|
|
# Hash helpers
|
|
# ============================================================================
|
|
|
|
class TestHashHelpers:
|
|
def test_hash_prompt_none(self):
|
|
assert hash_prompt(None) is None
|
|
|
|
def test_hash_prompt_empty(self):
|
|
assert hash_prompt("") is None
|
|
|
|
def test_hash_prompt_stable(self):
|
|
h1 = hash_prompt("Translate this")
|
|
h2 = hash_prompt("Translate this")
|
|
assert h1 == h2
|
|
|
|
def test_hash_prompt_different_inputs(self):
|
|
h1 = hash_prompt("A")
|
|
h2 = hash_prompt("B")
|
|
assert h1 != h2
|
|
|
|
def test_hash_prompt_short(self):
|
|
h = hash_prompt("test")
|
|
# We truncate to 16 chars in the implementation
|
|
assert len(h) == 16
|
|
|
|
def test_hash_glossary_none(self):
|
|
assert hash_glossary_terms(None) is None
|
|
|
|
def test_hash_glossary_empty(self):
|
|
assert hash_glossary_terms([]) is None
|
|
|
|
def test_hash_glossary_stable(self):
|
|
terms = [{"source": "Hello", "target": "Bonjour"}]
|
|
h1 = hash_glossary_terms(terms)
|
|
h2 = hash_glossary_terms(terms)
|
|
assert h1 == h2
|
|
|
|
def test_hash_glossary_order_invariant(self):
|
|
"""The order of terms should not change the hash (sorted internally)."""
|
|
a = [
|
|
{"source": "Hello", "target": "Bonjour"},
|
|
{"source": "World", "target": "Monde"},
|
|
]
|
|
b = [
|
|
{"source": "World", "target": "Monde"},
|
|
{"source": "Hello", "target": "Bonjour"},
|
|
]
|
|
assert hash_glossary_terms(a) == hash_glossary_terms(b)
|
|
|
|
def test_hash_glossary_metadata_invariant(self):
|
|
"""The name/id of the glossary should not affect the hash."""
|
|
a = [{"source": "X", "target": "Y", "name": "g1", "id": "g1"}]
|
|
b = [{"source": "X", "target": "Y", "name": "g2", "id": "g2"}]
|
|
assert hash_glossary_terms(a) == hash_glossary_terms(b)
|
|
|
|
|
|
# ============================================================================
|
|
# LRU backend
|
|
# ============================================================================
|
|
|
|
class TestLRUBackend:
|
|
def test_get_missing_returns_none(self):
|
|
c = _LRUBackend()
|
|
assert c.get("nope") is None
|
|
assert c.misses == 1
|
|
|
|
def test_set_and_get(self):
|
|
c = _LRUBackend()
|
|
c.set("k", "v")
|
|
assert c.get("k") == "v"
|
|
assert c.hits == 1
|
|
|
|
def test_overwrite(self):
|
|
c = _LRUBackend()
|
|
c.set("k", "v1")
|
|
c.set("k", "v2")
|
|
assert c.get("k") == "v2"
|
|
assert len(c._cache) == 1
|
|
|
|
def test_lru_eviction(self):
|
|
c = _LRUBackend(maxsize=2)
|
|
c.set("a", "1")
|
|
c.set("b", "2")
|
|
c.set("c", "3") # Should evict "a" (least recently used)
|
|
assert c.get("a") is None
|
|
assert c.get("b") == "2"
|
|
assert c.get("c") == "3"
|
|
|
|
def test_get_promotes_to_mru(self):
|
|
c = _LRUBackend(maxsize=2)
|
|
c.set("a", "1")
|
|
c.set("b", "2")
|
|
c.get("a") # Promote "a"
|
|
c.set("c", "3") # Should evict "b" (now LRU)
|
|
assert c.get("a") == "1"
|
|
assert c.get("b") is None
|
|
assert c.get("c") == "3"
|
|
|
|
def test_clear(self):
|
|
c = _LRUBackend()
|
|
c.set("a", "1")
|
|
c.set("b", "2")
|
|
c.clear()
|
|
assert c.get("a") is None
|
|
# After clear, both counters are reset
|
|
assert c.hits == 0
|
|
# The get after clear is a miss (1)
|
|
assert c.misses == 1
|
|
|
|
def test_stats(self):
|
|
c = _LRUBackend(maxsize=10)
|
|
c.set("a", "1")
|
|
c.get("a") # hit
|
|
c.get("b") # miss
|
|
s = c.stats()
|
|
assert s["backend"] == "lru"
|
|
assert s["hits"] == 1
|
|
assert s["misses"] == 1
|
|
assert s["size"] == 1
|
|
assert s["maxsize"] == 10
|
|
|
|
def test_thread_safety(self):
|
|
c = _LRUBackend(maxsize=100)
|
|
errors = []
|
|
|
|
def worker(i):
|
|
try:
|
|
for j in range(100):
|
|
c.set(f"k_{i}_{j}", f"v_{i}_{j}")
|
|
c.get(f"k_{i}_{j}")
|
|
except Exception as e:
|
|
errors.append(e)
|
|
|
|
threads = [threading.Thread(target=worker, args=(i,)) for i in range(5)]
|
|
for t in threads:
|
|
t.start()
|
|
for t in threads:
|
|
t.join()
|
|
assert errors == []
|
|
|
|
|
|
# ============================================================================
|
|
# Redis backend — fallback behavior
|
|
# ============================================================================
|
|
|
|
class TestRedisBackendFallback:
|
|
"""When Redis is unavailable, the backend should fall back to LRU."""
|
|
|
|
def test_redis_unavailable_falls_back_to_lru(self):
|
|
lru = _LRUBackend()
|
|
backend = _RedisBackend(
|
|
url="redis://invalid-host:9999/0",
|
|
lru_fallback=lru,
|
|
)
|
|
# First call probes the connection, fails, and falls back
|
|
assert backend.get("test_key") is None # miss in LRU
|
|
backend.set("test_key", "hello")
|
|
# Second get should hit the LRU
|
|
assert backend.get("test_key") == "hello"
|
|
# Errors should have been counted
|
|
assert backend.errors >= 1
|
|
# The backend should not have a client
|
|
assert backend._client is None
|
|
|
|
def test_redis_set_falls_back_to_lru(self):
|
|
lru = _LRUBackend()
|
|
backend = _RedisBackend(
|
|
url="redis://invalid-host:9999/0",
|
|
lru_fallback=lru,
|
|
)
|
|
backend.set("k", "v")
|
|
# The LRU should have it
|
|
assert lru.get("k") == "v"
|
|
|
|
def test_stats_includes_lru(self):
|
|
lru = _LRUBackend()
|
|
backend = _RedisBackend(
|
|
url="redis://invalid-host:9999/0",
|
|
lru_fallback=lru,
|
|
)
|
|
backend.set("k", "v")
|
|
stats = backend.stats()
|
|
assert stats["backend"] == "redis"
|
|
assert stats["available"] is False
|
|
assert stats["errors"] >= 1
|
|
assert "lru_size" in stats
|
|
|
|
|
|
# ============================================================================
|
|
# Redis backend — happy path with a mock
|
|
# ============================================================================
|
|
|
|
class TestRedisBackendWithMock:
|
|
"""When Redis is reachable, the backend should use it."""
|
|
|
|
def _build_backend_with_mock_client(self, client):
|
|
"""Build a RedisBackend with a pre-set mock client."""
|
|
lru = _LRUBackend()
|
|
backend = _RedisBackend(
|
|
url="redis://fake:6379/0",
|
|
lru_fallback=lru,
|
|
)
|
|
backend._client = client
|
|
backend._available = True
|
|
return backend, lru
|
|
|
|
def test_get_uses_redis(self):
|
|
client = MagicMock()
|
|
client.get.return_value = "cached_value"
|
|
backend, lru = self._build_backend_with_mock_client(client)
|
|
|
|
result = backend.get("k")
|
|
assert result == "cached_value"
|
|
client.get.assert_called_once()
|
|
# Should also have been written to LRU
|
|
assert lru.get("k") == "cached_value"
|
|
|
|
def test_get_miss_redis(self):
|
|
client = MagicMock()
|
|
client.get.return_value = None
|
|
backend, lru = self._build_backend_with_mock_client(client)
|
|
|
|
result = backend.get("k")
|
|
assert result is None
|
|
assert backend.misses == 1
|
|
|
|
def test_set_uses_redis(self):
|
|
client = MagicMock()
|
|
backend, lru = self._build_backend_with_mock_client(client)
|
|
|
|
backend.set("k", "v", ttl=60)
|
|
# Should have called Redis SET
|
|
client.set.assert_called_once()
|
|
args = client.set.call_args[0]
|
|
assert args[1] == "v" # value
|
|
# The TTL is passed as a keyword arg (ex=ttl)
|
|
assert client.set.call_args[1]["ex"] == 60
|
|
|
|
def test_set_uses_default_ttl(self):
|
|
client = MagicMock()
|
|
backend, lru = self._build_backend_with_mock_client(client)
|
|
|
|
backend.set("k", "v")
|
|
# Default TTL = 86400 (24h) is passed as ex=...
|
|
assert client.set.call_args[1]["ex"] == 86400
|
|
|
|
def test_redis_error_during_get_resets_client(self):
|
|
client = MagicMock()
|
|
client.get.side_effect = RuntimeError("redis down")
|
|
backend, lru = self._build_backend_with_mock_client(client)
|
|
|
|
result = backend.get("k")
|
|
# Falls back to LRU
|
|
assert result is None
|
|
# Client should be reset so we re-probe next time
|
|
assert backend._client is None
|
|
assert backend._available is False
|
|
assert backend.errors == 1
|
|
|
|
def test_redis_error_during_set_resets_client(self):
|
|
client = MagicMock()
|
|
client.set.side_effect = RuntimeError("redis down")
|
|
backend, lru = self._build_backend_with_mock_client(client)
|
|
|
|
backend.set("k", "v")
|
|
# Should have fallen back to LRU
|
|
assert lru.get("k") == "v"
|
|
assert backend._client is None
|
|
assert backend.errors == 1
|
|
|
|
def test_namespace_prefix(self):
|
|
client = MagicMock()
|
|
client.get.return_value = None
|
|
backend, lru = self._build_backend_with_mock_client(client)
|
|
|
|
backend.get("mykey")
|
|
called_key = client.get.call_args[0][0]
|
|
assert called_key.startswith("translate_cache:")
|
|
assert called_key.endswith(":mykey")
|
|
|
|
def test_custom_namespace(self):
|
|
client = MagicMock()
|
|
client.get.return_value = None
|
|
lru = _LRUBackend()
|
|
backend = _RedisBackend(
|
|
url="redis://fake:6379/0",
|
|
namespace="custom_ns",
|
|
lru_fallback=lru,
|
|
)
|
|
backend._client = client
|
|
backend._available = True
|
|
|
|
backend.get("mykey")
|
|
called_key = client.get.call_args[0][0]
|
|
assert called_key.startswith("custom_ns:")
|
|
|
|
|
|
# ============================================================================
|
|
# TranslationCache — high-level API
|
|
# ============================================================================
|
|
|
|
class TestTranslationCache:
|
|
def test_default_backend_is_lru(self, monkeypatch):
|
|
"""No REDIS_CACHE_ENABLED → LRU backend."""
|
|
monkeypatch.delenv("REDIS_CACHE_ENABLED", raising=False)
|
|
monkeypatch.delenv("REDIS_URL", raising=False)
|
|
cache = TranslationCache(backend="auto")
|
|
assert cache.backend_name == "lru"
|
|
|
|
def test_explicit_lru_backend(self, monkeypatch):
|
|
monkeypatch.setenv("REDIS_CACHE_ENABLED", "true")
|
|
cache = TranslationCache(backend="lru")
|
|
assert cache.backend_name == "lru"
|
|
|
|
def test_get_and_set(self):
|
|
cache = TranslationCache(backend="lru")
|
|
cache.set("Hello", "fr", "en", "google", "Bonjour")
|
|
assert cache.get("Hello", "fr", "en", "google") == "Bonjour"
|
|
|
|
def test_get_miss(self):
|
|
cache = TranslationCache(backend="lru")
|
|
assert cache.get("Nope", "fr", "en", "google") is None
|
|
|
|
def test_user_isolation(self):
|
|
cache = TranslationCache(backend="lru")
|
|
cache.set("Hello", "fr", "en", "google", "Bonjour", user_id="alice")
|
|
cache.set("Hello", "fr", "en", "google", "Hola", user_id="bob")
|
|
assert cache.get("Hello", "fr", "en", "google", user_id="alice") == "Bonjour"
|
|
assert cache.get("Hello", "fr", "en", "google", user_id="bob") == "Hola"
|
|
|
|
def test_prompt_isolation(self):
|
|
cache = TranslationCache(backend="lru")
|
|
cache.set("Hello", "fr", "en", "google", "V1",
|
|
custom_prompt_hash="p1")
|
|
cache.set("Hello", "fr", "en", "google", "V2",
|
|
custom_prompt_hash="p2")
|
|
assert cache.get("Hello", "fr", "en", "google", custom_prompt_hash="p1") == "V1"
|
|
assert cache.get("Hello", "fr", "en", "google", custom_prompt_hash="p2") == "V2"
|
|
|
|
def test_glossary_isolation(self):
|
|
cache = TranslationCache(backend="lru")
|
|
cache.set("Hello", "fr", "en", "google", "V1",
|
|
glossary_hash="g1")
|
|
cache.set("Hello", "fr", "en", "google", "V2",
|
|
glossary_hash="g2")
|
|
assert cache.get("Hello", "fr", "en", "google", glossary_hash="g1") == "V1"
|
|
assert cache.get("Hello", "fr", "en", "google", glossary_hash="g2") == "V2"
|
|
|
|
def test_clear(self):
|
|
cache = TranslationCache(backend="lru")
|
|
cache.set("k", "fr", "en", "google", "v")
|
|
cache.clear()
|
|
assert cache.get("k", "fr", "en", "google") is None
|
|
|
|
def test_stats_returns_dict(self):
|
|
cache = TranslationCache(backend="lru")
|
|
cache.set("k", "fr", "en", "google", "v")
|
|
cache.get("k", "fr", "en", "google")
|
|
s = cache.stats()
|
|
assert isinstance(s, dict)
|
|
assert s["backend"] == "lru"
|
|
|
|
|
|
# ============================================================================
|
|
# Singleton accessor
|
|
# ============================================================================
|
|
|
|
class TestGetCache:
|
|
def test_returns_singleton(self, monkeypatch):
|
|
reset_cache_for_tests()
|
|
c1 = get_cache()
|
|
c2 = get_cache()
|
|
assert c1 is c2
|
|
|
|
def test_reset_for_tests(self, monkeypatch):
|
|
reset_cache_for_tests()
|
|
c1 = get_cache()
|
|
reset_cache_for_tests()
|
|
c2 = get_cache()
|
|
assert c1 is not c2
|
|
|
|
def test_singleton_thread_safe(self, monkeypatch):
|
|
reset_cache_for_tests()
|
|
caches = []
|
|
errors = []
|
|
|
|
def worker():
|
|
try:
|
|
caches.append(get_cache())
|
|
except Exception as e:
|
|
errors.append(e)
|
|
|
|
threads = [threading.Thread(target=worker) for _ in range(10)]
|
|
for t in threads:
|
|
t.start()
|
|
for t in threads:
|
|
t.join()
|
|
assert errors == []
|
|
assert all(c is caches[0] for c in caches)
|
|
|
|
|
|
# ============================================================================
|
|
# End-to-end: integrate with hash_prompt / hash_glossary
|
|
# ============================================================================
|
|
|
|
class TestE2EHashIntegration:
|
|
def test_real_prompt_and_glossary(self):
|
|
"""A realistic scenario: same text, different prompt/glossary
|
|
should give different cache keys and thus different entries."""
|
|
cache = TranslationCache(backend="lru")
|
|
|
|
prompt_a = "Translate this as a formal document."
|
|
prompt_b = "Translate this as a casual chat message."
|
|
glossary_a = [{"source": "API", "target": "API"}]
|
|
glossary_b = [{"source": "API", "target": "interface"}]
|
|
|
|
cache.set(
|
|
"Hello", "fr", "en", "google",
|
|
"Bonjour (formel)",
|
|
custom_prompt_hash=hash_prompt(prompt_a),
|
|
glossary_hash=hash_glossary_terms(glossary_a),
|
|
)
|
|
cache.set(
|
|
"Hello", "fr", "en", "google",
|
|
"Salut (décontracté)",
|
|
custom_prompt_hash=hash_prompt(prompt_b),
|
|
glossary_hash=hash_glossary_terms(glossary_b),
|
|
)
|
|
|
|
# Reading with the same prompt/glossary should give the right entry
|
|
assert cache.get(
|
|
"Hello", "fr", "en", "google",
|
|
custom_prompt_hash=hash_prompt(prompt_a),
|
|
glossary_hash=hash_glossary_terms(glossary_a),
|
|
) == "Bonjour (formel)"
|
|
assert cache.get(
|
|
"Hello", "fr", "en", "google",
|
|
custom_prompt_hash=hash_prompt(prompt_b),
|
|
glossary_hash=hash_glossary_terms(glossary_b),
|
|
) == "Salut (décontracté)"
|