Codex
72 lines
2.9 KiB
Python
72 lines
2.9 KiB
Python
import os
|
|
import unittest
|
|
from datetime import datetime, timezone
|
|
|
|
os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://user:pass@localhost:5432/ai_ict_test")
|
|
|
|
from src.services.market_state.exchange_status_provider import OkxAnnouncementsProvider, OkxSystemStatusProvider
|
|
|
|
|
|
class ExchangeStatusProviderTests(unittest.TestCase):
|
|
def test_system_status_provider_uses_cached_snapshot_when_live_fetch_fails(self) -> None:
|
|
repository = _SnapshotRepo(
|
|
{
|
|
"provider_name": "okx_system_status",
|
|
"fetched_ts": "2026-05-14T02:00:00+00:00",
|
|
"event_count": 1,
|
|
"code": "active",
|
|
"active_events": [{"title": "API maintenance"}],
|
|
"raw_payload": {"data": [{"title": "API maintenance"}]},
|
|
}
|
|
)
|
|
provider = OkxSystemStatusProvider(repository=repository, cache_max_age_seconds=21600)
|
|
provider._fetch_payload = lambda: (_ for _ in ()).throw(RuntimeError("ssl eof")) # type: ignore[attr-defined]
|
|
|
|
snapshot = provider.fetch_snapshot(now=datetime(2026, 5, 14, 2, 30, tzinfo=timezone.utc))
|
|
|
|
self.assertEqual(snapshot["status"], "active")
|
|
self.assertEqual(snapshot["freshness_code"], "provider_cached")
|
|
self.assertEqual(snapshot["cache_state"], "cached_fallback")
|
|
self.assertIsNone(snapshot["error"])
|
|
self.assertIn("ssl eof", snapshot["warning"])
|
|
|
|
def test_announcements_provider_uses_cached_snapshot_when_live_fetch_fails(self) -> None:
|
|
repository = _SnapshotRepo(
|
|
{
|
|
"provider_name": "okx_announcements",
|
|
"fetched_ts": "2026-05-14T02:00:00+00:00",
|
|
"event_count": 1,
|
|
"code": "active",
|
|
"active_events": [{"title": "Listing schedule update"}],
|
|
"raw_payload": {"data": [{"details": [{"title": "Listing schedule update"}]}]},
|
|
}
|
|
)
|
|
provider = OkxAnnouncementsProvider(repository=repository, cache_max_age_seconds=21600)
|
|
provider._fetch_payload = lambda: (_ for _ in ()).throw(RuntimeError("ssl eof")) # type: ignore[attr-defined]
|
|
|
|
snapshot = provider.fetch_snapshot(now=datetime(2026, 5, 14, 2, 30, tzinfo=timezone.utc))
|
|
|
|
self.assertEqual(snapshot["status"], "active")
|
|
self.assertEqual(snapshot["freshness_code"], "provider_cached")
|
|
self.assertEqual(snapshot["cache_state"], "cached_fallback")
|
|
self.assertIsNone(snapshot["error"])
|
|
self.assertIn("ssl eof", snapshot["warning"])
|
|
|
|
|
|
class _SnapshotRepo:
|
|
def __init__(self, row: dict) -> None:
|
|
self.row = row
|
|
|
|
def fetch_news_provider_snapshot(self, provider_name: str):
|
|
if provider_name == self.row.get("provider_name"):
|
|
return dict(self.row)
|
|
return None
|
|
|
|
def upsert_news_provider_snapshot(self, **kwargs):
|
|
self.row = dict(kwargs)
|
|
return 1
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|