137 lines
4.1 KiB
Python
137 lines
4.1 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import unittest
|
|
from types import SimpleNamespace
|
|
|
|
from backend.features.system.health import HealthServiceMixin
|
|
from backend.features.system.routes import SystemRoutesMixin
|
|
|
|
|
|
class _Connection:
|
|
def __enter__(self):
|
|
return self
|
|
|
|
def __exit__(self, *_args):
|
|
return None
|
|
|
|
def execute(self, _query: str):
|
|
return self
|
|
|
|
def fetchone(self):
|
|
return (1,)
|
|
|
|
|
|
class _Database:
|
|
def connect(self):
|
|
return _Connection()
|
|
|
|
def status(self):
|
|
return {
|
|
"database": "secret-review.db",
|
|
"snapshot_dates": 31,
|
|
"updated_at": "2026-08-07T11:34:55+08:00",
|
|
"last_sync": {
|
|
"id": 991,
|
|
"trade_date": "20260807",
|
|
"source": "tushare",
|
|
"status": "success",
|
|
"finished_at": "2026-08-07T11:34:55+08:00",
|
|
"message": "https://provider.invalid?token=secret",
|
|
},
|
|
}
|
|
|
|
|
|
class _Service(HealthServiceMixin):
|
|
def __init__(self):
|
|
self.database = _Database()
|
|
self.jobs = SimpleNamespace(
|
|
repository=SimpleNamespace(
|
|
recent=lambda _limit: [
|
|
{
|
|
"id": 77,
|
|
"status": "success",
|
|
"finished_at": "2026-08-07T11:34:55+08:00",
|
|
"message": "secret upstream response",
|
|
}
|
|
]
|
|
)
|
|
)
|
|
self._system_credentials = {
|
|
"tushare_token": "secret-token",
|
|
"primary_model_id": "primary-id",
|
|
"fallback_model_id": "fallback-id",
|
|
}
|
|
|
|
@property
|
|
def configured(self):
|
|
return bool(self._system_credentials.get("tushare_token"))
|
|
|
|
def _platform_llm_profile(self):
|
|
return {
|
|
"primary": {
|
|
"api_key": "secret-primary-key",
|
|
"base_url": "https://models.invalid/v1",
|
|
"model": "secret-primary-model",
|
|
},
|
|
"fallback": {
|
|
"api_key": "secret-fallback-key",
|
|
"base_url": "https://models.invalid/v1",
|
|
"model": "secret-fallback-model",
|
|
},
|
|
}
|
|
|
|
@staticmethod
|
|
def _profile_configured(profile):
|
|
return all(profile.get(key) for key in ("api_key", "base_url", "model"))
|
|
|
|
|
|
class SystemHealthTests(unittest.TestCase):
|
|
def test_public_route_uses_the_service_health_contract(self):
|
|
handler = SystemRoutesMixin()
|
|
expected = {"ok": True, "components": {"process": {"status": "ready"}}}
|
|
handler.application_service = SimpleNamespace(health_status=lambda: expected)
|
|
responses = []
|
|
handler.send_json = responses.append
|
|
|
|
handled = handler._handle_system_public_get(SimpleNamespace(path="/api/health"))
|
|
|
|
self.assertTrue(handled)
|
|
self.assertEqual(responses, [expected])
|
|
|
|
def test_health_distinguishes_runtime_components(self):
|
|
result = _Service().health_status()
|
|
|
|
self.assertTrue(result["ok"])
|
|
self.assertEqual(result["status"], "ready")
|
|
self.assertEqual(
|
|
set(result["components"]),
|
|
{"process", "database", "data_sources", "jobs", "models"},
|
|
)
|
|
self.assertEqual(result["components"]["data_sources"]["latest_trade_date"], "20260807")
|
|
self.assertEqual(result["components"]["models"]["primary"], "ready")
|
|
self.assertEqual(result["components"]["models"]["fallback"], "ready")
|
|
|
|
def test_public_health_never_exposes_engineering_or_secret_fields(self):
|
|
payload = json.dumps(_Service().health_status(), ensure_ascii=False).lower()
|
|
|
|
for forbidden in (
|
|
"tushare",
|
|
"ifind",
|
|
"secret",
|
|
"token",
|
|
"https://",
|
|
".db",
|
|
"primary-id",
|
|
"fallback-id",
|
|
"model",
|
|
):
|
|
if forbidden == "model":
|
|
self.assertNotIn("secret-primary-model", payload)
|
|
else:
|
|
self.assertNotIn(forbidden, payload)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|