"""
Service Integration Tests
Tests for API, AI, and visualization microservices.
Verifies service health, endpoints, and inter-service communication.
"""
import pytest
import json
from typing import Dict, Any
class TestMainAPIService:
"""Tests for main API service with C tools integration."""
def test_api_health_check(self, api_client):
"""Test main API health check endpoint."""
response = api_client.get("/health")
assert response.status_code == 200
data = response.json()
assert data["status"] in ["healthy", "ok"]
assert "tools" in data or "message" in data
def test_api_root_endpoint(self, api_client):
"""Test main API root endpoint."""
response = api_client.get("/")
assert response.status_code == 200
data = response.json()
assert data["name"] == "Tikker API"
assert "version" in data
assert "endpoints" in data
def test_get_daily_stats(self, api_client):
"""Test daily statistics endpoint."""
response = api_client.get("/api/stats/daily")
assert response.status_code in [200, 503]
if response.status_code == 200:
data = response.json()
assert "presses" in data or "status" in data
def test_get_top_words(self, api_client):
"""Test top words endpoint."""
response = api_client.get("/api/words/top?limit=10")
assert response.status_code in [200, 503]
if response.status_code == 200:
data = response.json()
assert isinstance(data, list) or isinstance(data, dict)
def test_decode_file_endpoint(self, api_client):
"""Test file decoding endpoint."""
payload = {
"input_file": "test_input.txt",
"output_file": "test_output.txt",
"verbose": False
}
response = api_client.post("/api/decode", json=payload)
assert response.status_code in [200, 400, 404, 503]
def test_api_health_timeout(self, api_client):
"""Test API health endpoint response time."""
import time
start = time.time()
response = api_client.get("/health")
elapsed = time.time() - start
assert elapsed < 5.0
assert response.status_code == 200
class TestAIService:
"""Tests for AI microservice."""
def test_ai_health_check(self, ai_client):
"""Test AI service health check."""
response = ai_client.get("/health")
assert response.status_code == 200
data = response.json()
assert data["status"] == "healthy"
assert "ai_available" in data
def test_ai_root_endpoint(self, ai_client):
"""Test AI service root endpoint."""
response = ai_client.get("/")
assert response.status_code == 200
data = response.json()
assert data["name"] == "Tikker AI Service"
assert "endpoints" in data
def test_ai_analyze_endpoint(self, ai_client):
"""Test AI text analysis endpoint."""
payload = {
"text": "This is a test message for analysis",
"analysis_type": "general"
}
response = ai_client.post("/analyze", json=payload)
assert response.status_code in [200, 503]
def test_ai_analyze_activity(self, ai_client):
"""Test AI activity analysis."""
payload = {
"text": "typing keyboard input keystroke logs",
"analysis_type": "activity"
}
response = ai_client.post("/analyze", json=payload)
assert response.status_code in [200, 503]
def test_ai_empty_text_validation(self, ai_client):
"""Test AI service rejects empty text."""
payload = {
"text": "",
"analysis_type": "general"
}
response = ai_client.post("/analyze", json=payload)
if response.status_code == 503:
pass
else:
assert response.status_code == 400
class TestVizService:
"""Tests for visualization microservice."""
def test_viz_health_check(self, viz_client):
"""Test visualization service health check."""
response = viz_client.get("/health")
assert response.status_code == 200
data = response.json()
assert data["status"] == "healthy"
assert "viz_available" in data
def test_viz_root_endpoint(self, viz_client):
"""Test visualization service root endpoint."""
response = viz_client.get("/")
assert response.status_code == 200
data = response.json()
assert data["name"] == "Tikker Visualization Service"
assert "supported_charts" in data
def test_viz_bar_chart(self, viz_client):
"""Test bar chart generation."""
payload = {
"title": "Test Bar Chart",
"data": {"A": 10, "B": 20, "C": 15},
"chart_type": "bar",
"width": 10,
"height": 6
}
response = viz_client.post("/chart", json=payload)
assert response.status_code in [200, 503]
if response.status_code == 200:
data = response.json()
assert data["status"] == "success"
assert data["chart_type"] == "bar"
assert "image_base64" in data
def test_viz_line_chart(self, viz_client):
"""Test line chart generation."""
payload = {
"title": "Test Line Chart",
"data": {"Jan": 100, "Feb": 120, "Mar": 140},
"chart_type": "line"
}
response = viz_client.post("/chart", json=payload)
assert response.status_code in [200, 503]
def test_viz_pie_chart(self, viz_client):
"""Test pie chart generation."""
payload = {
"title": "Test Pie Chart",
"data": {"Category1": 30, "Category2": 40, "Category3": 30},
"chart_type": "pie"
}
response = viz_client.post("/chart", json=payload)
assert response.status_code in [200, 503]
def test_viz_chart_download(self, viz_client):
"""Test chart download endpoint."""
payload = {
"title": "Download Test",
"data": {"X": 50, "Y": 75},
"chart_type": "bar"
}
response = viz_client.post("/chart/download", json=payload)
assert response.status_code in [200, 503]
def test_viz_invalid_chart_type(self, viz_client):
"""Test invalid chart type handling."""
payload = {
"title": "Invalid Chart",
"data": {"A": 10},
"chart_type": "invalid"
}
response = viz_client.post("/chart", json=payload)
if response.status_code == 503:
pass
else:
assert response.status_code == 400
class TestServiceIntegration:
"""Tests for service-to-service communication."""
def test_all_services_healthy(self, api_client, ai_client, viz_client):
"""Test all services report healthy status."""
api_response = api_client.get("/health")
ai_response = ai_client.get("/health")
viz_response = viz_client.get("/health")
assert api_response.status_code == 200
assert ai_response.status_code == 200
assert viz_response.status_code == 200
def test_api_to_ai_communication(self, api_client, ai_client):
"""Test API can communicate with AI service."""
api_health = api_client.get("/health")
ai_health = ai_client.get("/health")
assert api_health.status_code == 200
assert ai_health.status_code == 200
def test_api_to_viz_communication(self, api_client, viz_client):
"""Test API can communicate with visualization service."""
api_health = api_client.get("/health")
viz_health = viz_client.get("/health")
assert api_health.status_code == 200
assert viz_health.status_code == 200
def test_concurrent_service_requests(self, api_client, ai_client, viz_client):
"""Test multiple concurrent requests to different services."""
responses = {
"api": api_client.get("/health"),
"ai": ai_client.get("/health"),
"viz": viz_client.get("/health")
}
for service, response in responses.items():
assert response.status_code == 200, f"{service} service failed"
class TestErrorHandling:
"""Tests for error handling and edge cases."""
def test_api_invalid_endpoint(self, api_client):
"""Test API handles invalid endpoints."""
response = api_client.get("/api/invalid")
assert response.status_code == 404
def test_ai_invalid_endpoint(self, ai_client):
"""Test AI service handles invalid endpoints."""
response = ai_client.get("/invalid")
assert response.status_code == 404
def test_viz_invalid_endpoint(self, viz_client):
"""Test visualization service handles invalid endpoints."""
response = viz_client.get("/invalid")
assert response.status_code == 404
def test_api_malformed_json(self, api_client):
"""Test API handles malformed JSON."""
response = api_client.post(
"/api/decode",
content="invalid json",
headers={"Content-Type": "application/json"}
)
assert response.status_code in [400, 422]
def test_ai_malformed_json(self, ai_client):
"""Test AI service handles malformed JSON."""
response = ai_client.post(
"/analyze",
content="invalid json",
headers={"Content-Type": "application/json"}
)
assert response.status_code in [400, 422]
@pytest.fixture
def api_client():
"""Create API test client."""
from fastapi.testclient import TestClient
try:
from api_c_integration import app
return TestClient(app)
except:
return None
@pytest.fixture
def ai_client():
"""Create AI service test client."""
from fastapi.testclient import TestClient
try:
from ai_service import app
return TestClient(app)
except:
return None
@pytest.fixture
def viz_client():
"""Create visualization service test client."""
from fastapi.testclient import TestClient
try:
from viz_service import app
return TestClient(app)
except:
return None
Reference in New Issue View Git Blame Copy Permalink