200 lines
6.8 KiB
Python
200 lines
6.8 KiB
Python
import os
|
|
import io
|
|
import uuid
|
|
import asyncio
|
|
from time import sleep
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
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.template_search import TemplateSearchService # noqa: E402
|
|
|
|
|
|
def _dummy_docx_bytes():
|
|
try:
|
|
from docx import Document # type: ignore
|
|
except Exception:
|
|
return b"PK\x03\x04"
|
|
d = Document()
|
|
p = d.add_paragraph()
|
|
p.add_run("Cache Test ")
|
|
p.add_run("{{TOKEN}}")
|
|
buf = io.BytesIO()
|
|
d.save(buf)
|
|
return buf.getvalue()
|
|
|
|
|
|
def _upload_template(client: TestClient, name: str, category: str = "GENERAL") -> int:
|
|
files = {"file": (f"{name}.docx", _dummy_docx_bytes(), "application/vnd.openxmlformats-officedocument.wordprocessingml.document")}
|
|
resp = client.post(
|
|
"/api/templates/upload",
|
|
data={"name": name, "category": category, "semantic_version": "1.0.0"},
|
|
files=files,
|
|
)
|
|
assert resp.status_code == 200, resp.text
|
|
return int(resp.json()["id"])
|
|
|
|
|
|
@pytest.fixture()
|
|
def client_no_redis():
|
|
class _User:
|
|
def __init__(self):
|
|
self.id = "tmpl-cache-user"
|
|
self.username = "tester"
|
|
self.is_admin = True
|
|
self.is_active = True
|
|
|
|
app.dependency_overrides[get_current_user] = lambda: _User()
|
|
# Disable Redis to exercise in-memory fallback
|
|
settings.cache_enabled = False
|
|
settings.redis_url = ""
|
|
# Clear template caches
|
|
try:
|
|
asyncio.run(TemplateSearchService.invalidate_all())
|
|
except Exception:
|
|
pass
|
|
try:
|
|
yield TestClient(app)
|
|
finally:
|
|
app.dependency_overrides.pop(get_current_user, None)
|
|
|
|
|
|
def test_templates_search_caches_in_memory(client_no_redis: TestClient):
|
|
# Upload in unique category to avoid interference
|
|
category = f"CAT_{uuid.uuid4().hex[:8]}"
|
|
_upload_template(client_no_redis, f"T-{uuid.uuid4().hex[:6]}", category)
|
|
|
|
params = {"category": category}
|
|
r1 = client_no_redis.get("/api/templates/search", params=params)
|
|
assert r1.status_code == 200
|
|
d1 = r1.json()
|
|
|
|
# Second call should be served from cache and be identical
|
|
r2 = client_no_redis.get("/api/templates/search", params=params)
|
|
assert r2.status_code == 200
|
|
d2 = r2.json()
|
|
assert d1 == d2
|
|
|
|
|
|
def test_templates_search_invalidation_on_upload_in_memory(client_no_redis: TestClient):
|
|
# Use unique empty category; first query caches empty list
|
|
category = f"EMPTY_{uuid.uuid4().hex[:8]}"
|
|
r_empty = client_no_redis.get("/api/templates/search", params={"category": category})
|
|
assert r_empty.status_code == 200
|
|
assert r_empty.json() == []
|
|
|
|
# Upload a template into that category -> triggers invalidation
|
|
_upload_template(client_no_redis, f"New-{uuid.uuid4().hex[:6]}", category)
|
|
sleep(0.05)
|
|
|
|
# Query again should reflect the new item (cache invalidated)
|
|
r_after = client_no_redis.get("/api/templates/search", params={"category": category})
|
|
assert r_after.status_code == 200
|
|
ids = [it["id"] for it in r_after.json()]
|
|
assert len(ids) >= 1
|
|
|
|
|
|
def test_templates_search_invalidation_on_keyword_update_in_memory(client_no_redis: TestClient):
|
|
category = f"KW_{uuid.uuid4().hex[:8]}"
|
|
tid = _upload_template(client_no_redis, f"KW-{uuid.uuid4().hex[:6]}", category)
|
|
|
|
# Initial query has_keywords=true should be empty and cached
|
|
r1 = client_no_redis.get("/api/templates/search", params={"category": category, "has_keywords": True})
|
|
assert r1.status_code == 200
|
|
assert r1.json() == []
|
|
|
|
# Add a keyword to the template -> invalidates caches
|
|
r_add = client_no_redis.post(f"/api/templates/{tid}/keywords", json={"keywords": ["alpha"]})
|
|
assert r_add.status_code == 200
|
|
sleep(0.05)
|
|
|
|
# Now search with has_keywords=true should include our template
|
|
r2 = client_no_redis.get("/api/templates/search", params={"category": category, "has_keywords": True})
|
|
assert r2.status_code == 200
|
|
ids2 = {it["id"] for it in r2.json()}
|
|
assert tid in ids2
|
|
|
|
|
|
@pytest.fixture()
|
|
def client_with_redis():
|
|
if not settings.redis_url:
|
|
pytest.skip("Redis not configured for caching tests")
|
|
|
|
class _User:
|
|
def __init__(self):
|
|
self.id = "tmpl-cache-user-redis"
|
|
self.username = "tester"
|
|
self.is_admin = True
|
|
self.is_active = True
|
|
|
|
app.dependency_overrides[get_current_user] = lambda: _User()
|
|
settings.cache_enabled = True
|
|
# Clear template caches
|
|
try:
|
|
asyncio.run(TemplateSearchService.invalidate_all())
|
|
except Exception:
|
|
pass
|
|
try:
|
|
yield TestClient(app)
|
|
finally:
|
|
app.dependency_overrides.pop(get_current_user, None)
|
|
|
|
|
|
def test_templates_search_caches_with_redis(client_with_redis: TestClient):
|
|
category = f"RCAT_{uuid.uuid4().hex[:8]}"
|
|
_upload_template(client_with_redis, f"RT-{uuid.uuid4().hex[:6]}", category)
|
|
|
|
params = {"category": category}
|
|
r1 = client_with_redis.get("/api/templates/search", params=params)
|
|
assert r1.status_code == 200
|
|
d1 = r1.json()
|
|
|
|
r2 = client_with_redis.get("/api/templates/search", params=params)
|
|
assert r2.status_code == 200
|
|
d2 = r2.json()
|
|
assert d1 == d2
|
|
|
|
|
|
def test_templates_search_invalidation_on_upload_with_redis(client_with_redis: TestClient):
|
|
category = f"RADD_{uuid.uuid4().hex[:8]}"
|
|
r_empty = client_with_redis.get("/api/templates/search", params={"category": category})
|
|
assert r_empty.status_code == 200 and r_empty.json() == []
|
|
|
|
_upload_template(client_with_redis, f"RNew-{uuid.uuid4().hex[:6]}", category)
|
|
sleep(0.05)
|
|
|
|
r_after = client_with_redis.get("/api/templates/search", params={"category": category})
|
|
assert r_after.status_code == 200
|
|
assert len(r_after.json()) >= 1
|
|
|
|
|
|
def test_templates_search_invalidation_on_keyword_update_with_redis(client_with_redis: TestClient):
|
|
category = f"RKW_{uuid.uuid4().hex[:8]}"
|
|
tid = _upload_template(client_with_redis, f"RTKW-{uuid.uuid4().hex[:6]}", category)
|
|
|
|
r1 = client_with_redis.get("/api/templates/search", params={"category": category, "has_keywords": True})
|
|
assert r1.status_code == 200 and r1.json() == []
|
|
|
|
r_add = client_with_redis.post(f"/api/templates/{tid}/keywords", json={"keywords": ["beta"]})
|
|
assert r_add.status_code == 200
|
|
sleep(0.05)
|
|
|
|
r2 = client_with_redis.get("/api/templates/search", params={"category": category, "has_keywords": True})
|
|
ids = {it["id"] for it in r2.json()}
|
|
assert tid in ids
|
|
|
|
|