398 lines
14 KiB
Python
398 lines
14 KiB
Python
"""Tests for fundamental data providers — parsing, error handling, and rotation."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from datetime import datetime, timezone
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
|
|
import httpx
|
|
import pytest
|
|
|
|
from shared.fundamentals.base import FundamentalsProvider, FundamentalsSnapshot
|
|
from shared.fundamentals.alpha_vantage import (
|
|
AlphaVantageProvider,
|
|
_safe_float,
|
|
_safe_float_div100,
|
|
)
|
|
from shared.fundamentals.fmp import FMPProvider
|
|
from shared.fundamentals.yahoo import YahooFinanceProvider
|
|
from shared.fundamentals.rotating import RotatingProvider
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# FundamentalsSnapshot schema tests
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestFundamentalsSnapshot:
|
|
def test_create_with_all_fields(self) -> None:
|
|
snap = FundamentalsSnapshot(
|
|
ticker="AAPL",
|
|
eps_ttm=6.57,
|
|
pe_ratio=28.3,
|
|
peg_ratio=2.1,
|
|
revenue_growth_yoy=0.08,
|
|
profit_margin=0.26,
|
|
debt_to_equity=1.87,
|
|
market_cap=2_800_000_000_000.0,
|
|
fetched_at=datetime.now(timezone.utc),
|
|
)
|
|
assert snap.ticker == "AAPL"
|
|
assert snap.eps_ttm == 6.57
|
|
assert snap.pe_ratio == 28.3
|
|
assert snap.peg_ratio == 2.1
|
|
assert snap.revenue_growth_yoy == 0.08
|
|
assert snap.profit_margin == 0.26
|
|
assert snap.debt_to_equity == 1.87
|
|
assert snap.market_cap == 2_800_000_000_000.0
|
|
assert isinstance(snap.fetched_at, datetime)
|
|
|
|
def test_create_with_optional_fields_none(self) -> None:
|
|
snap = FundamentalsSnapshot(ticker="XYZ", fetched_at=datetime.now(timezone.utc))
|
|
assert snap.ticker == "XYZ"
|
|
assert snap.eps_ttm is None
|
|
assert snap.pe_ratio is None
|
|
assert snap.peg_ratio is None
|
|
assert snap.revenue_growth_yoy is None
|
|
assert snap.profit_margin is None
|
|
assert snap.debt_to_equity is None
|
|
assert snap.market_cap is None
|
|
|
|
def test_serialization_round_trip(self) -> None:
|
|
snap = FundamentalsSnapshot(ticker="MSFT", eps_ttm=11.0, pe_ratio=35.0, fetched_at=datetime.now(timezone.utc))
|
|
data = snap.model_dump()
|
|
rebuilt = FundamentalsSnapshot.model_validate(data)
|
|
assert rebuilt.ticker == "MSFT"
|
|
assert rebuilt.eps_ttm == 11.0
|
|
assert rebuilt.pe_ratio == 35.0
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Helper function tests
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestHelpers:
|
|
def test_safe_float_valid(self) -> None:
|
|
assert _safe_float("3.14") == 3.14
|
|
|
|
def test_safe_float_none(self) -> None:
|
|
assert _safe_float(None) is None
|
|
|
|
def test_safe_float_dash(self) -> None:
|
|
assert _safe_float("-") is None
|
|
|
|
def test_safe_float_empty(self) -> None:
|
|
assert _safe_float("") is None
|
|
|
|
def test_safe_float_none_string(self) -> None:
|
|
assert _safe_float("None") is None
|
|
|
|
def test_safe_float_invalid(self) -> None:
|
|
assert _safe_float("abc") is None
|
|
|
|
def test_safe_float_div100(self) -> None:
|
|
assert _safe_float_div100("150") == 1.5
|
|
|
|
def test_safe_float_div100_none(self) -> None:
|
|
assert _safe_float_div100(None) is None
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Alpha Vantage provider tests
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _mock_av_response(json_data: dict, status_code: int = 200) -> httpx.Response:
|
|
"""Build a fake httpx.Response with the given JSON body."""
|
|
return httpx.Response(
|
|
status_code=status_code,
|
|
json=json_data,
|
|
request=httpx.Request("GET", "https://www.alphavantage.co/query"),
|
|
)
|
|
|
|
|
|
class TestAlphaVantageProvider:
|
|
@pytest.fixture
|
|
def provider(self) -> AlphaVantageProvider:
|
|
return AlphaVantageProvider(api_key="test-key")
|
|
|
|
async def test_parse_valid_response(self, provider: AlphaVantageProvider) -> None:
|
|
mock_data = {
|
|
"Symbol": "AAPL",
|
|
"EPS": "6.57",
|
|
"PERatio": "28.3",
|
|
"PEGRatio": "2.1",
|
|
"QuarterlyRevenueGrowthYOY": "0.08",
|
|
"ProfitMargin": "0.26",
|
|
"DebtToEquity": "187",
|
|
"MarketCapitalization": "2800000000000",
|
|
}
|
|
provider._client = AsyncMock()
|
|
provider._client.get = AsyncMock(return_value=_mock_av_response(mock_data))
|
|
|
|
result = await provider.fetch("AAPL")
|
|
|
|
assert result is not None
|
|
assert result.ticker == "AAPL"
|
|
assert result.eps_ttm == 6.57
|
|
assert result.pe_ratio == 28.3
|
|
assert result.peg_ratio == 2.1
|
|
assert result.revenue_growth_yoy == 0.08
|
|
assert result.profit_margin == 0.26
|
|
assert result.debt_to_equity == pytest.approx(1.87)
|
|
assert result.market_cap == 2_800_000_000_000.0
|
|
|
|
async def test_rate_limit_note(self, provider: AlphaVantageProvider) -> None:
|
|
mock_data = {
|
|
"Note": "Thank you for using Alpha Vantage! Our API call frequency limit is 5 calls per minute."
|
|
}
|
|
provider._client = AsyncMock()
|
|
provider._client.get = AsyncMock(return_value=_mock_av_response(mock_data))
|
|
|
|
result = await provider.fetch("AAPL")
|
|
assert result is None
|
|
|
|
async def test_rate_limit_information(self, provider: AlphaVantageProvider) -> None:
|
|
mock_data = {
|
|
"Information": "Please consider upgrading to a premium plan."
|
|
}
|
|
provider._client = AsyncMock()
|
|
provider._client.get = AsyncMock(return_value=_mock_av_response(mock_data))
|
|
|
|
result = await provider.fetch("AAPL")
|
|
assert result is None
|
|
|
|
async def test_empty_response(self, provider: AlphaVantageProvider) -> None:
|
|
provider._client = AsyncMock()
|
|
provider._client.get = AsyncMock(return_value=_mock_av_response({}))
|
|
|
|
result = await provider.fetch("INVALID")
|
|
assert result is None
|
|
|
|
async def test_http_error(self, provider: AlphaVantageProvider) -> None:
|
|
provider._client = AsyncMock()
|
|
provider._client.get = AsyncMock(
|
|
return_value=_mock_av_response({}, status_code=500)
|
|
)
|
|
|
|
result = await provider.fetch("AAPL")
|
|
assert result is None
|
|
|
|
async def test_network_exception(self, provider: AlphaVantageProvider) -> None:
|
|
provider._client = AsyncMock()
|
|
provider._client.get = AsyncMock(side_effect=httpx.ConnectError("timeout"))
|
|
|
|
result = await provider.fetch("AAPL")
|
|
assert result is None
|
|
|
|
async def test_partial_data(self, provider: AlphaVantageProvider) -> None:
|
|
"""Provider should handle responses with only some fields populated."""
|
|
mock_data = {
|
|
"Symbol": "GOOG",
|
|
"EPS": "5.80",
|
|
"PERatio": "None",
|
|
"PEGRatio": "-",
|
|
}
|
|
provider._client = AsyncMock()
|
|
provider._client.get = AsyncMock(return_value=_mock_av_response(mock_data))
|
|
|
|
result = await provider.fetch("GOOG")
|
|
assert result is not None
|
|
assert result.eps_ttm == 5.80
|
|
assert result.pe_ratio is None
|
|
assert result.peg_ratio is None
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# FMP provider tests
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _mock_fmp_response(json_data: list | dict, status_code: int = 200) -> httpx.Response:
|
|
return httpx.Response(
|
|
status_code=status_code,
|
|
json=json_data,
|
|
request=httpx.Request("GET", "https://financialmodelingprep.com/api/v3/key-metrics-ttm/AAPL"),
|
|
)
|
|
|
|
|
|
class TestFMPProvider:
|
|
@pytest.fixture
|
|
def provider(self) -> FMPProvider:
|
|
return FMPProvider(api_key="test-key")
|
|
|
|
async def test_parse_valid_response(self, provider: FMPProvider) -> None:
|
|
mock_data = [
|
|
{
|
|
"netIncomePerShareTTM": 6.57,
|
|
"peRatioTTM": 28.3,
|
|
"pegRatioTTM": 2.1,
|
|
"revenueGrowth": 0.08,
|
|
"netProfitMarginTTM": 0.26,
|
|
"debtToEquityTTM": 1.87,
|
|
"marketCapTTM": 2800000000000,
|
|
}
|
|
]
|
|
provider._client = AsyncMock()
|
|
provider._client.get = AsyncMock(return_value=_mock_fmp_response(mock_data))
|
|
|
|
result = await provider.fetch("AAPL")
|
|
|
|
assert result is not None
|
|
assert result.ticker == "AAPL"
|
|
assert result.eps_ttm == 6.57
|
|
assert result.pe_ratio == 28.3
|
|
assert result.debt_to_equity == 1.87
|
|
assert result.market_cap == 2_800_000_000_000.0
|
|
|
|
async def test_empty_array(self, provider: FMPProvider) -> None:
|
|
provider._client = AsyncMock()
|
|
provider._client.get = AsyncMock(return_value=_mock_fmp_response([]))
|
|
|
|
result = await provider.fetch("INVALID")
|
|
assert result is None
|
|
|
|
async def test_non_list_response(self, provider: FMPProvider) -> None:
|
|
provider._client = AsyncMock()
|
|
provider._client.get = AsyncMock(
|
|
return_value=_mock_fmp_response({"error": "not found"})
|
|
)
|
|
|
|
result = await provider.fetch("INVALID")
|
|
assert result is None
|
|
|
|
async def test_network_error(self, provider: FMPProvider) -> None:
|
|
provider._client = AsyncMock()
|
|
provider._client.get = AsyncMock(side_effect=httpx.ConnectError("refused"))
|
|
|
|
result = await provider.fetch("AAPL")
|
|
assert result is None
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Yahoo Finance provider tests
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestYahooFinanceProvider:
|
|
@pytest.fixture
|
|
def provider(self) -> YahooFinanceProvider:
|
|
return YahooFinanceProvider()
|
|
|
|
async def test_parse_valid_info(self, provider: YahooFinanceProvider) -> None:
|
|
mock_info = {
|
|
"trailingEps": 6.57,
|
|
"trailingPE": 28.3,
|
|
"pegRatio": 2.1,
|
|
"revenueGrowth": 0.08,
|
|
"profitMargins": 0.26,
|
|
"debtToEquity": 187.0,
|
|
"marketCap": 2800000000000,
|
|
}
|
|
with patch(
|
|
"shared.fundamentals.yahoo._fetch_info", return_value=mock_info
|
|
):
|
|
result = await provider.fetch("AAPL")
|
|
|
|
assert result is not None
|
|
assert result.ticker == "AAPL"
|
|
assert result.eps_ttm == 6.57
|
|
assert result.pe_ratio == 28.3
|
|
assert result.peg_ratio == 2.1
|
|
assert result.revenue_growth_yoy == 0.08
|
|
assert result.profit_margin == 0.26
|
|
assert result.debt_to_equity == pytest.approx(1.87)
|
|
assert result.market_cap == 2_800_000_000_000.0
|
|
|
|
async def test_empty_info(self, provider: YahooFinanceProvider) -> None:
|
|
with patch(
|
|
"shared.fundamentals.yahoo._fetch_info", return_value={}
|
|
):
|
|
result = await provider.fetch("INVALID")
|
|
assert result is None
|
|
|
|
async def test_exception(self, provider: YahooFinanceProvider) -> None:
|
|
with patch(
|
|
"shared.fundamentals.yahoo._fetch_info",
|
|
side_effect=Exception("yfinance error"),
|
|
):
|
|
result = await provider.fetch("AAPL")
|
|
assert result is None
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# RotatingProvider tests
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class _StubProvider(FundamentalsProvider):
|
|
"""Test stub that returns a fixed result or raises."""
|
|
|
|
def __init__(
|
|
self,
|
|
result: FundamentalsSnapshot | None = None,
|
|
error: Exception | None = None,
|
|
) -> None:
|
|
self._result = result
|
|
self._error = error
|
|
|
|
async def fetch(self, ticker: str) -> FundamentalsSnapshot | None:
|
|
if self._error:
|
|
raise self._error
|
|
return self._result
|
|
|
|
|
|
class TestRotatingProvider:
|
|
def test_requires_at_least_one_provider(self) -> None:
|
|
with pytest.raises(ValueError, match="at least one"):
|
|
RotatingProvider(providers=[])
|
|
|
|
async def test_returns_first_success(self) -> None:
|
|
snap = FundamentalsSnapshot(ticker="AAPL", eps_ttm=6.0, fetched_at=datetime.now(timezone.utc))
|
|
p1 = _StubProvider(result=snap)
|
|
p2 = _StubProvider(result=FundamentalsSnapshot(ticker="AAPL", eps_ttm=99.0, fetched_at=datetime.now(timezone.utc)))
|
|
|
|
rp = RotatingProvider(providers=[p1, p2])
|
|
result = await rp.fetch("AAPL")
|
|
|
|
assert result is not None
|
|
assert result.eps_ttm == 6.0 # from p1, not p2
|
|
|
|
async def test_falls_back_on_none(self) -> None:
|
|
snap = FundamentalsSnapshot(ticker="AAPL", pe_ratio=30.0, fetched_at=datetime.now(timezone.utc))
|
|
p1 = _StubProvider(result=None)
|
|
p2 = _StubProvider(result=snap)
|
|
|
|
rp = RotatingProvider(providers=[p1, p2])
|
|
result = await rp.fetch("AAPL")
|
|
|
|
assert result is not None
|
|
assert result.pe_ratio == 30.0 # fell through to p2
|
|
|
|
async def test_returns_none_when_all_fail(self) -> None:
|
|
p1 = _StubProvider(result=None)
|
|
p2 = _StubProvider(result=None)
|
|
|
|
rp = RotatingProvider(providers=[p1, p2])
|
|
result = await rp.fetch("AAPL")
|
|
assert result is None
|
|
|
|
async def test_handles_exceptions(self) -> None:
|
|
snap = FundamentalsSnapshot(ticker="AAPL", eps_ttm=5.0, fetched_at=datetime.now(timezone.utc))
|
|
p1 = _StubProvider(error=RuntimeError("boom"))
|
|
p2 = _StubProvider(result=snap)
|
|
|
|
rp = RotatingProvider(providers=[p1, p2])
|
|
result = await rp.fetch("AAPL")
|
|
|
|
assert result is not None
|
|
assert result.eps_ttm == 5.0 # fell through after p1 raised
|
|
|
|
async def test_all_raise(self) -> None:
|
|
p1 = _StubProvider(error=RuntimeError("boom1"))
|
|
p2 = _StubProvider(error=ValueError("boom2"))
|
|
|
|
rp = RotatingProvider(providers=[p1, p2])
|
|
result = await rp.fetch("AAPL")
|
|
assert result is None
|