94 lines
2.8 KiB
Python
94 lines
2.8 KiB
Python
import os
|
|
import sys
|
|
from pathlib import Path
|
|
from time import sleep
|
|
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
|
|
# Ensure required env vars for app import/config
|
|
os.environ.setdefault("SECRET_KEY", "x" * 32)
|
|
os.environ.setdefault("DATABASE_URL", "sqlite:////tmp/delphi_test.sqlite")
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
if str(ROOT) not in sys.path:
|
|
sys.path.insert(0, str(ROOT))
|
|
|
|
from app.main import app # noqa: E402
|
|
from app.auth.security import get_current_user # noqa: E402
|
|
from app.config import settings # noqa: E402
|
|
from app.services.cache import invalidate_search_cache # noqa: E402
|
|
|
|
|
|
@pytest.fixture(scope="module")
|
|
def client():
|
|
class _User:
|
|
def __init__(self):
|
|
self.id = "cache-tester"
|
|
self.username = "tester"
|
|
self.is_admin = True
|
|
self.is_active = True
|
|
|
|
app.dependency_overrides[get_current_user] = lambda: _User()
|
|
|
|
# Enable cache for this test module if redis is configured
|
|
settings.cache_enabled = True
|
|
yield TestClient(app)
|
|
app.dependency_overrides.pop(get_current_user, None)
|
|
|
|
|
|
@pytest.mark.skipif(not settings.redis_url, reason="Redis not configured for caching tests")
|
|
def test_advanced_search_caches_by_criteria_and_user(client: TestClient):
|
|
criteria = {
|
|
"query": "cache-token",
|
|
"search_types": ["customer"],
|
|
"limit": 10,
|
|
"offset": 0,
|
|
}
|
|
|
|
# First call: cold cache
|
|
r1 = client.post("/api/search/advanced", json=criteria)
|
|
assert r1.status_code == 200
|
|
d1 = r1.json()
|
|
|
|
# Second call: should be served from cache and identical
|
|
r2 = client.post("/api/search/advanced", json=criteria)
|
|
assert r2.status_code == 200
|
|
d2 = r2.json()
|
|
assert d1 == d2
|
|
|
|
|
|
@pytest.mark.skipif(not settings.redis_url, reason="Redis not configured for caching tests")
|
|
def test_advanced_search_cache_invalidation_on_data_change(client: TestClient):
|
|
criteria = {
|
|
"query": "invalidate-token",
|
|
"search_types": ["customer"],
|
|
"limit": 10,
|
|
"offset": 0,
|
|
}
|
|
|
|
r1 = client.post("/api/search/advanced", json=criteria)
|
|
assert r1.status_code == 200
|
|
d1 = r1.json()
|
|
|
|
# Mutate data via customers API which triggers invalidation
|
|
resp = client.post("/api/customers/", json={
|
|
"id": "CACHE-INVALIDATE-1",
|
|
"last": "Cache",
|
|
"first": "Invalidate",
|
|
"email": "invalidate@example.com",
|
|
"city": "Austin",
|
|
"abrev": "TX",
|
|
})
|
|
assert resp.status_code == 200
|
|
|
|
# Best-effort async invalidation; give Redis a moment if needed
|
|
sleep(0.05)
|
|
|
|
r2 = client.post("/api/search/advanced", json=criteria)
|
|
assert r2.status_code == 200
|
|
d2 = r2.json()
|
|
# Total_results or results content may change; at minimum the payload should not be the same object
|
|
assert d1 != d2
|
|
|