"""
Phase 2 Integration Tests
TDD Green Phase: Integration testing for Phase 2 features
"""
import pytest
import asyncio
from typing import Dict, Any
from src.server import MCPStockDetailsServer
from src.exceptions import MCPStockDetailsError, InsufficientDataError
class TestPhase2Integration:
"""Integration tests for Phase 2 enhanced features"""
@pytest.fixture
def server(self):
"""Create server instance for integration testing"""
return MCPStockDetailsServer()
@pytest.mark.asyncio
async def test_enhanced_company_overview_integration(self, server):
"""Test enhanced company overview end-to-end"""
# Test Samsung Electronics
result = await server._handle_get_company_overview({
"company_code": "005930",
"include_subsidiaries": True
})
assert result is not None
assert len(result) > 0
content = result[0].text
# Should have company overview (enhanced or basic fallback)
assert "Company Overview" in content
assert "005930" in content
assert "Samsung" in content or "Company" in content
# Basic company information should be present
assert "Company Name" in content or "company_name" in content
@pytest.mark.asyncio
async def test_enhanced_financial_statements_integration(self, server):
"""Test enhanced financial statements end-to-end"""
# Test with comprehensive analysis
result = await server._handle_get_financial_statements({
"company_code": "005930",
"year": 2023,
"include_analysis": True,
"include_ratios": True,
"include_trends": True,
"output_format": "detailed",
"cache_ttl": 1800
})
assert result is not None
assert len(result) > 0
content = result[0].text
# Should have comprehensive analysis
assert "DETAILED FINANCIAL STATEMENTS" in content
assert "FINANCIAL STATEMENTS" in content
assert "COMPREHENSIVE FINANCIAL ANALYSIS" in content
assert "FINANCIAL RATIOS" in content
assert "TREND ANALYSIS" in content
assert "Processing Time" in content
@pytest.mark.asyncio
async def test_advanced_analysis_options_integration(self, server):
"""Test advanced analysis options end-to-end"""
result = await server._handle_get_financial_statements({
"company_code": "005930",
"year": 2023,
"analysis_options": {
"include_dupont": True,
"include_cash_flow_analysis": True,
"include_financial_health_score": True,
"include_forecasting": True,
"forecast_years": 2
}
})
assert result is not None
content = result[0].text
# Should have advanced analysis sections
assert "DUPONT ANALYSIS" in content or "dupont_analysis" in content
assert "CASH FLOW ANALYSIS" in content or "cash_flow_analysis" in content
assert "FORECASTING" in content or "forecasting" in content
@pytest.mark.asyncio
async def test_output_formats_integration(self, server):
"""Test different output formats end-to-end"""
# Test detailed format
detailed = await server._handle_get_financial_statements({
"company_code": "005930",
"year": 2023,
"output_format": "detailed"
})
# Test summary format
summary = await server._handle_get_financial_statements({
"company_code": "005930",
"year": 2023,
"output_format": "summary"
})
# Test executive format
executive = await server._handle_get_financial_statements({
"company_code": "005930",
"year": 2023,
"output_format": "executive",
"include_analysis": True
})
# All should return valid content
assert detailed is not None and len(detailed) > 0
assert summary is not None and len(summary) > 0
assert executive is not None and len(executive) > 0
# Check format-specific content
detailed_content = detailed[0].text
summary_content = summary[0].text
executive_content = executive[0].text
assert "DETAILED FINANCIAL STATEMENTS" in detailed_content
assert "FINANCIAL SUMMARY" in summary_content
assert "EXECUTIVE FINANCIAL REPORT" in executive_content
@pytest.mark.asyncio
async def test_caching_integration(self, server):
"""Test caching functionality end-to-end"""
# First call should populate cache
result1 = await server._handle_get_financial_statements({
"company_code": "005930",
"year": 2023,
"include_analysis": True,
"cache_ttl": 3600
})
# Second call should hit cache
result2 = await server._handle_get_financial_statements({
"company_code": "005930",
"year": 2023,
"include_analysis": True,
"cache_ttl": 3600
})
# Results should be identical
assert result1 is not None and result2 is not None
assert result1[0].text == result2[0].text
# Cache should have statistics
if server.cache_manager:
stats = server.cache_manager.stats()
assert stats["total_gets"] >= 2
@pytest.mark.asyncio
async def test_error_handling_integration(self, server):
"""Test error handling end-to-end"""
# Test invalid company code
with pytest.raises((MCPStockDetailsError, InsufficientDataError)):
await server._handle_get_financial_statements({
"company_code": "999999",
"year": 2023,
"include_analysis": True
})
# Test empty company code
with pytest.raises(MCPStockDetailsError):
await server._handle_get_company_overview({
"company_code": ""
})
@pytest.mark.asyncio
async def test_performance_integration(self, server):
"""Test performance requirements end-to-end"""
import time
start_time = time.time()
# Run comprehensive analysis
result = await server._handle_get_financial_statements({
"company_code": "005930",
"year": 2023,
"include_analysis": True,
"include_ratios": True,
"include_trends": True,
"include_peer_comparison": True,
"analysis_options": {
"include_dupont": True,
"include_cash_flow_analysis": True,
"include_forecasting": True
}
})
execution_time = time.time() - start_time
# Should complete within 10 seconds for integration test
assert execution_time < 10.0
assert result is not None
# Content should be rich and comprehensive
content = result[0].text
assert len(content) > 1500 # Should be substantial content (adjusted for actual output)
@pytest.mark.asyncio
async def test_system_stability_integration(self, server):
"""Test system stability with multiple concurrent requests"""
# Run multiple requests concurrently
tasks = []
for i in range(5):
task = server._handle_get_financial_statements({
"company_code": "005930",
"year": 2023,
"include_analysis": True,
"cache_ttl": 1800
})
tasks.append(task)
# All should complete successfully
results = await asyncio.gather(*tasks, return_exceptions=True)
# Check all succeeded
successful_results = [r for r in results if not isinstance(r, Exception)]
assert len(successful_results) == 5
# All should have content
for result in successful_results:
assert result is not None
assert len(result) > 0
assert len(result[0].text) > 100
