abc06af3a1
- scheduler.py: daemon thread inside Flask refreshes all real Thai collectors on interval (default 60min, REFRESH_INTERVAL_SECONDS) via shared daily cache + writes timestamped marker - create_app starts scheduler (skipped in TESTING); shared daily_cache now an extension - GET /api/v1/data/last-refresh: automation status + last refresh (every N hours) - Live verified: refresh_all pulls 5/5 real sources (tourism/auto/NPL/energy/macro) - 2 tests; full suite 197 OK
52 lines
2.1 KiB
Python
52 lines
2.1 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)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|