Codex Release Bot
77 lines
2.8 KiB
Python
77 lines
2.8 KiB
Python
"""Scheduled IP-only server site domain detection."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from datetime import datetime, timezone
|
|
from unittest.mock import AsyncMock
|
|
|
|
import pytest
|
|
|
|
from server.application.services import ip_domain_detect_schedule as sched
|
|
from server.background.schedule_runner import _cron_match
|
|
from server.domain.models import Server
|
|
|
|
|
|
def test_cron_matches_beijing_2330():
|
|
# 2026-06-17 23:30 Beijing = 2026-06-17 15:30 UTC
|
|
at = datetime(2026, 6, 17, 15, 30, tzinfo=timezone.utc)
|
|
assert _cron_match("30 23 * * *", at) is True
|
|
|
|
|
|
def test_cron_not_match_other_minutes():
|
|
at = datetime(2026, 6, 17, 15, 29, tzinfo=timezone.utc)
|
|
assert _cron_match("30 23 * * *", at) is False
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_run_scheduled_skips_when_no_servers(monkeypatch):
|
|
monkeypatch.setattr(sched.settings, "IP_DOMAIN_DETECT_ENABLED", "true")
|
|
monkeypatch.setattr(sched, "list_ip_only_without_site_server_ids", AsyncMock(return_value=[]))
|
|
monkeypatch.setattr(sched, "_already_ran_today", AsyncMock(return_value=False))
|
|
monkeypatch.setattr(sched, "_detect_domain_batch_running", AsyncMock(return_value=False))
|
|
mark = AsyncMock()
|
|
monkeypatch.setattr(sched, "_mark_ran_today", mark)
|
|
|
|
result = await sched.run_scheduled_ip_domain_detect()
|
|
|
|
assert result["action"] == "skipped"
|
|
assert result["reason"] == "no_ip_only_servers"
|
|
mark.assert_awaited_once()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_run_scheduled_starts_batch_job(monkeypatch):
|
|
monkeypatch.setattr(sched.settings, "IP_DOMAIN_DETECT_ENABLED", "true")
|
|
monkeypatch.setattr(sched, "list_ip_only_without_site_server_ids", AsyncMock(return_value=[3, 4]))
|
|
monkeypatch.setattr(sched, "_already_ran_today", AsyncMock(return_value=False))
|
|
monkeypatch.setattr(sched, "_detect_domain_batch_running", AsyncMock(return_value=False))
|
|
monkeypatch.setattr(sched, "_mark_ran_today", AsyncMock())
|
|
|
|
start = AsyncMock(return_value={"job_id": 88, "label": "自动检测站点域名", "status": "running"})
|
|
monkeypatch.setattr(
|
|
"server.application.services.server_batch_service.start_batch_job",
|
|
start,
|
|
)
|
|
|
|
result = await sched.run_scheduled_ip_domain_detect()
|
|
|
|
assert result["action"] == "started"
|
|
assert result["job_id"] == 88
|
|
assert result["server_count"] == 2
|
|
start.assert_awaited_once_with(
|
|
op="detect-domain",
|
|
server_ids=[3, 4],
|
|
operator="system",
|
|
params={"scheduled": True, "cron": "30 23 * * *"},
|
|
)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_run_scheduled_skips_if_already_ran_today(monkeypatch):
|
|
monkeypatch.setattr(sched.settings, "IP_DOMAIN_DETECT_ENABLED", "true")
|
|
monkeypatch.setattr(sched, "_already_ran_today", AsyncMock(return_value=True))
|
|
|
|
result = await sched.run_scheduled_ip_domain_detect()
|
|
|
|
assert result == {"action": "skipped", "reason": "already_ran_today"}
|