100 lines
3.3 KiB
Python
100 lines
3.3 KiB
Python
"""
|
|
Smoke tests for the FastAPI layer.
|
|
|
|
These deliberately do NOT require a live Postgres/pgvector or Ollama
|
|
instance - they assert that every endpoint is reachable and degrades
|
|
gracefully (returns a sensible JSON error/empty-result rather than a
|
|
500/crash) when those dependencies are unavailable, which is the
|
|
situation on a fresh checkout before `.env` + the database are set up.
|
|
|
|
For tests that exercise real retrieval/generation against a live
|
|
database and Ollama, see `tests/test_rag_integration.py` (skipped
|
|
automatically unless `RUN_INTEGRATION_TESTS=1` is set - see that file).
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
|
|
|
|
# Make sure required-but-not-secret-critical settings have *something* set
|
|
# before app.main is imported, so settings.py's _require() checks don't
|
|
# blow up the test run when no real .env is present.
|
|
os.environ.setdefault("USE_PGVECTOR", "true")
|
|
os.environ.setdefault("DB_PASSWORD", "test-password-not-real")
|
|
os.environ.setdefault("USE_S3", "false")
|
|
os.environ.setdefault("USE_GOOGLE_CSE", "false")
|
|
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
|
|
from app.main import app
|
|
|
|
client = TestClient(app)
|
|
|
|
|
|
def test_root() -> None:
|
|
resp = client.get("/")
|
|
assert resp.status_code == 200
|
|
if "text/html" in resp.headers.get("content-type", ""):
|
|
assert "<html" in resp.text.lower()
|
|
else:
|
|
assert "service" in resp.json()
|
|
|
|
|
|
def test_health_degrades_gracefully_without_dependencies() -> None:
|
|
resp = client.get("/api/health")
|
|
assert resp.status_code == 200
|
|
body = resp.json()
|
|
assert body["status"] in {"ok", "degraded"}
|
|
assert isinstance(body["database"], bool)
|
|
assert isinstance(body["ollama"], bool)
|
|
|
|
|
|
def test_brands_returns_empty_list_without_database() -> None:
|
|
resp = client.get("/api/brands")
|
|
assert resp.status_code == 200
|
|
assert resp.json() == {"brands": []}
|
|
|
|
|
|
def test_brand_products_returns_empty_without_database() -> None:
|
|
resp = client.get("/api/brands/Parle/products")
|
|
assert resp.status_code == 200
|
|
body = resp.json()
|
|
assert body["brand"] == "Parle"
|
|
assert body["products"] == []
|
|
assert body["total"] == 0
|
|
|
|
|
|
def test_product_detail_404_when_missing() -> None:
|
|
resp = client.get("/api/brands/Parle/products/does-not-exist")
|
|
assert resp.status_code == 404
|
|
|
|
|
|
def test_chat_validates_empty_query() -> None:
|
|
resp = client.post("/api/chat", json={"query": ""})
|
|
assert resp.status_code == 422 # min_length=1 violated
|
|
|
|
|
|
def test_chat_rejects_too_many_top_k() -> None:
|
|
resp = client.post("/api/chat", json={"query": "snacks", "top_k": 999})
|
|
assert resp.status_code == 422 # le=15 violated
|
|
|
|
|
|
def test_catalog_generate_returns_job_id() -> None:
|
|
resp = client.post("/api/catalog/generate", json={"brand": "TestBrand", "max_products": 1})
|
|
assert resp.status_code == 202
|
|
body = resp.json()
|
|
assert body["brand"] == "TestBrand"
|
|
assert body["status"] in {"pending", "running", "done", "failed"}
|
|
|
|
|
|
def test_openapi_schema_lists_all_routers() -> None:
|
|
resp = client.get("/openapi.json")
|
|
assert resp.status_code == 200
|
|
paths = resp.json()["paths"]
|
|
for expected in ("/api/health", "/api/brands", "/api/search", "/api/chat", "/api/catalog/generate"):
|
|
assert expected in paths, f"missing route: {expected}"
|