"""Tests for REST API endpoints."""
import pytest
from unittest.mock import Mock, AsyncMock, patch, MagicMock
from datetime import datetime, timezone, timedelta
from typing import Dict, List, Any
import json
from fastapi.testclient import TestClient
from fastapi import status
from src.server.rest_api import (
app,
NewsAPI,
AnalysisAPI,
MonitoringAPI,
AdminAPI,
APIError,
NewsRequest,
AnalysisRequest,
MonitoringRequest,
NewsResponse,
AnalysisResponse,
MonitoringResponse
)
class TestNewsAPI:
"""Test cases for News API endpoints."""
@pytest.fixture
def client(self):
"""Create test client for API testing."""
return TestClient(app)
@pytest.fixture
def sample_news_data(self):
"""Sample news data for testing."""
return [
{
"id": "news_1",
"title": "삼성전자 3분기 실적 발표",
"content": "삼성전자가 좋은 실적을 발표했습니다.",
"url": "https://example.com/news/1",
"source": "naver",
"published_at": "2024-01-01T10:00:00Z",
"sentiment": {"score": 0.8, "label": "positive"},
"market_impact": {"score": 0.7, "prediction": "bullish"}
},
{
"id": "news_2",
"title": "삼성전자 주가 상승",
"content": "삼성전자 주가가 크게 상승했습니다.",
"url": "https://example.com/news/2",
"source": "daum",
"published_at": "2024-01-01T11:00:00Z",
"sentiment": {"score": 0.9, "label": "positive"},
"market_impact": {"score": 0.8, "prediction": "bullish"}
}
]
def test_get_news_endpoint(self, client, sample_news_data):
"""Test GET /api/v1/news endpoint."""
with patch('src.server.rest_api.get_mcp_server') as mock_get_server:
# Mock MCP server and response
mock_server = AsyncMock()
mock_response = Mock()
mock_response.success = True
mock_response.data = {
"news": sample_news_data,
"metadata": {"total_count": 2}
}
mock_server.handle_request.return_value = mock_response
mock_get_server.return_value = mock_server
response = client.get("/api/v1/news?query=삼성전자&limit=10")
assert response.status_code == status.HTTP_200_OK
data = response.json()
assert "news" in data
assert "total_count" in data
assert len(data["news"]) == 2
def test_get_news_with_filters(self, client):
"""Test GET /api/v1/news with various filters."""
with patch('src.server.rest_api.NewsAPI.get_news') as mock_get_news:
mock_get_news.return_value = {
"news": [],
"total_count": 0,
"page": 1,
"page_size": 10
}
response = client.get(
"/api/v1/news?"
"query=삼성전자&"
"sources=naver,daum&"
"sentiment=positive&"
"date_from=2024-01-01&"
"date_to=2024-01-31&"
"limit=20&"
"page=1"
)
assert response.status_code == status.HTTP_200_OK
data = response.json()
assert "news" in data
def test_get_news_by_id(self, client, sample_news_data):
"""Test GET /api/v1/news/{news_id} endpoint."""
with patch('src.server.rest_api.NewsAPI.get_news_by_id') as mock_get_news:
mock_get_news.return_value = sample_news_data[0]
response = client.get("/api/v1/news/news_1")
assert response.status_code == status.HTTP_200_OK
data = response.json()
assert data["id"] == "news_1"
assert data["title"] == "삼성전자 3분기 실적 발표"
def test_get_news_by_id_not_found(self, client):
"""Test GET /api/v1/news/{news_id} with non-existent ID."""
with patch('src.server.rest_api.NewsAPI.get_news_by_id') as mock_get_news:
mock_get_news.side_effect = APIError("News not found", status_code=404)
response = client.get("/api/v1/news/non_existent")
assert response.status_code == status.HTTP_404_NOT_FOUND
def test_search_news_endpoint(self, client):
"""Test POST /api/v1/news/search endpoint."""
search_request = {
"keywords": ["삼성전자", "실적"],
"filters": {
"sentiment": "positive",
"sources": ["naver", "daum"],
"date_range": "7d"
},
"sort_by": "published_at",
"sort_order": "desc",
"limit": 20,
"page": 1
}
with patch('src.server.rest_api.NewsAPI.search_news') as mock_search:
mock_search.return_value = {
"news": [],
"search_metadata": {
"total_count": 0,
"search_time": 0.15,
"filters_applied": search_request["filters"]
}
}
response = client.post("/api/v1/news/search", json=search_request)
assert response.status_code == status.HTTP_200_OK
data = response.json()
assert "news" in data
assert "search_metadata" in data
def test_get_news_timeline(self, client):
"""Test GET /api/v1/news/timeline endpoint."""
with patch('src.server.rest_api.NewsAPI.get_timeline') as mock_timeline:
mock_timeline.return_value = {
"timeline": [
{
"timestamp": "2024-01-01T10:00:00Z",
"count": 5,
"avg_sentiment": 0.6,
"topics": ["삼성전자", "실적"]
},
{
"timestamp": "2024-01-01T11:00:00Z",
"count": 8,
"avg_sentiment": 0.7,
"topics": ["삼성전자", "주가"]
}
],
"metadata": {
"entity": "삼성전자",
"date_range": "24h",
"resolution": "1h"
}
}
response = client.get(
"/api/v1/news/timeline?"
"entity=삼성전자&"
"date_range=24h&"
"resolution=1h"
)
assert response.status_code == status.HTTP_200_OK
data = response.json()
assert "timeline" in data
assert len(data["timeline"]) == 2
def test_get_trending_topics(self, client):
"""Test GET /api/v1/news/trending endpoint."""
with patch('src.server.rest_api.NewsAPI.get_trending_topics') as mock_trending:
mock_trending.return_value = {
"trending_topics": [
{"topic": "삼성전자", "score": 0.95, "count": 50, "trend": "rising"},
{"topic": "실적발표", "score": 0.87, "count": 30, "trend": "stable"},
{"topic": "주가상승", "score": 0.82, "count": 25, "trend": "rising"}
],
"metadata": {
"time_window": "24h",
"updated_at": "2024-01-01T12:00:00Z"
}
}
response = client.get("/api/v1/news/trending?time_window=24h")
assert response.status_code == status.HTTP_200_OK
data = response.json()
assert "trending_topics" in data
assert len(data["trending_topics"]) == 3
def test_news_aggregation(self, client):
"""Test GET /api/v1/news/aggregation endpoint."""
with patch('src.server.rest_api.NewsAPI.get_aggregation') as mock_agg:
mock_agg.return_value = {
"aggregations": {
"by_source": {
"naver": {"count": 120, "avg_sentiment": 0.65},
"daum": {"count": 95, "avg_sentiment": 0.72}
},
"by_sentiment": {
"positive": {"count": 150, "percentage": 70.0},
"negative": {"count": 45, "percentage": 21.0},
"neutral": {"count": 20, "percentage": 9.0}
}
},
"metadata": {
"total_news": 215,
"date_range": "7d",
"group_by": ["source", "sentiment"]
}
}
response = client.get(
"/api/v1/news/aggregation?"
"group_by=source,sentiment&"
"date_range=7d"
)
assert response.status_code == status.HTTP_200_OK
data = response.json()
assert "aggregations" in data
assert "by_source" in data["aggregations"]
def test_export_news(self, client):
"""Test GET /api/v1/news/export endpoint."""
with patch('src.server.rest_api.NewsAPI.export_news') as mock_export:
mock_export.return_value = {
"export_url": "https://api.example.com/exports/news_20240101.json",
"format": "json",
"file_size": "2.5MB",
"expires_at": "2024-01-02T12:00:00Z"
}
response = client.get(
"/api/v1/news/export?"
"format=json&"
"query=삼성전자&"
"date_range=7d"
)
assert response.status_code == status.HTTP_200_OK
data = response.json()
assert "export_url" in data
assert data["format"] == "json"
def test_news_pagination(self, client):
"""Test news pagination functionality."""
with patch('src.server.rest_api.NewsAPI.get_news') as mock_get_news:
# First page
mock_get_news.return_value = {
"news": [{"id": f"news_{i}"} for i in range(10)],
"total_count": 25,
"page": 1,
"page_size": 10,
"total_pages": 3
}
response = client.get("/api/v1/news?page=1&limit=10")
assert response.status_code == status.HTTP_200_OK
data = response.json()
assert data["page"] == 1
assert data["total_pages"] == 3
assert len(data["news"]) == 10
def test_news_validation_errors(self, client):
"""Test validation errors in news endpoints."""
# Invalid date format
response = client.get("/api/v1/news?date_from=invalid-date")
assert response.status_code == status.HTTP_422_UNPROCESSABLE_ENTITY
# Invalid limit (too large)
response = client.get("/api/v1/news?limit=10000")
assert response.status_code == status.HTTP_422_UNPROCESSABLE_ENTITY
# Invalid page number
response = client.get("/api/v1/news?page=0")
assert response.status_code == status.HTTP_422_UNPROCESSABLE_ENTITY
class TestAnalysisAPI:
"""Test cases for Analysis API endpoints."""
@pytest.fixture
def client(self):
"""Create test client for API testing."""
return TestClient(app)
def test_analyze_sentiment_endpoint(self, client):
"""Test POST /api/v1/analysis/sentiment endpoint."""
request_data = {
"news_ids": ["news_1", "news_2"],
"options": {
"include_confidence": True,
"language": "korean",
"domain": "financial"
}
}
with patch('src.server.rest_api.AnalysisAPI.analyze_sentiment') as mock_analyze:
mock_analyze.return_value = {
"sentiment_results": [
{
"news_id": "news_1",
"sentiment": "positive",
"score": 0.85,
"confidence": 0.92,
"keywords": ["호조", "상승", "실적"]
},
{
"news_id": "news_2",
"sentiment": "positive",
"score": 0.78,
"confidence": 0.89,
"keywords": ["상승", "증가", "주가"]
}
],
"summary": {
"avg_sentiment": 0.815,
"sentiment_distribution": {"positive": 2, "negative": 0, "neutral": 0}
}
}
response = client.post("/api/v1/analysis/sentiment", json=request_data)
assert response.status_code == status.HTTP_200_OK
data = response.json()
assert "sentiment_results" in data
assert "summary" in data
assert len(data["sentiment_results"]) == 2
def test_analyze_market_impact_endpoint(self, client):
"""Test POST /api/v1/analysis/market-impact endpoint."""
request_data = {
"news_ids": ["news_1"],
"market_context": {
"current_price": 70000,
"volume": 1000000,
"market_cap": "400T",
"sector": "technology"
},
"prediction_horizon": "1d"
}
with patch('src.server.rest_api.AnalysisAPI.analyze_market_impact') as mock_analyze:
mock_analyze.return_value = {
"impact_results": [
{
"news_id": "news_1",
"impact_score": 0.75,
"price_prediction": "bullish",
"confidence_interval": [0.65, 0.85],
"predicted_change": 0.05,
"risk_factors": ["market_volatility", "sector_rotation"]
}
],
"portfolio_impact": {
"expected_return": 0.035,
"risk_score": 0.24,
"recommended_action": "buy"
}
}
response = client.post("/api/v1/analysis/market-impact", json=request_data)
assert response.status_code == status.HTTP_200_OK
data = response.json()
assert "impact_results" in data
assert "portfolio_impact" in data
def test_detect_rumors_endpoint(self, client):
"""Test POST /api/v1/analysis/rumors endpoint."""
request_data = {
"news_ids": ["news_1", "news_2"],
"detection_params": {
"sensitivity": "high",
"check_sources": True,
"cross_reference": True
}
}
with patch('src.server.rest_api.AnalysisAPI.detect_rumors') as mock_detect:
mock_detect.return_value = {
"rumor_results": [
{
"news_id": "news_1",
"rumor_score": 0.15,
"rumor_type": "none",
"confidence": 0.95,
"evidence": [],
"source_reliability": 0.88
},
{
"news_id": "news_2",
"rumor_score": 0.72,
"rumor_type": "speculation",
"confidence": 0.84,
"evidence": ["unverified_source", "vague_language"],
"source_reliability": 0.45
}
],
"summary": {
"high_risk_count": 1,
"avg_rumor_score": 0.435,
"recommendations": ["verify_sources", "cross_check"]
}
}
response = client.post("/api/v1/analysis/rumors", json=request_data)
assert response.status_code == status.HTTP_200_OK
data = response.json()
assert "rumor_results" in data
assert "summary" in data
def test_summarize_news_endpoint(self, client):
"""Test POST /api/v1/analysis/summarize endpoint."""
request_data = {
"news_ids": ["news_1", "news_2"],
"summary_type": "extractive",
"length": "short",
"focus_topics": ["실적", "주가"],
"language": "korean"
}
with patch('src.server.rest_api.AnalysisAPI.summarize_news') as mock_summarize:
mock_summarize.return_value = {
"summary_results": [
{
"news_id": "news_1",
"summary": "삼성전자가 3분기 좋은 실적을 발표했습니다.",
"key_points": ["실적 발표", "좋은 성과"],
"confidence": 0.91
}
],
"combined_summary": {
"overall_summary": "삼성전자 실적 호조로 주가 상승세가 지속되고 있습니다.",
"main_themes": ["실적 호조", "주가 상승"],
"sentiment_overview": "positive"
}
}
response = client.post("/api/v1/analysis/summarize", json=request_data)
assert response.status_code == status.HTTP_200_OK
data = response.json()
assert "summary_results" in data
assert "combined_summary" in data
def test_batch_analysis_endpoint(self, client):
"""Test POST /api/v1/analysis/batch endpoint."""
request_data = {
"analyses": [
{
"type": "sentiment",
"news_ids": ["news_1", "news_2"],
"options": {"include_confidence": True}
},
{
"type": "market_impact",
"news_ids": ["news_1"],
"options": {"prediction_horizon": "1d"}
},
{
"type": "rumors",
"news_ids": ["news_2"],
"options": {"sensitivity": "medium"}
}
],
"output_format": "detailed"
}
with patch('src.server.rest_api.AnalysisAPI.run_batch_analysis') as mock_batch:
mock_batch.return_value = {
"batch_results": [
{"analysis_type": "sentiment", "status": "completed", "results": {}},
{"analysis_type": "market_impact", "status": "completed", "results": {}},
{"analysis_type": "rumors", "status": "completed", "results": {}}
],
"summary": {
"total_analyses": 3,
"completed": 3,
"failed": 0,
"processing_time": 2.45
}
}
response = client.post("/api/v1/analysis/batch", json=request_data)
assert response.status_code == status.HTTP_200_OK
data = response.json()
assert "batch_results" in data
assert "summary" in data
assert data["summary"]["completed"] == 3
def test_analysis_history_endpoint(self, client):
"""Test GET /api/v1/analysis/history endpoint."""
with patch('src.server.rest_api.AnalysisAPI.get_analysis_history') as mock_history:
mock_history.return_value = {
"history": [
{
"id": "analysis_1",
"type": "sentiment",
"created_at": "2024-01-01T10:00:00Z",
"news_count": 5,
"status": "completed",
"results_summary": {"avg_sentiment": 0.7}
},
{
"id": "analysis_2",
"type": "market_impact",
"created_at": "2024-01-01T09:30:00Z",
"news_count": 3,
"status": "completed",
"results_summary": {"avg_impact": 0.6}
}
],
"pagination": {
"total_count": 2,
"page": 1,
"page_size": 10
}
}
response = client.get("/api/v1/analysis/history?page=1&limit=10")
assert response.status_code == status.HTTP_200_OK
data = response.json()
assert "history" in data
assert "pagination" in data
def test_analysis_comparison_endpoint(self, client):
"""Test POST /api/v1/analysis/compare endpoint."""
request_data = {
"baseline_period": {
"start": "2024-01-01T00:00:00Z",
"end": "2024-01-07T23:59:59Z"
},
"comparison_period": {
"start": "2024-01-08T00:00:00Z",
"end": "2024-01-14T23:59:59Z"
},
"metrics": ["sentiment", "market_impact", "volume"],
"entities": ["삼성전자", "LG전자"]
}
with patch('src.server.rest_api.AnalysisAPI.compare_periods') as mock_compare:
mock_compare.return_value = {
"comparison_results": {
"sentiment_change": 0.15,
"impact_change": -0.05,
"volume_change": 0.25
},
"statistical_significance": {
"sentiment": {"p_value": 0.023, "significant": True},
"impact": {"p_value": 0.156, "significant": False}
},
"insights": [
"Sentiment significantly improved in the comparison period",
"Market impact remained relatively stable"
]
}
response = client.post("/api/v1/analysis/compare", json=request_data)
assert response.status_code == status.HTTP_200_OK
data = response.json()
assert "comparison_results" in data
assert "statistical_significance" in data
def test_analysis_validation_errors(self, client):
"""Test validation errors in analysis endpoints."""
# Empty news_ids
response = client.post("/api/v1/analysis/sentiment", json={"news_ids": []})
assert response.status_code == status.HTTP_422_UNPROCESSABLE_ENTITY
# Invalid analysis type in batch
response = client.post("/api/v1/analysis/batch", json={
"analyses": [{"type": "invalid_type", "news_ids": ["news_1"]}]
})
assert response.status_code == status.HTTP_422_UNPROCESSABLE_ENTITY
class TestMonitoringAPI:
"""Test cases for Monitoring API endpoints."""
@pytest.fixture
def client(self):
"""Create test client for API testing."""
return TestClient(app)
def test_get_status_endpoint(self, client):
"""Test GET /api/v1/monitoring/status endpoint."""
with patch('src.server.rest_api.MonitoringAPI.get_status') as mock_status:
mock_status.return_value = {
"system_status": "healthy",
"services": {
"news_collector": {"status": "running", "uptime": "24h 15m"},
"analysis_engine": {"status": "running", "uptime": "24h 10m"},
"database": {"status": "connected", "connections": 15},
"cache": {"status": "running", "hit_rate": 0.95}
},
"performance": {
"cpu_usage": 45.2,
"memory_usage": 67.8,
"disk_usage": 34.5,
"response_time": 120
},
"last_updated": "2024-01-01T12:00:00Z"
}
response = client.get("/api/v1/monitoring/status")
assert response.status_code == status.HTTP_200_OK
data = response.json()
assert "system_status" in data
assert "services" in data
assert "performance" in data
def test_get_metrics_endpoint(self, client):
"""Test GET /api/v1/monitoring/metrics endpoint."""
with patch('src.server.rest_api.MonitoringAPI.get_metrics') as mock_metrics:
mock_metrics.return_value = {
"metrics": {
"news_processing": {
"articles_processed": 1500,
"processing_rate": 125.5,
"error_rate": 0.02,
"avg_processing_time": 0.85
},
"api_performance": {
"requests_per_minute": 245,
"avg_response_time": 180,
"success_rate": 0.998,
"error_breakdown": {"4xx": 5, "5xx": 1}
},
"analysis_performance": {
"sentiment_analyses": 850,
"market_impact_analyses": 320,
"rumor_detections": 45,
"summarizations": 120
}
},
"time_range": {
"start": "2024-01-01T00:00:00Z",
"end": "2024-01-01T12:00:00Z",
"duration": "12h"
}
}
response = client.get("/api/v1/monitoring/metrics?time_range=12h")
assert response.status_code == status.HTTP_200_OK
data = response.json()
assert "metrics" in data
assert "time_range" in data
def test_configure_alerts_endpoint(self, client):
"""Test POST /api/v1/monitoring/alerts/configure endpoint."""
alert_config = {
"alert_name": "high_error_rate",
"conditions": {
"metric": "error_rate",
"threshold": 0.05,
"operator": "greater_than",
"duration": "5m"
},
"actions": [
{
"type": "email",
"recipients": ["admin@example.com"],
"template": "error_alert"
},
{
"type": "webhook",
"url": "https://hooks.slack.com/...",
"method": "POST"
}
],
"enabled": True,
"priority": "high"
}
with patch('src.server.rest_api.MonitoringAPI.configure_alert') as mock_configure:
mock_configure.return_value = {
"alert_id": "alert_123",
"status": "active",
"created_at": "2024-01-01T12:00:00Z",
"next_check": "2024-01-01T12:05:00Z"
}
response = client.post("/api/v1/monitoring/alerts/configure", json=alert_config)
assert response.status_code == status.HTTP_201_CREATED
data = response.json()
assert "alert_id" in data
assert data["status"] == "active"
def test_get_alerts_history_endpoint(self, client):
"""Test GET /api/v1/monitoring/alerts/history endpoint."""
with patch('src.server.rest_api.MonitoringAPI.get_alerts_history') as mock_history:
mock_history.return_value = {
"alerts_history": [
{
"id": "alert_event_1",
"alert_name": "high_error_rate",
"triggered_at": "2024-01-01T10:30:00Z",
"resolved_at": "2024-01-01T10:45:00Z",
"severity": "high",
"message": "Error rate exceeded 5% threshold",
"actions_taken": ["email_sent", "webhook_called"]
},
{
"id": "alert_event_2",
"alert_name": "low_cache_hit_rate",
"triggered_at": "2024-01-01T09:15:00Z",
"resolved_at": "2024-01-01T09:30:00Z",
"severity": "medium",
"message": "Cache hit rate below 90%",
"actions_taken": ["email_sent"]
}
],
"pagination": {
"total_count": 2,
"page": 1,
"page_size": 10
}
}
response = client.get("/api/v1/monitoring/alerts/history?page=1&limit=10")
assert response.status_code == status.HTTP_200_OK
data = response.json()
assert "alerts_history" in data
assert "pagination" in data
def test_get_health_check_endpoint(self, client):
"""Test GET /api/v1/monitoring/health endpoint."""
with patch('src.server.rest_api.MonitoringAPI.health_check') as mock_health:
mock_health.return_value = {
"status": "healthy",
"checks": {
"database": {"status": "healthy", "response_time": 15},
"cache": {"status": "healthy", "response_time": 5},
"external_apis": {"status": "degraded", "response_time": 2500},
"disk_space": {"status": "healthy", "available": "85%"}
},
"uptime": "5d 12h 30m",
"version": "1.0.0",
"timestamp": "2024-01-01T12:00:00Z"
}
response = client.get("/api/v1/monitoring/health")
assert response.status_code == status.HTTP_200_OK
data = response.json()
assert data["status"] == "healthy"
assert "checks" in data
assert "uptime" in data
def test_get_performance_dashboard_endpoint(self, client):
"""Test GET /api/v1/monitoring/dashboard endpoint."""
with patch('src.server.rest_api.MonitoringAPI.get_dashboard_data') as mock_dashboard:
mock_dashboard.return_value = {
"dashboard_data": {
"overview": {
"total_news_processed": 25000,
"active_users": 150,
"api_requests_today": 15000,
"system_uptime": "99.95%"
},
"charts": {
"news_volume": [
{"time": "2024-01-01T00:00:00Z", "value": 120},
{"time": "2024-01-01T01:00:00Z", "value": 135}
],
"sentiment_distribution": {
"positive": 65, "negative": 15, "neutral": 20
},
"response_times": [
{"endpoint": "/api/v1/news", "avg_time": 150},
{"endpoint": "/api/v1/analysis/sentiment", "avg_time": 220}
]
}
},
"last_updated": "2024-01-01T12:00:00Z"
}
response = client.get("/api/v1/monitoring/dashboard")
assert response.status_code == status.HTTP_200_OK
data = response.json()
assert "dashboard_data" in data
assert "overview" in data["dashboard_data"]
def test_real_time_metrics_endpoint(self, client):
"""Test GET /api/v1/monitoring/realtime endpoint."""
with patch('src.server.rest_api.MonitoringAPI.get_realtime_metrics') as mock_realtime:
mock_realtime.return_value = {
"realtime_metrics": {
"current_rps": 45.2,
"active_connections": 25,
"queue_size": 12,
"cpu_usage": 48.5,
"memory_usage": 62.3,
"error_rate": 0.015,
"avg_response_time": 180
},
"timestamp": "2024-01-01T12:00:00Z",
"update_interval": 5
}
response = client.get("/api/v1/monitoring/realtime")
assert response.status_code == status.HTTP_200_OK
data = response.json()
assert "realtime_metrics" in data
assert "timestamp" in data
class TestAdminAPI:
"""Test cases for Admin API endpoints."""
@pytest.fixture
def client(self):
"""Create test client for API testing."""
return TestClient(app)
def test_admin_authentication_required(self, client):
"""Test that admin endpoints require authentication."""
# Without auth header
response = client.get("/api/v1/admin/system/info")
assert response.status_code == status.HTTP_401_UNAUTHORIZED
def test_get_system_info_endpoint(self, client):
"""Test GET /api/v1/admin/system/info endpoint."""
headers = {"Authorization": "Bearer admin_token"}
with patch('src.server.rest_api.AdminAPI.get_system_info') as mock_info:
mock_info.return_value = {
"system_info": {
"version": "1.0.0",
"build_date": "2024-01-01T00:00:00Z",
"python_version": "3.11.0",
"environment": "production",
"database_version": "PostgreSQL 15.0",
"redis_version": "7.0.0"
},
"statistics": {
"total_news_articles": 125000,
"total_analyses": 45000,
"total_api_calls": 2500000,
"uptime_days": 45
}
}
response = client.get("/api/v1/admin/system/info", headers=headers)
assert response.status_code == status.HTTP_200_OK
data = response.json()
assert "system_info" in data
assert "statistics" in data
def test_manage_collectors_endpoint(self, client):
"""Test POST /api/v1/admin/collectors/manage endpoint."""
headers = {"Authorization": "Bearer admin_token"}
manage_request = {
"action": "restart",
"collector": "naver",
"config": {
"interval": 300,
"max_articles": 100
}
}
with patch('src.server.rest_api.AdminAPI.manage_collectors') as mock_manage:
mock_manage.return_value = {
"status": "success",
"message": "Naver collector restarted successfully",
"collector_status": {
"name": "naver",
"status": "running",
"last_collection": "2024-01-01T11:55:00Z",
"articles_collected": 1250
}
}
response = client.post(
"/api/v1/admin/collectors/manage",
json=manage_request,
headers=headers
)
assert response.status_code == status.HTTP_200_OK
data = response.json()
assert data["status"] == "success"
assert "collector_status" in data
def test_database_operations_endpoint(self, client):
"""Test POST /api/v1/admin/database/operations endpoint."""
headers = {"Authorization": "Bearer admin_token"}
db_request = {
"operation": "cleanup",
"parameters": {
"older_than_days": 30,
"dry_run": True
}
}
with patch('src.server.rest_api.AdminAPI.database_operations') as mock_db:
mock_db.return_value = {
"operation": "cleanup",
"status": "completed",
"affected_records": 5000,
"execution_time": 2.45,
"dry_run": True,
"summary": "Would delete 5000 records older than 30 days"
}
response = client.post(
"/api/v1/admin/database/operations",
json=db_request,
headers=headers
)
assert response.status_code == status.HTTP_200_OK
data = response.json()
assert data["operation"] == "cleanup"
assert data["dry_run"] is True
def test_cache_management_endpoint(self, client):
"""Test POST /api/v1/admin/cache/manage endpoint."""
headers = {"Authorization": "Bearer admin_token"}
cache_request = {
"action": "flush",
"cache_type": "analysis_results",
"keys": ["sentiment:*", "market_impact:*"]
}
with patch('src.server.rest_api.AdminAPI.manage_cache') as mock_cache:
mock_cache.return_value = {
"action": "flush",
"cache_type": "analysis_results",
"keys_affected": 1250,
"status": "completed",
"execution_time": 0.85
}
response = client.post(
"/api/v1/admin/cache/manage",
json=cache_request,
headers=headers
)
assert response.status_code == status.HTTP_200_OK
data = response.json()
assert data["action"] == "flush"
assert data["keys_affected"] == 1250
def test_export_data_endpoint(self, client):
"""Test POST /api/v1/admin/export endpoint."""
headers = {"Authorization": "Bearer admin_token"}
export_request = {
"data_type": "news_articles",
"format": "json",
"filters": {
"date_from": "2024-01-01",
"date_to": "2024-01-31",
"sources": ["naver", "daum"]
},
"include_analysis": True
}
with patch('src.server.rest_api.AdminAPI.export_data') as mock_export:
mock_export.return_value = {
"export_id": "export_123",
"status": "initiated",
"estimated_records": 15000,
"estimated_size": "45MB",
"download_url": None,
"expires_at": "2024-01-02T12:00:00Z"
}
response = client.post(
"/api/v1/admin/export",
json=export_request,
headers=headers
)
assert response.status_code == status.HTTP_202_ACCEPTED
data = response.json()
assert "export_id" in data
assert data["status"] == "initiated"
def test_import_data_endpoint(self, client):
"""Test POST /api/v1/admin/import endpoint."""
headers = {"Authorization": "Bearer admin_token"}
# Mock file upload
with patch('src.server.rest_api.AdminAPI.import_data') as mock_import:
mock_import.return_value = {
"import_id": "import_123",
"status": "processing",
"file_size": "25MB",
"estimated_records": 8500,
"validation_results": {
"valid_records": 8450,
"invalid_records": 50,
"warnings": 15
}
}
files = {"file": ("data.json", b"mock file content", "application/json")}
data = {"data_type": "news_articles", "validate_only": False}
response = client.post(
"/api/v1/admin/import",
files=files,
data=data,
headers=headers
)
assert response.status_code == status.HTTP_202_ACCEPTED
response_data = response.json()
assert "import_id" in response_data
def test_system_maintenance_endpoint(self, client):
"""Test POST /api/v1/admin/maintenance endpoint."""
headers = {"Authorization": "Bearer admin_token"}
maintenance_request = {
"operation": "enable_maintenance_mode",
"scheduled_start": "2024-01-02T02:00:00Z",
"estimated_duration": "30m",
"message": "System maintenance in progress"
}
with patch('src.server.rest_api.AdminAPI.system_maintenance') as mock_maintenance:
mock_maintenance.return_value = {
"maintenance_mode": True,
"scheduled_start": "2024-01-02T02:00:00Z",
"estimated_end": "2024-01-02T02:30:00Z",
"status": "scheduled",
"maintenance_id": "maint_123"
}
response = client.post(
"/api/v1/admin/maintenance",
json=maintenance_request,
headers=headers
)
assert response.status_code == status.HTTP_200_OK
data = response.json()
assert data["maintenance_mode"] is True
assert "maintenance_id" in data
class TestAPIErrorHandling:
"""Test cases for API error handling."""
@pytest.fixture
def client(self):
"""Create test client for API testing."""
return TestClient(app)
def test_404_error_handling(self, client):
"""Test 404 error handling."""
response = client.get("/api/v1/nonexistent")
assert response.status_code == status.HTTP_404_NOT_FOUND
data = response.json()
assert "detail" in data
def test_500_error_handling(self, client):
"""Test 500 error handling."""
with patch('src.server.rest_api.NewsAPI.get_news') as mock_get_news:
mock_get_news.side_effect = Exception("Internal server error")
response = client.get("/api/v1/news")
assert response.status_code == status.HTTP_500_INTERNAL_SERVER_ERROR
def test_rate_limiting(self, client):
"""Test API rate limiting."""
# Make multiple requests quickly
responses = []
for i in range(100):
response = client.get("/api/v1/news")
responses.append(response)
# Should eventually hit rate limit
rate_limited = any(r.status_code == status.HTTP_429_TOO_MANY_REQUESTS for r in responses)
assert rate_limited or all(r.status_code == status.HTTP_200_OK for r in responses)
def test_malformed_json_handling(self, client):
"""Test handling of malformed JSON in requests."""
response = client.post(
"/api/v1/analysis/sentiment",
data="invalid json",
headers={"Content-Type": "application/json"}
)
assert response.status_code == status.HTTP_422_UNPROCESSABLE_ENTITY
def test_api_versioning(self, client):
"""Test API versioning."""
# Current version should work
response = client.get("/api/v1/news")
assert response.status_code in [status.HTTP_200_OK, status.HTTP_500_INTERNAL_SERVER_ERROR]
# Non-existent version should return 404
response = client.get("/api/v2/news")
assert response.status_code == status.HTTP_404_NOT_FOUND
def test_cors_headers(self, client):
"""Test CORS headers are present."""
response = client.options("/api/v1/news")
assert "access-control-allow-origin" in [h[0].lower() for h in response.headers.raw]
def test_security_headers(self, client):
"""Test security headers are present."""
response = client.get("/api/v1/news")
headers = dict(response.headers)
# Should have basic security headers
assert "x-content-type-options" in [h.lower() for h in headers.keys()]
