Codex Release Bot
86 lines
3.1 KiB
Python
86 lines
3.1 KiB
Python
"""Scheduled unset target_path detection."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from datetime import datetime, timezone
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
from server.application.services import unset_path_detect_schedule as sched
|
|
from server.background.schedule_runner import _cron_match
|
|
from server.domain.models import Server
|
|
|
|
|
|
def test_cron_matches_beijing_2am():
|
|
# 2026-06-17 02:00 Beijing = 2026-06-16 18:00 UTC
|
|
at = datetime(2026, 6, 16, 18, 0, tzinfo=timezone.utc)
|
|
assert _cron_match("0 2 * * *", at) is True
|
|
|
|
|
|
def test_cron_not_match_other_hours():
|
|
at = datetime(2026, 6, 16, 19, 0, tzinfo=timezone.utc) # 03:00 Beijing
|
|
assert _cron_match("0 2 * * *", at) is False
|
|
|
|
|
|
def test_server_ssh_configured():
|
|
assert sched.server_ssh_configured(
|
|
Server(name="a", domain="1.1.1.1", auth_method="password", password="x"),
|
|
)
|
|
assert not sched.server_ssh_configured(
|
|
Server(name="b", domain="1.1.1.1", auth_method="key"),
|
|
)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_run_scheduled_skips_when_no_unset_servers(monkeypatch):
|
|
monkeypatch.setattr(sched.settings, "UNSET_PATH_DETECT_ENABLED", "true")
|
|
monkeypatch.setattr(sched, "list_unset_server_ids", AsyncMock(return_value=[]))
|
|
monkeypatch.setattr(sched, "_already_ran_today", AsyncMock(return_value=False))
|
|
monkeypatch.setattr(sched, "_detect_path_batch_running", AsyncMock(return_value=False))
|
|
mark = AsyncMock()
|
|
monkeypatch.setattr(sched, "_mark_ran_today", mark)
|
|
|
|
result = await sched.run_scheduled_unset_path_detect()
|
|
|
|
assert result["action"] == "skipped"
|
|
assert result["reason"] == "no_unset_servers"
|
|
mark.assert_awaited_once()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_run_scheduled_starts_batch_job(monkeypatch):
|
|
monkeypatch.setattr(sched.settings, "UNSET_PATH_DETECT_ENABLED", "true")
|
|
monkeypatch.setattr(sched, "list_unset_server_ids", AsyncMock(return_value=[1, 2]))
|
|
monkeypatch.setattr(sched, "_already_ran_today", AsyncMock(return_value=False))
|
|
monkeypatch.setattr(sched, "_detect_path_batch_running", AsyncMock(return_value=False))
|
|
monkeypatch.setattr(sched, "_mark_ran_today", AsyncMock())
|
|
|
|
start = AsyncMock(return_value={"job_id": 99, "label": "自动检测目标路径", "status": "running"})
|
|
monkeypatch.setattr(
|
|
"server.application.services.server_batch_service.start_batch_job",
|
|
start,
|
|
)
|
|
|
|
result = await sched.run_scheduled_unset_path_detect()
|
|
|
|
assert result["action"] == "started"
|
|
assert result["job_id"] == 99
|
|
assert result["server_count"] == 2
|
|
start.assert_awaited_once_with(
|
|
op="detect-path",
|
|
server_ids=[1, 2],
|
|
operator="system",
|
|
params={"scheduled": True, "cron": "0 2 * * *"},
|
|
)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_run_scheduled_skips_if_already_ran_today(monkeypatch):
|
|
monkeypatch.setattr(sched.settings, "UNSET_PATH_DETECT_ENABLED", "true")
|
|
monkeypatch.setattr(sched, "_already_ran_today", AsyncMock(return_value=True))
|
|
|
|
result = await sched.run_scheduled_unset_path_detect()
|
|
|
|
assert result == {"action": "skipped", "reason": "already_ran_today"}
|