test_enhanced_financial_statements.pyβ’16.3 kB
"""
Unit tests for enhanced get_financial_statements functionality
TDD Red Phase: Write failing tests for enhanced financial statements
"""
import pytest
import asyncio
from unittest.mock import Mock, AsyncMock, patch
from datetime import datetime, date
from typing import Any, Dict, List, Optional
# Test imports will initially fail - this is expected in TDD Red phase
from src.server import MCPStockDetailsServer
from src.tools.financial_tools import FinancialAnalyzer
from src.cache import CacheManager, MemoryCache, RedisCache
from src.exceptions import MCPStockDetailsError, InsufficientDataError
from src.config import get_settings
class TestEnhancedFinancialStatements:
"""Test cases for enhanced financial statements functionality"""
@pytest.fixture
def server_with_cache(self):
"""Create server instance with caching enabled"""
# Create cache manager
memory_cache = MemoryCache(max_size=100, default_ttl=1800)
redis_cache = RedisCache(mock_mode=True, default_ttl=3600)
cache_manager = CacheManager(l1_cache=memory_cache, l2_cache=redis_cache)
# Create server with enhanced components
server = MCPStockDetailsServer()
server.cache_manager = cache_manager
server.financial_analyzer = FinancialAnalyzer()
return server
@pytest.fixture
def sample_financial_data(self):
"""Sample financial statement data for testing"""
return {
"005930": {
"2023": {
"income_statement": {
"revenue": 258_774_000_000_000,
"operating_profit": 23_009_000_000_000,
"net_profit": 15_349_000_000_000,
"gross_profit": 125_432_000_000_000
},
"balance_sheet": {
"total_assets": 426_071_000_000_000,
"total_liabilities": 106_387_000_000_000,
"total_equity": 319_684_000_000_000,
"current_assets": 200_000_000_000_000,
"current_liabilities": 80_000_000_000_000
},
"cash_flow": {
"operating_cash_flow": 35_843_000_000_000,
"investing_cash_flow": -45_234_000_000_000,
"financing_cash_flow": -8_456_000_000_000,
"free_cash_flow": 28_567_000_000_000
}
},
"2022": {
"income_statement": {
"revenue": 244_161_000_000_000,
"operating_profit": 21_123_000_000_000,
"net_profit": 13_492_000_000_000,
"gross_profit": 118_234_000_000_000
},
"balance_sheet": {
"total_assets": 399_045_000_000_000,
"total_liabilities": 98_765_000_000_000,
"total_equity": 300_280_000_000_000,
"current_assets": 180_000_000_000_000,
"current_liabilities": 75_000_000_000_000
},
"cash_flow": {
"operating_cash_flow": 32_156_000_000_000,
"investing_cash_flow": -38_945_000_000_000,
"financing_cash_flow": -7_234_000_000_000,
"free_cash_flow": 25_432_000_000_000
}
}
}
}
@pytest.mark.asyncio
async def test_enhanced_financial_statements_basic(self, server_with_cache, sample_financial_data):
"""Test enhanced financial statements with basic parameters"""
company_code = "005930"
year = 2023
# Mock the DART collector to return sample data
with patch.object(server_with_cache.dart_collector, 'get_financial_statements') as mock_get_statements:
mock_get_statements.return_value = sample_financial_data[company_code][str(year)]
# Call enhanced get_financial_statements
result = await server_with_cache._handle_get_financial_statements({
"company_code": company_code,
"year": year,
"include_analysis": True,
"include_ratios": True,
"include_trends": False,
"cache_ttl": 1800
})
assert result is not None
assert len(result) > 0
# Parse result content
content = result[0].text
assert company_code in content
assert str(year) in content
assert "COMPREHENSIVE FINANCIAL ANALYSIS" in content
assert "FINANCIAL RATIOS" in content
assert "INCOME STATEMENT" in content
assert "BALANCE SHEET" in content
assert "CASH FLOW" in content
@pytest.mark.asyncio
async def test_enhanced_financial_statements_with_trends(self, server_with_cache, sample_financial_data):
"""Test enhanced financial statements with trend analysis"""
company_code = "005930"
year = 2023
with patch.object(server_with_cache.dart_collector, 'get_financial_statements') as mock_get_statements:
# Return multi-year data for trend analysis
mock_get_statements.side_effect = lambda company_code, year: sample_financial_data[company_code].get(str(year), {})
result = await server_with_cache._handle_get_financial_statements({
"company_code": company_code,
"year": year,
"include_analysis": True,
"include_ratios": True,
"include_trends": True,
"trend_years": 2,
"cache_ttl": 3600
})
assert result is not None
content = result[0].text
# Should include trend analysis
assert "trend_analysis" in content or "TREND ANALYSIS" in content
@pytest.mark.asyncio
async def test_enhanced_financial_statements_quarterly(self, server_with_cache):
"""Test enhanced financial statements with quarterly data"""
company_code = "005930"
year = 2023
quarter = 4
with patch.object(server_with_cache.dart_collector, 'get_quarterly_statements') as mock_get_quarterly:
mock_quarterly_data = {
"Q4_2023": {
"revenue": 67_783_000_000_000,
"operating_profit": 6_234_000_000_000,
"net_profit": 4_123_000_000_000
}
}
mock_get_quarterly.return_value = mock_quarterly_data
result = await server_with_cache._handle_get_financial_statements({
"company_code": company_code,
"year": year,
"quarter": quarter,
"include_analysis": True,
"include_seasonal_analysis": True
})
assert result is not None
content = result[0].text
assert "QUARTERLY" in content or f"Q{quarter}" in content
assert "seasonal_analysis" in content or "SEASONAL ANALYSIS" in content
@pytest.mark.asyncio
async def test_enhanced_financial_statements_caching(self, server_with_cache, sample_financial_data):
"""Test caching functionality for enhanced financial statements"""
company_code = "005930"
year = 2023
with patch.object(server_with_cache.dart_collector, 'get_financial_statements') as mock_get_statements:
mock_get_statements.return_value = sample_financial_data[company_code][str(year)]
# First call should hit the database
result1 = await server_with_cache._handle_get_financial_statements({
"company_code": company_code,
"year": year,
"include_analysis": True,
"cache_ttl": 1800
})
# Second call should hit the cache
result2 = await server_with_cache._handle_get_financial_statements({
"company_code": company_code,
"year": year,
"include_analysis": True,
"cache_ttl": 1800
})
# DART collector should only be called once
assert mock_get_statements.call_count == 1
# Results should be identical
assert result1[0].text == result2[0].text
# Check cache statistics
cache_stats = server_with_cache.cache_manager.stats()
assert cache_stats["l1_hits"] >= 1 or cache_stats["l2_hits"] >= 1
@pytest.mark.asyncio
async def test_enhanced_financial_statements_peer_comparison(self, server_with_cache, sample_financial_data):
"""Test enhanced financial statements with peer comparison"""
company_code = "005930"
year = 2023
with patch.object(server_with_cache.dart_collector, 'get_financial_statements') as mock_get_statements:
mock_get_statements.return_value = sample_financial_data[company_code][str(year)]
result = await server_with_cache._handle_get_financial_statements({
"company_code": company_code,
"year": year,
"include_analysis": True,
"include_peer_comparison": True,
"industry_code": "26211" # Semiconductor industry
})
assert result is not None
content = result[0].text
assert "peer_comparison" in content or "PEER COMPARISON" in content
assert "industry_benchmarks" in content or "Analysis completed successfully" in content
assert "relative_performance" in content or "Performance vs Peers" in content
@pytest.mark.asyncio
async def test_enhanced_financial_statements_custom_analysis(self, server_with_cache, sample_financial_data):
"""Test enhanced financial statements with custom analysis options"""
company_code = "005930"
year = 2023
with patch.object(server_with_cache.dart_collector, 'get_financial_statements') as mock_get_statements:
mock_get_statements.return_value = sample_financial_data[company_code][str(year)]
result = await server_with_cache._handle_get_financial_statements({
"company_code": company_code,
"year": year,
"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
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_enhanced_financial_statements_error_handling(self, server_with_cache):
"""Test error handling for enhanced financial statements"""
# Test invalid company code
with pytest.raises(MCPStockDetailsError):
await server_with_cache._handle_get_financial_statements({
"company_code": "",
"year": 2023
})
# Test future year
with pytest.raises(MCPStockDetailsError):
await server_with_cache._handle_get_financial_statements({
"company_code": "005930",
"year": 2031 # Beyond the allowed range
})
# Test insufficient data - company code that triggers InsufficientDataError
with patch.object(server_with_cache.dart_collector, 'get_financial_statements') as mock_get_statements:
# Mock to return None to trigger insufficient data
mock_get_statements.return_value = None
with pytest.raises((MCPStockDetailsError, InsufficientDataError)):
await server_with_cache._handle_get_financial_statements({
"company_code": "999999",
"year": 2023,
"include_analysis": True,
"include_trends": True,
"trend_years": 5
})
@pytest.mark.asyncio
async def test_enhanced_financial_statements_output_formats(self, server_with_cache, sample_financial_data):
"""Test different output formats for enhanced financial statements"""
company_code = "005930"
year = 2023
with patch.object(server_with_cache.dart_collector, 'get_financial_statements') as mock_get_statements:
mock_get_statements.return_value = sample_financial_data[company_code][str(year)]
# Test detailed format (default)
result_detailed = await server_with_cache._handle_get_financial_statements({
"company_code": company_code,
"year": year,
"output_format": "detailed"
})
# Test summary format
result_summary = await server_with_cache._handle_get_financial_statements({
"company_code": company_code,
"year": year,
"output_format": "summary"
})
# Test executive format
result_executive = await server_with_cache._handle_get_financial_statements({
"company_code": company_code,
"year": year,
"output_format": "executive",
"include_analysis": True
})
# All formats should return content
assert result_detailed is not None
assert result_summary is not None
assert result_executive is not None
# Executive format should be more concise
exec_content = result_executive[0].text
detailed_content = result_detailed[0].text
# Executive should have key insights
assert "KEY INSIGHTS" in exec_content
assert "EXECUTIVE SUMMARY" in exec_content
@pytest.mark.asyncio
async def test_enhanced_financial_statements_performance(self, server_with_cache, sample_financial_data):
"""Test performance optimizations for enhanced financial statements"""
company_code = "005930"
year = 2023
with patch.object(server_with_cache.dart_collector, 'get_financial_statements') as mock_get_statements:
mock_get_statements.return_value = sample_financial_data[company_code][str(year)]
import time
start_time = time.time()
# Call with all enhancements
result = await server_with_cache._handle_get_financial_statements({
"company_code": company_code,
"year": year,
"include_analysis": True,
"include_ratios": True,
"include_trends": True,
"include_peer_comparison": True,
"trend_years": 3,
"cache_ttl": 3600
})
execution_time = time.time() - start_time
# Should complete within reasonable time (5 seconds)
assert execution_time < 5.0
assert result is not None
# Performance metrics should be included
content = result[0].text
assert "processing_time" in content or len(content) > 1000 # Rich content indicates processing