290 lines
10 KiB
Python
290 lines
10 KiB
Python
from __future__ import annotations
|
|
|
|
from decimal import Decimal
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from app.models.alerts import AlertHistoryLoadError, AlertHistoryRepository
|
|
from app.models.portfolio import PortfolioConfig
|
|
from app.services.alerts import AlertService, _normalize_alert_evaluation_input, build_portfolio_alert_context
|
|
|
|
|
|
@pytest.fixture
|
|
def alert_service(tmp_path: Path) -> AlertService:
|
|
return AlertService(history_path=tmp_path / "alert_history.json")
|
|
|
|
|
|
def test_normalize_alert_evaluation_input_accepts_decimal_boundary_values() -> None:
|
|
config = PortfolioConfig(gold_value=215_000.0, entry_price=215.0, loan_amount=151_000.0)
|
|
|
|
normalized = _normalize_alert_evaluation_input(
|
|
config,
|
|
{
|
|
"ltv_ratio": Decimal("0.7023"),
|
|
"spot_price": Decimal("215.0"),
|
|
"quote_updated_at": 123,
|
|
},
|
|
)
|
|
|
|
assert normalized.ltv_ratio == Decimal("0.7023")
|
|
assert normalized.spot_price == Decimal("215.0")
|
|
assert normalized.updated_at == "123"
|
|
assert normalized.warning_threshold == Decimal("0.7")
|
|
assert normalized.critical_threshold == Decimal("0.75")
|
|
|
|
|
|
def test_alert_service_evaluate_accepts_string_boundary_values(alert_service: AlertService) -> None:
|
|
config = PortfolioConfig(gold_value=215_000.0, entry_price=215.0, loan_amount=151_000.0)
|
|
|
|
status = alert_service.evaluate(
|
|
config,
|
|
{
|
|
"ltv_ratio": "0.7023",
|
|
"spot_price": "215.0",
|
|
"quote_updated_at": "2026-03-24T12:00:00Z",
|
|
},
|
|
persist=False,
|
|
)
|
|
|
|
assert status.severity == "warning"
|
|
assert status.ltv_ratio == pytest.approx(0.7023, rel=1e-9)
|
|
assert [event.severity for event in status.history] == ["warning"]
|
|
|
|
|
|
def test_alert_service_rejects_missing_ltv_ratio(alert_service: AlertService) -> None:
|
|
config = PortfolioConfig(gold_value=215_000.0, entry_price=215.0, loan_amount=151_000.0)
|
|
|
|
with pytest.raises(ValueError, match="portfolio.ltv_ratio must be present"):
|
|
alert_service.evaluate(
|
|
config,
|
|
{
|
|
"spot_price": 215.0,
|
|
"quote_updated_at": "2026-03-24T12:00:00Z",
|
|
},
|
|
persist=False,
|
|
)
|
|
|
|
|
|
def test_normalize_alert_evaluation_input_rejects_bool_values() -> None:
|
|
config = PortfolioConfig(gold_value=215_000.0, entry_price=215.0, loan_amount=151_000.0)
|
|
|
|
with pytest.raises(TypeError, match="portfolio.ltv_ratio must be numeric, got bool"):
|
|
_normalize_alert_evaluation_input(
|
|
config,
|
|
{
|
|
"ltv_ratio": True,
|
|
"spot_price": 215.0,
|
|
"quote_updated_at": "2026-03-24T12:00:00Z",
|
|
},
|
|
)
|
|
|
|
|
|
def test_alert_service_reports_ok_when_ltv_is_below_warning(alert_service: AlertService) -> None:
|
|
config = PortfolioConfig(gold_value=215_000.0, entry_price=215.0, loan_amount=120_000.0)
|
|
portfolio = build_portfolio_alert_context(
|
|
config, spot_price=215.0, source="test", updated_at="2026-03-24T12:00:00Z"
|
|
)
|
|
|
|
status = alert_service.evaluate(config, portfolio)
|
|
|
|
assert status.severity == "ok"
|
|
assert status.message == "LTV is within configured thresholds."
|
|
assert status.history == []
|
|
|
|
|
|
def test_alert_service_preview_does_not_persist_history(alert_service: AlertService) -> None:
|
|
config = PortfolioConfig(
|
|
gold_value=215_000.0,
|
|
entry_price=215.0,
|
|
loan_amount=151_000.0,
|
|
ltv_warning=0.70,
|
|
margin_threshold=0.75,
|
|
)
|
|
portfolio = build_portfolio_alert_context(config, spot_price=215.0, source="settings-preview", updated_at="")
|
|
|
|
preview_status = alert_service.evaluate(config, portfolio, persist=False)
|
|
|
|
assert preview_status.severity == "warning"
|
|
assert [event.severity for event in preview_status.history] == ["warning"]
|
|
assert alert_service.repository.load() == []
|
|
|
|
persisted_status = alert_service.evaluate(config, portfolio)
|
|
|
|
assert [event.severity for event in persisted_status.history] == ["warning"]
|
|
assert len(alert_service.repository.load()) == 1
|
|
|
|
|
|
def test_alert_service_logs_warning_once_when_warning_threshold_is_crossed(alert_service: AlertService) -> None:
|
|
config = PortfolioConfig(
|
|
gold_value=215_000.0,
|
|
entry_price=215.0,
|
|
loan_amount=151_000.0,
|
|
ltv_warning=0.70,
|
|
margin_threshold=0.75,
|
|
email_alerts=True,
|
|
)
|
|
portfolio = build_portfolio_alert_context(
|
|
config, spot_price=215.0, source="test", updated_at="2026-03-24T12:00:00Z"
|
|
)
|
|
|
|
first_status = alert_service.evaluate(config, portfolio)
|
|
second_status = alert_service.evaluate(config, portfolio)
|
|
|
|
assert first_status.severity == "warning"
|
|
assert first_status.email_alerts_enabled is True
|
|
assert len(first_status.history) == 1
|
|
assert first_status.history[0].severity == "warning"
|
|
assert "70.2%" in first_status.history[0].message
|
|
assert len(second_status.history) == 1
|
|
|
|
|
|
def test_alert_service_escalates_to_critical_and_keeps_history(alert_service: AlertService) -> None:
|
|
config = PortfolioConfig(
|
|
gold_value=215_000.0,
|
|
entry_price=215.0,
|
|
loan_amount=150_500.0,
|
|
ltv_warning=0.70,
|
|
margin_threshold=0.75,
|
|
)
|
|
|
|
warning_portfolio = build_portfolio_alert_context(
|
|
config,
|
|
spot_price=215.0,
|
|
source="test",
|
|
updated_at="2026-03-24T12:00:00Z",
|
|
)
|
|
critical_portfolio = build_portfolio_alert_context(
|
|
config,
|
|
spot_price=200.0,
|
|
source="test",
|
|
updated_at="2026-03-24T13:00:00Z",
|
|
)
|
|
|
|
alert_service.evaluate(config, warning_portfolio)
|
|
critical_status = alert_service.evaluate(config, critical_portfolio)
|
|
|
|
assert critical_status.severity == "critical"
|
|
assert "above the critical threshold" in critical_status.message
|
|
assert [event.severity for event in critical_status.history] == ["critical", "warning"]
|
|
assert critical_status.history[0].ltv_ratio == pytest.approx(0.7525, rel=1e-6)
|
|
|
|
|
|
def test_alert_history_repository_raises_on_corrupt_history_file(tmp_path: Path) -> None:
|
|
history_path = tmp_path / "alert_history.json"
|
|
history_path.write_text("{not valid json", encoding="utf-8")
|
|
repository = AlertHistoryRepository(history_path=history_path)
|
|
|
|
with pytest.raises(AlertHistoryLoadError, match="Alert history is not valid JSON"):
|
|
repository.load()
|
|
|
|
|
|
def test_alert_history_repository_raises_on_non_list_payload(tmp_path: Path) -> None:
|
|
history_path = tmp_path / "alert_history.json"
|
|
history_path.write_text('{"severity": "warning"}', encoding="utf-8")
|
|
repository = AlertHistoryRepository(history_path=history_path)
|
|
|
|
with pytest.raises(AlertHistoryLoadError, match="Alert history payload must be a list"):
|
|
repository.load()
|
|
|
|
|
|
def test_alert_history_repository_raises_on_invalid_list_entry(tmp_path: Path) -> None:
|
|
history_path = tmp_path / "alert_history.json"
|
|
history_path.write_text('[{"severity": "warning"}]', encoding="utf-8")
|
|
repository = AlertHistoryRepository(history_path=history_path)
|
|
|
|
with pytest.raises(AlertHistoryLoadError, match="Alert history entry 0 is invalid"):
|
|
repository.load()
|
|
|
|
|
|
def test_alert_service_marks_history_unavailable_on_corrupt_storage(alert_service: AlertService) -> None:
|
|
alert_service.repository.history_path.write_text("{not valid json", encoding="utf-8")
|
|
config = PortfolioConfig(
|
|
gold_value=215_000.0,
|
|
entry_price=215.0,
|
|
loan_amount=151_000.0,
|
|
ltv_warning=0.70,
|
|
margin_threshold=0.75,
|
|
)
|
|
portfolio = build_portfolio_alert_context(
|
|
config,
|
|
spot_price=215.0,
|
|
source="test",
|
|
updated_at="2026-03-24T12:00:00Z",
|
|
)
|
|
|
|
status = alert_service.evaluate(config, portfolio)
|
|
|
|
assert status.severity == "warning"
|
|
assert status.history == []
|
|
assert status.history_unavailable is True
|
|
assert (
|
|
status.history_notice
|
|
== "Alert history is temporarily unavailable due to a storage error. New alerts are not being recorded."
|
|
)
|
|
assert alert_service.repository.history_path.read_text(encoding="utf-8") == "{not valid json"
|
|
|
|
|
|
def test_alert_service_preview_marks_history_unavailable_on_corrupt_storage(alert_service: AlertService) -> None:
|
|
alert_service.repository.history_path.write_text('[{"severity": "warning"}]', encoding="utf-8")
|
|
config = PortfolioConfig(
|
|
gold_value=215_000.0,
|
|
entry_price=215.0,
|
|
loan_amount=151_000.0,
|
|
ltv_warning=0.70,
|
|
margin_threshold=0.75,
|
|
)
|
|
portfolio = build_portfolio_alert_context(
|
|
config,
|
|
spot_price=215.0,
|
|
source="settings-preview",
|
|
updated_at="",
|
|
)
|
|
|
|
status = alert_service.evaluate(config, portfolio, persist=False)
|
|
|
|
assert status.severity == "warning"
|
|
assert [event.severity for event in status.history] == ["warning"]
|
|
assert status.history_unavailable is True
|
|
assert (
|
|
status.history_notice
|
|
== "Alert history is temporarily unavailable due to a storage error. New alerts are not being recorded."
|
|
)
|
|
|
|
|
|
def test_alert_service_preserves_persisted_history_during_ok_evaluation(alert_service: AlertService) -> None:
|
|
warning_config = PortfolioConfig(
|
|
gold_value=215_000.0,
|
|
entry_price=215.0,
|
|
loan_amount=151_000.0,
|
|
ltv_warning=0.70,
|
|
margin_threshold=0.75,
|
|
)
|
|
ok_config = PortfolioConfig(
|
|
gold_value=215_000.0,
|
|
entry_price=215.0,
|
|
loan_amount=120_000.0,
|
|
ltv_warning=0.70,
|
|
margin_threshold=0.75,
|
|
)
|
|
|
|
warning_portfolio = build_portfolio_alert_context(
|
|
warning_config,
|
|
spot_price=215.0,
|
|
source="test",
|
|
updated_at="2026-03-24T12:00:00Z",
|
|
)
|
|
ok_portfolio = build_portfolio_alert_context(
|
|
ok_config,
|
|
spot_price=215.0,
|
|
source="test",
|
|
updated_at="2026-03-24T13:00:00Z",
|
|
)
|
|
|
|
alert_service.evaluate(warning_config, warning_portfolio)
|
|
ok_status = alert_service.evaluate(ok_config, ok_portfolio)
|
|
|
|
assert ok_status.severity == "ok"
|
|
assert [event.severity for event in ok_status.history] == ["warning"]
|
|
assert len(alert_service.repository.load()) == 1
|