03195dc55d
Automatically keep the real dated dividend ledger fresh inside the app's own refresh loop (this app runs on its own server, independent of Hermes): - backend/app/scheduler.py: AppDataScheduler gained a cooldown-gated _maybe_refresh_dated_dividends() that fetches real dated dividend history (siamchart /stock-info) into data/dividends/ledger.json at most once per dividend_cooldown_seconds (default 6h) — dividend history changes only a few times a year, so we never hammer the source every refresh tick. The fetch is non-fatal: a network failure leaves the previous ledger intact. - backend/app/__init__.py: passes DIVIDEND_REFRESH_COOLDOWN_SECONDS to the scheduler (default 21600s). - tests: cooldown fires once then skips, and refetches after it elapses (2) — full backend 294 passed.
101 lines
4.2 KiB
Python
101 lines
4.2 KiB
Python
"""Tests for the in-app data scheduler."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
from unittest.mock import patch
|
|
|
|
from app.scheduler import AppDataScheduler
|
|
|
|
|
|
class _FakeCache:
|
|
def __init__(self): self.calls = {}
|
|
def fetch_or_stale(self, key, fetcher):
|
|
self.calls[key] = self.calls.get(key, 0) + 1
|
|
return fetcher()
|
|
|
|
|
|
class SchedulerTest(unittest.TestCase):
|
|
def test_refresh_all_calls_collectors(self):
|
|
cache = _FakeCache()
|
|
snap_dir = Path(tempfile.mkdtemp())
|
|
sched = AppDataScheduler(cache, snap_dir, interval_seconds=99999)
|
|
# job fetchers import app.* modules; patch their fetch fns to return dicts
|
|
def fake(modname, fn):
|
|
return lambda: {"ok": True}
|
|
with patch("app.auto_credit.fetch_auto_credit", return_value=type("F", (), {"to_dict": lambda self: {"n": 1}})()), \
|
|
patch("app.auto_npl.fetch_auto_npl", return_value=type("F", (), {"to_dict": lambda self: {"n": 2}})()), \
|
|
patch("app.energy_thai.fetch_energy_thai", return_value=type("F", (), {"to_dict": lambda self: {"n": 3}})()), \
|
|
patch("app.macro_thai.fetch_macro_thai", return_value=type("F", (), {"to_dict": lambda self: {"n": 4}})()):
|
|
res = sched.refresh_all()
|
|
# tourism job is bot_tourism.BotTourismSource().fetch — not trivially patched; allow it to fail gracefully
|
|
self.assertGreaterEqual(len(res), 4)
|
|
self.assertTrue(any(r["ok"] for r in res))
|
|
|
|
def test_marker_written(self):
|
|
cache = _FakeCache()
|
|
snap_dir = Path(tempfile.mkdtemp())
|
|
sched = AppDataScheduler(cache, snap_dir, interval_seconds=99999)
|
|
with patch("app.macro_thai.fetch_macro_thai", return_value=type("F", (), {"to_dict": lambda self: {"n": 1}})()):
|
|
sched.refresh_all()
|
|
marker = snap_dir / "scheduler" / "last_refresh.json"
|
|
self.assertTrue(marker.exists())
|
|
import json
|
|
data = json.loads(marker.read_text())
|
|
self.assertIn("at", data)
|
|
|
|
|
|
class DividendCooldownTest(unittest.TestCase):
|
|
def _sched(self, snap_dir, cooldown):
|
|
return AppDataScheduler(_FakeCache(), snap_dir, interval_seconds=99999,
|
|
dividend_cooldown_seconds=cooldown)
|
|
|
|
def test_refreshes_once_then_respects_cooldown(self):
|
|
snap_dir = Path(tempfile.mkdtemp())
|
|
sched = self._sched(snap_dir, cooldown=3600) # 1h cooldown
|
|
calls = {"n": 0}
|
|
|
|
def fake_fetch(sym):
|
|
calls["n"] += 1
|
|
return [("2025-01-01", 1.0)]
|
|
|
|
fake_fv = {"factors": [{"symbol": "PTT"}]}
|
|
with patch("app.siamchart_factors.build_factor_view", return_value=fake_fv), \
|
|
patch("app.siamchart.fetch_dividend_history", side_effect=fake_fetch):
|
|
sched._maybe_refresh_dated_dividends() # first -> fetch
|
|
n_after_first = calls["n"]
|
|
sched._maybe_refresh_dated_dividends() # second, within cooldown -> skip
|
|
self.assertEqual(n_after_first, 1) # fetched once
|
|
self.assertEqual(calls["n"], 1) # second call skipped
|
|
self.assertTrue((snap_dir / "dividends" / "ledger.json").exists())
|
|
|
|
def test_refetches_after_cooldown_elapses(self):
|
|
snap_dir = Path(tempfile.mkdtemp())
|
|
sched = self._sched(snap_dir, cooldown=1) # 1s cooldown
|
|
calls = {"n": 0}
|
|
|
|
def fake_fetch(sym):
|
|
calls["n"] += 1
|
|
return [("2025-01-01", 1.0)]
|
|
|
|
fake_fv = {"factors": [{"symbol": "PTT"}]}
|
|
with patch("app.siamchart_factors.build_factor_view", return_value=fake_fv), \
|
|
patch("app.siamchart.fetch_dividend_history", side_effect=fake_fetch):
|
|
sched._maybe_refresh_dated_dividends()
|
|
self.assertEqual(calls["n"], 1)
|
|
# force cooldown to have elapsed by rewriting the marker back in time
|
|
import json
|
|
import datetime as _dt
|
|
sched._div_marker.write_text(
|
|
json.dumps({"at": (_dt.datetime.now().astimezone() - _dt.timedelta(hours=1)).isoformat()}),
|
|
encoding="utf-8",
|
|
)
|
|
sched._maybe_refresh_dated_dividends()
|
|
self.assertEqual(calls["n"], 2) # refetched
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|