ghost-shopper-api / python /tests /test_gs_analytics.py
azzaraqi
Deploy FastAPI backend with Supabase pooler support
b84ea83
Raw
History Blame Contribute Delete
3.53 kB
"""
Integration Tests — GS Analytics Endpoints
NOTE: GhostShopperAnalyticsService uses PostgreSQL-specific SQL (ROUND()::numeric).
These are integration tests that require the live backend to be running.
Tests:
- GET /analytics/outlets — outlet comparison
- GET /analytics/categories — category breakdown
- GET /analytics/outlets/:id/trends — per-outlet trends
"""
import pytest
import httpx
BASE = "http://localhost:5000/api/v1"
@pytest.fixture
def client():
with httpx.Client(base_url=BASE, timeout=10) as c:
yield c
class TestGsOutletComparison:
"""Tests for GET /analytics/outlets."""
def test_comparison_returns_success(self, client):
r = client.get("/analytics/outlets")
assert r.status_code == 200
body = r.json()
assert body["status"] == "success"
def test_comparison_data_structure(self, client):
r = client.get("/analytics/outlets")
body = r.json()
data = body["data"]
assert "outlets" in data
assert isinstance(data["outlets"], list)
assert "category_labels" in data
def test_comparison_category_labels_complete(self, client):
r = client.get("/analytics/outlets")
labels = r.json()["data"]["category_labels"]
for key in ("A", "B", "C", "D", "E"):
assert key in labels
def test_outlet_entry_has_expected_fields(self, client):
r = client.get("/analytics/outlets")
data = r.json()["data"]
if data["outlets"]:
entry = data["outlets"][0]
assert "id" in entry
assert "name" in entry
assert "avg_score" in entry
assert "visit_count" in entry
class TestGsCategoryBreakdown:
"""Tests for GET /analytics/categories."""
def test_category_breakdown_returns_success(self, client):
r = client.get("/analytics/categories")
assert r.status_code == 200
body = r.json()
assert body["status"] == "success"
def test_category_breakdown_data_structure(self, client):
r = client.get("/analytics/categories")
body = r.json()
data = body["data"]
assert "categories" in data
assert isinstance(data["categories"], list)
def test_category_entry_has_expected_fields(self, client):
r = client.get("/analytics/categories")
data = r.json()["data"]
if data["categories"]:
entry = data["categories"][0]
assert "category" in entry
assert "label" in entry
assert "avg_score" in entry
assert "total_items" in entry
class TestGsOutletTrends:
"""Tests for GET /analytics/outlets/:id/trends."""
def test_invalid_outlet_returns_404(self, client):
import uuid
r = client.get(f"/analytics/outlets/{uuid.uuid4()}/trends")
assert r.status_code == 404
def test_valid_outlet_returns_data(self, client):
"""Find a real outlet, then query trends for it."""
r = client.get("/outlets")
if r.status_code != 200:
pytest.skip("No outlets endpoint available")
outlets = r.json().get("data", [])
if not outlets:
pytest.skip("No outlets in database")
outlet_id = outlets[0]["id"]
r = client.get(f"/analytics/outlets/{outlet_id}/trends")
assert r.status_code == 200
body = r.json()
assert body["status"] == "success"
assert "trends" in body["data"]
assert "outlet_name" in body["data"]