File size: 10,536 Bytes
2469150 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 |
#!/usr/bin/env python3
"""
System integration tests for FRED ML
Tests complete workflows and system integration
"""
import sys
import os
import pytest
import pandas as pd
import numpy as np
import requests
from unittest.mock import Mock, patch
from datetime import datetime, timedelta
# Add project root to path
sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), '../..')))
class TestSystemIntegration:
"""Test complete system integration and workflows"""
def setup_method(self):
"""Set up test environment"""
self.test_api_key = "test_fred_api_key"
self.test_data = pd.DataFrame({
'GDPC1': [22000, 22100, 22200, 22300, 22400],
'INDPRO': [100, 101, 102, 103, 104],
'CPIAUCSL': [250, 251, 252, 253, 254],
'FEDFUNDS': [2.0, 2.1, 2.2, 2.3, 2.4],
'PAYEMS': [150000, 151000, 152000, 153000, 154000]
}, index=pd.date_range('2023-01-01', periods=5, freq='M'))
def test_complete_analytics_workflow(self):
"""Test complete analytics workflow"""
try:
# Test analytics module import
from src.analysis.comprehensive_analytics import ComprehensiveAnalytics
# Create analytics instance
analytics = ComprehensiveAnalytics(self.test_api_key, output_dir="test_output")
# Test basic functionality
assert analytics is not None
assert hasattr(analytics, 'run_complete_analysis')
except ImportError as e:
pytest.skip(f"Analytics module not available: {e}")
except Exception as e:
pytest.fail(f"Analytics workflow test failed: {e}")
def test_fred_client_integration(self):
"""Test FRED client integration"""
try:
from src.core.enhanced_fred_client import EnhancedFREDClient
client = EnhancedFREDClient(self.test_api_key)
# Test client structure
assert hasattr(client, 'fetch_economic_data')
assert hasattr(client, 'fetch_quarterly_data')
except ImportError as e:
pytest.skip(f"FRED client module not available: {e}")
except Exception as e:
pytest.fail(f"FRED client integration test failed: {e}")
def test_mathematical_fixes_integration(self):
"""Test mathematical fixes integration"""
try:
from src.analysis.mathematical_fixes import MathematicalFixes
fixes = MathematicalFixes()
# Test comprehensive fixes
fixed_data, fix_info = fixes.apply_comprehensive_fixes(
self.test_data,
target_freq='Q',
growth_method='pct_change',
normalize_units=True
)
assert fixed_data is not None
assert isinstance(fix_info, dict)
except ImportError as e:
pytest.skip(f"Mathematical fixes module not available: {e}")
except Exception as e:
pytest.fail(f"Mathematical fixes integration test failed: {e}")
def test_forecasting_integration(self):
"""Test forecasting module integration"""
try:
from src.analysis.economic_forecasting import EconomicForecaster
forecaster = EconomicForecaster(self.test_data)
# Test forecaster structure
assert hasattr(forecaster, 'forecast_series')
assert hasattr(forecaster, 'backtest_forecast')
except ImportError as e:
pytest.skip(f"Forecasting module not available: {e}")
except Exception as e:
pytest.fail(f"Forecasting integration test failed: {e}")
def test_segmentation_integration(self):
"""Test segmentation module integration"""
try:
from src.analysis.economic_segmentation import EconomicSegmentation
segmentation = EconomicSegmentation(self.test_data)
# Test segmentation structure
assert hasattr(segmentation, 'cluster_time_periods')
assert hasattr(segmentation, 'cluster_economic_series')
except ImportError as e:
pytest.skip(f"Segmentation module not available: {e}")
except Exception as e:
pytest.fail(f"Segmentation integration test failed: {e}")
class TestAppIntegration:
"""Test application integration and functionality"""
@patch('requests.get')
def test_app_health_check(self, mock_get):
"""Test app health check with mocked requests"""
# Mock successful response
mock_response = Mock()
mock_response.status_code = 200
mock_get.return_value = mock_response
try:
response = requests.get("http://localhost:8501/_stcore/health", timeout=5)
assert response.status_code == 200
except Exception as e:
pytest.fail(f"App health check test failed: {e}")
def test_config_integration(self):
"""Test configuration integration"""
try:
# Test environment variable loading
import os
fred_key = os.getenv('FRED_API_KEY', 'test_key')
assert fred_key is not None
assert len(fred_key) > 0
except Exception as e:
pytest.fail(f"Config integration test failed: {e}")
def test_data_processing_integration(self):
"""Test data processing integration"""
# Create test data
test_data = pd.DataFrame({
'GDPC1': [22000, 22100, 22200],
'INDPRO': [100, 101, 102],
'CPIAUCSL': [250, 251, 252]
})
# Test basic data processing
assert not test_data.empty
assert len(test_data.columns) > 0
assert len(test_data) > 0
# Test data validation
for col in test_data.columns:
assert pd.api.types.is_numeric_dtype(test_data[col])
def test_visualization_integration(self):
"""Test visualization integration"""
try:
import matplotlib.pyplot as plt
# Test basic plotting
test_data = pd.DataFrame({
'GDPC1': [22000, 22100, 22200],
'INDPRO': [100, 101, 102]
})
fig, ax = plt.subplots()
test_data.plot(ax=ax)
plt.close()
assert True # If we get here, plotting worked
except ImportError:
pytest.skip("Matplotlib not available")
except Exception as e:
pytest.fail(f"Visualization integration test failed: {e}")
class TestErrorHandling:
"""Test error handling and edge cases"""
def test_invalid_api_key_handling(self):
"""Test handling of invalid API key"""
try:
from src.core.enhanced_fred_client import EnhancedFREDClient
# Test with invalid key
client = EnhancedFREDClient("invalid_key")
# Should not raise exception on initialization
assert client is not None
except ImportError as e:
pytest.skip(f"FRED client module not available: {e}")
except Exception as e:
pytest.fail(f"Invalid API key handling test failed: {e}")
def test_empty_data_handling(self):
"""Test handling of empty data"""
try:
from src.analysis.mathematical_fixes import MathematicalFixes
fixes = MathematicalFixes()
# Test with empty DataFrame
empty_data = pd.DataFrame()
# Should handle empty data gracefully
if hasattr(fixes, 'normalize_units'):
try:
result = fixes.normalize_units(empty_data)
assert result is not None
except Exception:
# It's OK for empty data to raise an exception
pass
except ImportError as e:
pytest.skip(f"Mathematical fixes module not available: {e}")
except Exception as e:
pytest.fail(f"Empty data handling test failed: {e}")
def test_missing_data_handling(self):
"""Test handling of missing data"""
# Create data with missing values
test_data = pd.DataFrame({
'GDPC1': [22000, np.nan, 22200],
'INDPRO': [100, 101, np.nan],
'CPIAUCSL': [250, 251, 252]
})
# Test basic missing data handling
assert test_data.isnull().sum().sum() > 0
# Test that we can handle missing data
cleaned_data = test_data.dropna()
assert len(cleaned_data) < len(test_data)
class TestPerformance:
"""Test performance and scalability"""
def test_large_data_handling(self):
"""Test handling of large datasets"""
# Create larger test dataset
large_data = pd.DataFrame({
'GDPC1': np.random.normal(22000, 1000, 1000),
'INDPRO': np.random.normal(100, 5, 1000),
'CPIAUCSL': np.random.normal(250, 10, 1000)
}, index=pd.date_range('2020-01-01', periods=1000, freq='D'))
# Test basic operations on large data
assert len(large_data) == 1000
assert len(large_data.columns) == 3
# Test memory usage (basic check)
memory_usage = large_data.memory_usage(deep=True).sum()
assert memory_usage > 0
def test_computation_efficiency(self):
"""Test computation efficiency"""
import time
# Test basic computation time
start_time = time.time()
# Perform some basic computations
test_data = pd.DataFrame({
'GDPC1': np.random.normal(22000, 1000, 100),
'INDPRO': np.random.normal(100, 5, 100)
})
# Basic operations
test_data.describe()
test_data.corr()
end_time = time.time()
computation_time = end_time - start_time
# Should complete within reasonable time (5 seconds)
assert computation_time < 5.0
if __name__ == "__main__":
pytest.main([__file__]) |