mirror of
http://88.130.71.182:3000/BlitTech/contexta_be.git
synced 2026-06-13 08:30:07 +00:00
92d4c2fc5e
- Add new routers: admin, appointments, campaigns - Add storage service and logging config - Add migrations directory and test suite with pytest config - Add supabase_migration_features.sql - Update models, dependencies, config, and existing routers - Remove whatsapp_service (deleted) - Update pyproject.toml and uv.lock dependencies Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
95 lines
3.6 KiB
Python
95 lines
3.6 KiB
Python
"""Tests for admin endpoints — access control and basic structure."""
|
|
import pytest
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
|
|
class TestAdminAccessControl:
|
|
"""Admin endpoints must return 401 without auth and 403 for non-admin users."""
|
|
|
|
ADMIN_ENDPOINTS = [
|
|
("GET", "/api/v1/admin/stats"),
|
|
("GET", "/api/v1/admin/users"),
|
|
("GET", "/api/v1/admin/chatbots"),
|
|
("GET", "/api/v1/admin/conversations"),
|
|
("GET", "/api/v1/admin/system/health"),
|
|
]
|
|
|
|
def test_admin_endpoints_require_auth(self, client):
|
|
for method, path in self.ADMIN_ENDPOINTS:
|
|
resp = client.request(method, path)
|
|
assert resp.status_code == 401, f"{method} {path} should require auth, got {resp.status_code}"
|
|
|
|
def test_non_admin_user_gets_403(self, client):
|
|
"""Authenticated user without is_admin flag should get 403."""
|
|
user = MagicMock()
|
|
user.id = "normal-user-id"
|
|
user.email = "user@example.com"
|
|
|
|
with patch("app.dependencies.get_supabase") as mock_sb:
|
|
sb = MagicMock()
|
|
# get_current_user: no suspension
|
|
sb.table.return_value.select.return_value.eq.return_value.execute.return_value = MagicMock(
|
|
data=[{"suspended_at": None, "is_admin": False}]
|
|
)
|
|
mock_sb.return_value = sb
|
|
|
|
with patch("app.dependencies.security") as mock_sec:
|
|
creds = MagicMock()
|
|
creds.credentials = "valid-token"
|
|
mock_sec.return_value = creds
|
|
|
|
with patch("app.database.get_supabase") as mock_db_sb:
|
|
db_sb = MagicMock()
|
|
db_sb.auth.get_user.return_value = MagicMock(user=user)
|
|
# Profile: not admin, not suspended
|
|
db_sb.table.return_value.select.return_value.eq.return_value.execute.return_value = MagicMock(
|
|
data=[{"is_admin": False, "suspended_at": None}]
|
|
)
|
|
mock_db_sb.return_value = db_sb
|
|
|
|
resp = client.get(
|
|
"/api/v1/admin/stats",
|
|
headers={"Authorization": "Bearer valid-token"},
|
|
)
|
|
|
|
assert resp.status_code in (401, 403)
|
|
|
|
|
|
class TestAdminModels:
|
|
def test_admin_stats_response_shape(self):
|
|
from app.models import AdminStatsResponse
|
|
stats = AdminStatsResponse(
|
|
total_users=10,
|
|
total_chatbots=5,
|
|
total_published_chatbots=3,
|
|
total_conversations=100,
|
|
total_messages=500,
|
|
active_subscriptions={"free": 8, "starter": 2},
|
|
)
|
|
assert stats.total_users == 10
|
|
assert stats.active_subscriptions["free"] == 8
|
|
|
|
def test_admin_user_list_item_defaults(self):
|
|
from app.models import AdminUserListItem
|
|
item = AdminUserListItem(id="id", email="test@example.com")
|
|
assert item.plan == "free"
|
|
assert item.is_admin == False
|
|
assert item.is_suspended == False
|
|
|
|
def test_admin_system_health_shape(self):
|
|
from app.models import AdminSystemHealth
|
|
from datetime import datetime
|
|
health = AdminSystemHealth(
|
|
db="healthy",
|
|
qdrant="healthy",
|
|
llm_providers={"openai": True, "anthropic": False},
|
|
timestamp=datetime.utcnow(),
|
|
)
|
|
assert health.db == "healthy"
|
|
assert health.llm_providers["openai"] == True
|
|
|
|
def test_change_plan_request_validates(self):
|
|
from app.models import AdminChangePlanRequest
|
|
req = AdminChangePlanRequest(plan="business", reason="Testing")
|
|
assert req.plan == "business"
|