89 lines
2.9 KiB
Python
89 lines
2.9 KiB
Python
"""Tests for the BOXIM takeover scheduler lifecycle."""
|
|
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
|
|
|
|
def test_app_has_startup_and_shutdown_events():
|
|
from main import app
|
|
|
|
assert app.router.on_startup
|
|
assert app.router.on_shutdown
|
|
|
|
|
|
@patch("services.takeover_service.TakeoverService")
|
|
@patch("services.boxim_client.BoxIMClient")
|
|
@patch("main.AsyncIOScheduler")
|
|
def test_scheduler_uses_boxim_and_restart_safe_service(
|
|
mock_scheduler_class,
|
|
mock_boxim_class,
|
|
mock_takeover_class,
|
|
):
|
|
import main
|
|
|
|
scheduler = MagicMock()
|
|
mock_scheduler_class.return_value = scheduler
|
|
boxim = MagicMock()
|
|
mock_boxim_class.return_value = boxim
|
|
takeover = MagicMock()
|
|
takeover.poll_messages = AsyncMock()
|
|
takeover.process_reply_tasks = AsyncMock()
|
|
mock_takeover_class.return_value = takeover
|
|
|
|
environment = {
|
|
"HUIHUI_PLATFORM_BASE_URL": "https://open.example/api",
|
|
"BOXIM_API_BASE_URL": "https://im.example/api",
|
|
"HUIHUI_APP_ID": "app-id",
|
|
"HUIHUI_ACCESS_ID": "access-id",
|
|
"HUIHUI_ACCESS_SECRET": "secret",
|
|
"BOXIM_POLL_INTERVAL_SECONDS": "1",
|
|
}
|
|
with patch("main.init_db"), patch("main.seed"), patch.dict(
|
|
"os.environ", environment, clear=False
|
|
):
|
|
main.on_startup()
|
|
|
|
config = mock_boxim_class.call_args.args[0]
|
|
assert config["HUIHUI_PLATFORM_BASE_URL"] == "https://open.example/api"
|
|
assert config["BOXIM_API_BASE_URL"] == "https://im.example/api"
|
|
mock_takeover_class.assert_called_once_with(main.SessionLocal, boxim)
|
|
|
|
assert scheduler.add_job.call_count == 2
|
|
poll_call, process_call = scheduler.add_job.call_args_list
|
|
assert poll_call.args[0] is takeover.poll_messages
|
|
assert poll_call.kwargs["id"] == "takeover_message_poll"
|
|
assert poll_call.kwargs["trigger"].interval.total_seconds() == 1
|
|
assert poll_call.kwargs["max_instances"] == 1
|
|
assert poll_call.kwargs["coalesce"] is True
|
|
assert process_call.args[0] is takeover.process_reply_tasks
|
|
assert process_call.kwargs["id"] == "takeover_reply_process"
|
|
assert process_call.kwargs["trigger"].interval.total_seconds() == 0.5
|
|
assert process_call.kwargs["max_instances"] == 1
|
|
assert process_call.kwargs["coalesce"] is True
|
|
scheduler.start.assert_called_once_with()
|
|
|
|
main.takeover_scheduler = None
|
|
|
|
|
|
@patch("main.AsyncIOScheduler")
|
|
def test_scheduler_failure_does_not_stop_the_api(mock_scheduler_class):
|
|
import main
|
|
|
|
mock_scheduler_class.side_effect = RuntimeError("scheduler crash")
|
|
with patch("main.init_db"), patch("main.seed"):
|
|
main.on_startup()
|
|
|
|
assert main.takeover_scheduler is None
|
|
|
|
|
|
def test_shutdown_stops_only_the_scheduler():
|
|
import main
|
|
|
|
scheduler = MagicMock()
|
|
scheduler.running = True
|
|
main.takeover_scheduler = scheduler
|
|
|
|
main.on_shutdown()
|
|
|
|
scheduler.shutdown.assert_called_once_with(wait=False)
|
|
assert main.takeover_scheduler is None
|