File size: 7,028 Bytes
2b395f2 6ce20d9 2b395f2 6ce20d9 2b395f2 6ce20d9 2b395f2 6ce20d9 2b395f2 6ce20d9 2b395f2 6ce20d9 2b395f2 6ce20d9 2b395f2 6ce20d9 2b395f2 2469150 2b395f2 6ce20d9 2b395f2 6ce20d9 2b395f2 6ce20d9 2b395f2 6ce20d9 2b395f2 2469150 2b395f2 6ce20d9 2b395f2 6ce20d9 2b395f2 6ce20d9 2469150 6ce20d9 2b395f2 2469150 6ce20d9 2469150 6ce20d9 2b395f2 6ce20d9 2469150 6ce20d9 2469150 6ce20d9 2469150 6ce20d9 2b395f2 6ce20d9 2469150 2b395f2 6ce20d9 2469150 6ce20d9 2469150 6ce20d9 2469150 6ce20d9 2469150 6ce20d9 |
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 |
#!/usr/bin/env python3
"""
Unit tests for FRED ML Lambda Function
Tests core functionality without AWS dependencies
"""
import pytest
import sys
import json
import pandas as pd
import numpy as np
from unittest.mock import Mock, patch
from pathlib import Path
# Add src to path
project_root = Path(__file__).parent.parent.parent
sys.path.append(str(project_root / 'src'))
class TestLambdaFunction:
"""Test cases for Lambda function core functionality"""
@pytest.fixture
def mock_event(self):
"""Mock Lambda event"""
return {
'indicators': ['GDP', 'UNRATE'],
'start_date': '2024-01-01',
'end_date': '2024-01-31',
'options': {
'visualizations': True,
'correlation': True,
'statistics': True
}
}
@pytest.fixture
def mock_context(self):
"""Mock Lambda context"""
context = Mock()
context.function_name = 'fred-ml-processor'
context.function_version = '$LATEST'
context.invoked_function_arn = 'arn:aws:lambda:us-west-2:123456789012:function:fred-ml-processor'
context.memory_limit_in_mb = 512
context.remaining_time_in_millis = 300000
return context
def test_create_dataframe(self):
"""Test DataFrame creation from series data"""
from src.lambda_fn.lambda_function import create_dataframe
# Create mock series data
dates = pd.date_range('2024-01-01', '2024-01-05', freq='D')
series_data = {
'GDP': pd.Series([100.0, 101.0, 102.0, 103.0, 104.0], index=dates),
'UNRATE': pd.Series([3.5, 3.6, 3.7, 3.8, 3.9], index=dates)
}
df = create_dataframe(series_data)
assert not df.empty
assert 'GDP' in df.columns
assert 'UNRATE' in df.columns
assert len(df) == 5
assert df.index.name == 'Date'
def test_generate_statistics(self):
"""Test statistics generation"""
from src.lambda_fn.lambda_function import generate_statistics
# Create test DataFrame
dates = pd.date_range('2024-01-01', '2024-01-05', freq='D')
df = pd.DataFrame({
'GDP': [100.0, 101.0, 102.0, 103.0, 104.0],
'UNRATE': [3.5, 3.6, 3.7, 3.8, 3.9]
}, index=dates)
stats = generate_statistics(df)
assert 'GDP' in stats
assert 'UNRATE' in stats
assert 'mean' in stats['GDP']
assert 'std' in stats['GDP']
assert 'min' in stats['GDP']
assert 'max' in stats['GDP']
assert 'count' in stats['GDP']
assert 'missing' in stats['GDP']
# Verify calculations
assert stats['GDP']['mean'] == 102.0
assert stats['GDP']['min'] == 100.0
assert stats['GDP']['max'] == 104.0
assert stats['GDP']['count'] == 5
def test_create_correlation_matrix(self):
"""Test correlation matrix creation"""
from src.lambda_fn.lambda_function import create_correlation_matrix
# Create test DataFrame
dates = pd.date_range('2024-01-01', '2024-01-05', freq='D')
df = pd.DataFrame({
'GDP': [100.0, 101.0, 102.0, 103.0, 104.0],
'UNRATE': [3.5, 3.6, 3.7, 3.8, 3.9]
}, index=dates)
corr_matrix = create_correlation_matrix(df)
assert 'GDP' in corr_matrix
assert 'UNRATE' in corr_matrix
assert 'GDP' in corr_matrix['GDP']
assert 'UNRATE' in corr_matrix['UNRATE']
# Verify correlation values
assert corr_matrix['GDP']['GDP'] == 1.0
assert corr_matrix['UNRATE']['UNRATE'] == 1.0
@patch('src.lambda_fn.lambda_function.requests.get')
def test_get_fred_data_success(self, mock_requests):
"""Test successful FRED data fetching"""
from src.lambda_fn.lambda_function import get_fred_data
# Mock successful API response
mock_response = Mock()
mock_response.status_code = 200
mock_response.json.return_value = {
'observations': [
{'date': '2024-01-01', 'value': '100.0'},
{'date': '2024-01-02', 'value': '101.0'},
{'date': '2024-01-03', 'value': '102.0'}
]
}
mock_requests.return_value = mock_response
# Mock environment variable
with patch('src.lambda_fn.lambda_function.FRED_API_KEY', 'test-api-key'):
result = get_fred_data('GDP', '2024-01-01', '2024-01-03')
assert result is not None
assert len(result) == 3
assert result.name == 'GDP'
assert result.iloc[0] == 100.0
assert result.iloc[1] == 101.0
assert result.iloc[2] == 102.0
@patch('src.lambda_fn.lambda_function.requests.get')
def test_get_fred_data_failure(self, mock_requests):
"""Test FRED data fetching failure"""
from src.lambda_fn.lambda_function import get_fred_data
# Mock failed API response
mock_response = Mock()
mock_response.status_code = 404
mock_requests.return_value = mock_response
result = get_fred_data('INVALID', '2024-01-01', '2024-01-03')
assert result is None
def test_create_dataframe_empty_data(self):
"""Test DataFrame creation with empty data"""
from src.lambda_fn.lambda_function import create_dataframe
# Test with empty series data
df = create_dataframe({})
assert df.empty
# Test with None values
df = create_dataframe({'GDP': None, 'UNRATE': None})
assert df.empty
def test_generate_statistics_empty_data(self):
"""Test statistics generation with empty data"""
from src.lambda_fn.lambda_function import generate_statistics
# Test with empty DataFrame
df = pd.DataFrame()
stats = generate_statistics(df)
assert stats == {}
# Test with DataFrame containing only NaN values
df = pd.DataFrame({
'GDP': [np.nan, np.nan, np.nan],
'UNRATE': [np.nan, np.nan, np.nan]
})
stats = generate_statistics(df)
# When all values are NaN, the function should return empty stats
assert stats == {}
def test_create_correlation_matrix_empty_data(self):
"""Test correlation matrix creation with empty data"""
from src.lambda_fn.lambda_function import create_correlation_matrix
# Test with empty DataFrame
df = pd.DataFrame()
corr_matrix = create_correlation_matrix(df)
assert corr_matrix == {}
# Test with single column
df = pd.DataFrame({'GDP': [100.0, 101.0, 102.0]})
corr_matrix = create_correlation_matrix(df)
assert 'GDP' in corr_matrix
assert corr_matrix['GDP']['GDP'] == 1.0 |