109 lines
3.8 KiB
Python
109 lines
3.8 KiB
Python
"""API endpoint tests"""
|
|
|
|
import pytest
|
|
from datetime import datetime
|
|
from unittest.mock import patch, AsyncMock
|
|
from httpx import AsyncClient, ASGITransport
|
|
|
|
from app.main import app
|
|
from app.core.config import settings
|
|
from app.core.security import create_access_token
|
|
|
|
|
|
@pytest.fixture
|
|
def auth_headers():
|
|
"""Create authentication headers"""
|
|
token = create_access_token({"sub": "1", "username": "testuser"})
|
|
return {"Authorization": f"Bearer {token}"}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_health_check():
|
|
"""Test health check endpoint"""
|
|
transport = ASGITransport(app=app)
|
|
async with AsyncClient(transport=transport, base_url="http://test") as client:
|
|
response = await client.get("/health")
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["status"] == "healthy"
|
|
assert "version" in data
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_root_endpoint():
|
|
"""Test root endpoint"""
|
|
transport = ASGITransport(app=app)
|
|
async with AsyncClient(transport=transport, base_url="http://test") as client:
|
|
response = await client.get("/")
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["name"] == settings.PROJECT_NAME
|
|
assert data["version"] == settings.VERSION
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_dashboard_stats_without_auth():
|
|
"""Test dashboard stats requires authentication"""
|
|
transport = ASGITransport(app=app)
|
|
async with AsyncClient(transport=transport, base_url="http://test") as client:
|
|
response = await client.get("/api/v1/dashboard/stats")
|
|
assert response.status_code == 401
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_dashboard_stats_with_auth(auth_headers):
|
|
"""Test dashboard stats with authentication"""
|
|
with patch("app.api.v1.dashboard.cache.get", return_value=None):
|
|
with patch("app.api.v1.dashboard.cache.set", return_value=True):
|
|
with patch("app.db.session.get_db") as mock_get_db:
|
|
mock_session = AsyncMock()
|
|
mock_result = AsyncMock()
|
|
mock_result.scalar.return_value = 0
|
|
mock_result.fetchall.return_value = []
|
|
mock_session.execute.return_value = mock_result
|
|
|
|
async def mock_db_context():
|
|
yield mock_session
|
|
|
|
mock_get_db.return_value = mock_db_context()
|
|
|
|
transport = ASGITransport(app=app)
|
|
async with AsyncClient(transport=transport, base_url="http://test") as client:
|
|
response = await client.get(
|
|
"/api/v1/dashboard/stats",
|
|
headers=auth_headers,
|
|
)
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert "total_datasources" in data
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_alerts_without_auth():
|
|
"""Test alerts endpoint requires authentication"""
|
|
transport = ASGITransport(app=app)
|
|
async with AsyncClient(transport=transport, base_url="http://test") as client:
|
|
response = await client.get("/api/v1/alerts")
|
|
assert response.status_code == 401
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_alerts_endpoint_with_auth(auth_headers):
|
|
"""Test alerts endpoint with authentication"""
|
|
transport = ASGITransport(app=app)
|
|
async with AsyncClient(transport=transport, base_url="http://test") as client:
|
|
response = await client.get("/api/v1/alerts", headers=auth_headers)
|
|
assert response.status_code == 200
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_invalid_token():
|
|
"""Test that invalid token is rejected"""
|
|
transport = ASGITransport(app=app)
|
|
async with AsyncClient(transport=transport, base_url="http://test") as client:
|
|
response = await client.get(
|
|
"/api/v1/dashboard/stats",
|
|
headers={"Authorization": "Bearer invalid_token"},
|
|
)
|
|
assert response.status_code == 401
|