Codex
33 lines
1.1 KiB
Python
33 lines
1.1 KiB
Python
import contextlib
|
|
import io
|
|
import json
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
from scripts import run_runtime_task_watchdog
|
|
|
|
|
|
class RunRuntimeTaskWatchdogScriptTests(unittest.TestCase):
|
|
def test_main_outputs_json(self) -> None:
|
|
with patch("scripts.run_runtime_task_watchdog.get_trading_runtime_watchdog") as watchdog_mock:
|
|
watchdog_mock.return_value = {
|
|
"checked_ts": "2026-05-12T00:00:00+00:00",
|
|
"stalled_task_count": 1,
|
|
"stalled_task_names": ["watch"],
|
|
"stalled_tasks": [{"name": "watch", "reason": "overdue"}],
|
|
}
|
|
output = io.StringIO()
|
|
with contextlib.redirect_stdout(output):
|
|
exit_code = run_runtime_task_watchdog.main(
|
|
env={"TRADING_RUNTIME_WATCHDOG_OUTPUT_JSON": "1", "TRADING_RUNTIME_WATCHDOG_RUN_ONCE": "0"}
|
|
)
|
|
|
|
payload = json.loads(output.getvalue())
|
|
self.assertEqual(exit_code, 2)
|
|
self.assertEqual(payload["stalled_task_count"], 1)
|
|
watchdog_mock.assert_called_once_with(run_once=False)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|