Spaces:
Sleeping
Sleeping
File size: 5,607 Bytes
c509b44 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 |
"""
Integration tests for new API endpoints
"""
import sys
from pathlib import Path
# Add backend to path
backend_dir = Path(__file__).parent.parent
sys.path.insert(0, str(backend_dir))
# Add root directory to path for backend.api imports
root_dir = Path(__file__).resolve().parents[2]
sys.path.insert(0, str(root_dir))
import pytest
from fastapi.testclient import TestClient
from fastapi import FastAPI
try:
from backend.api.main import app
except ImportError:
# Fallback if backend.api.main doesn't work
from api.main import app
@pytest.fixture
def client():
"""Create a test client."""
return TestClient(app)
def test_analytics_overview_endpoint(client):
"""Test /analytics/overview endpoint."""
response = client.get(
"/analytics/overview",
headers={"x-tenant-id": "test_tenant"},
params={"days": 30}
)
assert response.status_code == 200
data = response.json()
assert "tenant_id" in data
assert "overview" in data
assert "total_queries" in data["overview"]
assert "tool_usage" in data["overview"]
assert "redflag_count" in data["overview"]
def test_analytics_tool_usage_endpoint(client):
"""Test /analytics/tool-usage endpoint."""
response = client.get(
"/analytics/tool-usage",
headers={"x-tenant-id": "test_tenant"},
params={"days": 30}
)
assert response.status_code == 200
data = response.json()
assert "tenant_id" in data
assert "tool_usage" in data
assert "period_days" in data
def test_analytics_rag_quality_endpoint(client):
"""Test /analytics/rag-quality endpoint."""
response = client.get(
"/analytics/rag-quality",
headers={"x-tenant-id": "test_tenant"},
params={"days": 30}
)
assert response.status_code == 200
data = response.json()
assert "tenant_id" in data
assert "rag_quality" in data
def test_admin_rules_with_regex(client):
"""Test adding admin rule with regex pattern and severity."""
response = client.post(
"/admin/rules",
headers={"x-tenant-id": "test_tenant"},
json={
"rule": "Block password queries",
"pattern": ".*password.*",
"severity": "high",
"description": "Blocks password-related queries"
}
)
assert response.status_code == 200
data = response.json()
assert data["severity"] == "high"
assert ".*password.*" in data["pattern"]
# Get detailed rules
response = client.get(
"/admin/rules",
headers={"x-tenant-id": "test_tenant"},
params={"detailed": True}
)
assert response.status_code == 200
data = response.json()
assert "rules" in data
assert len(data["rules"]) > 0
assert data["rules"][0]["severity"] == "high"
def test_admin_violations_endpoint(client):
"""Test /admin/violations endpoint."""
response = client.get(
"/admin/violations",
headers={"x-tenant-id": "test_tenant"},
params={"limit": 50, "days": 30}
)
assert response.status_code == 200
data = response.json()
assert "tenant_id" in data
assert "violations" in data
assert "count" in data
def test_admin_tools_logs_endpoint(client):
"""Test /admin/tools/logs endpoint."""
response = client.get(
"/admin/tools/logs",
headers={"x-tenant-id": "test_tenant"},
params={"tool_name": "rag", "days": 7}
)
assert response.status_code == 200
data = response.json()
assert "tenant_id" in data
assert "tool_usage" in data
def test_agent_debug_endpoint(client):
"""Test /agent/debug endpoint."""
# Note: This will fail if LLM/MCP servers are not running
# But we can at least test the endpoint structure
response = client.post(
"/agent/debug",
json={
"tenant_id": "test_tenant",
"message": "Test message",
"temperature": 0.0
}
)
# Might fail if services not available, but should have proper error handling
assert response.status_code in [200, 500, 503] # Accept various status codes
def test_agent_plan_endpoint(client):
"""Test /agent/plan endpoint."""
# Note: This will fail if LLM/MCP servers are not running
response = client.post(
"/agent/plan",
json={
"tenant_id": "test_tenant",
"message": "What is the company policy?",
"temperature": 0.0
}
)
# Might fail if services not available
assert response.status_code in [200, 500, 503]
def test_missing_tenant_id_returns_400(client):
"""Test that endpoints return 400 when tenant ID is missing."""
endpoints = [
"/analytics/overview",
"/analytics/tool-usage",
"/admin/rules",
"/admin/violations"
]
for endpoint in endpoints:
response = client.get(endpoint)
assert response.status_code == 400, f"Endpoint {endpoint} should return 400"
def test_admin_tenants_endpoints(client):
"""Test tenant management endpoints (placeholders)."""
# List tenants
response = client.get("/admin/tenants")
assert response.status_code == 200
data = response.json()
assert "tenants" in data
# Create tenant (placeholder)
response = client.post("/admin/tenants", params={"tenant_id": "new_tenant"})
assert response.status_code == 200
# Delete tenant (placeholder)
response = client.delete("/admin/tenants/new_tenant")
assert response.status_code == 200
|