bu5hm4nn/vault-dash
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(CORE-001D3B): validate alert history entry types

Browse Source
This commit is contained in:
Bu5hm4nn
2026-03-26 15:19:42 +01:00
parent 65da5b8f1d
commit 99d22302ee
2 changed files with 40 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

17
app/models/alerts.py
View File

@@ -3,6 +3,7 @@
from __future__ import annotations from __future__ import annotations
import json import json
import math
from dataclasses import asdict, dataclass from dataclasses import asdict, dataclass
from pathlib import Path from pathlib import Path
from typing import Any from typing import Any
@@ -25,6 +26,22 @@ class AlertEvent:
updated_at: str updated_at: str
email_alerts_enabled: bool email_alerts_enabled: bool
def __post_init__(self) -> None:
for field_name in ("severity", "message", "updated_at"):
value = getattr(self, field_name)
if not isinstance(value, str):
raise TypeError(f"{field_name} must be a string")
for field_name in ("ltv_ratio", "warning_threshold", "critical_threshold", "spot_price"):
value = getattr(self, field_name)
if isinstance(value, bool) or not isinstance(value, (int, float)):
raise TypeError(f"{field_name} must be numeric")
numeric_value = float(value)
if not math.isfinite(numeric_value):
raise ValueError(f"{field_name} must be finite")
setattr(self, field_name, numeric_value)
if not isinstance(self.email_alerts_enabled, bool):
raise TypeError("email_alerts_enabled must be a bool")
def to_dict(self) -> dict[str, Any]: def to_dict(self) -> dict[str, Any]:
return asdict(self) return asdict(self)

23
tests/test_alerts.py
View File

@@ -188,6 +188,15 @@ def test_alert_history_repository_raises_on_non_list_payload(tmp_path: Path) ->
repository.load() repository.load()
def test_alert_history_repository_raises_on_non_object_entry(tmp_path: Path) -> None:
history_path = tmp_path / "alert_history.json"
history_path.write_text('["bad entry"]', encoding="utf-8")
repository = AlertHistoryRepository(history_path=history_path)
with pytest.raises(AlertHistoryLoadError, match="Alert history entry 0 must be an object"):
repository.load()
def test_alert_history_repository_raises_on_invalid_list_entry(tmp_path: Path) -> None: def test_alert_history_repository_raises_on_invalid_list_entry(tmp_path: Path) -> None:
history_path = tmp_path / "alert_history.json" history_path = tmp_path / "alert_history.json"
history_path.write_text('[{"severity": "warning"}]', encoding="utf-8") history_path.write_text('[{"severity": "warning"}]', encoding="utf-8")
@@ -197,6 +206,20 @@ def test_alert_history_repository_raises_on_invalid_list_entry(tmp_path: Path) -
repository.load() repository.load()
def test_alert_history_repository_raises_on_wrong_typed_fields(tmp_path: Path) -> None:
history_path = tmp_path / "alert_history.json"
history_path.write_text(
'[{"severity": "warning", "message": "bad", "ltv_ratio": "oops", "warning_threshold": 0.7, '
'"critical_threshold": 0.75, "spot_price": 215.0, "updated_at": "2026-03-24T12:00:00Z", '
'"email_alerts_enabled": false}]',
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: 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") alert_service.repository.history_path.write_text("{not valid json", encoding="utf-8")
config = PortfolioConfig( config = PortfolioConfig(