ForcePilot/backend/test/unit/external_systems/infrastructure/test_scheduler.py

"""external_systems scheduler handler 单元测试。

infrastructure 包 ``__init__.py`` 在本地非容器环境导入 ``container.py`` 时会失败，
因此本测试通过 ``importlib`` 直接加载 ``scheduler.py`` 模块，避免触发包级装配代码。
"""

from __future__ import annotations

import importlib.util
import sys
import types
from datetime import timedelta
from pathlib import Path
from typing import Any
from unittest.mock import AsyncMock, MagicMock

import pytest
from yuxi.scheduler.core.contracts import TaskContext, TaskResult
from yuxi.scheduler.framework.runtime import HandlerRegistry
from yuxi.utils.datetime_utils import utc_now_naive


def _load_scheduler_module() -> Any:
    """直接加载 scheduler.py 模块，绕过 infrastructure 包 __init__.py。"""
    module_path = Path(__file__).parents[5] / "backend/package/yuxi/external_systems/infrastructure/scheduler.py"
    spec = importlib.util.spec_from_file_location(
        "external_systems_scheduler",
        module_path.resolve(),
    )
    assert spec is not None and spec.loader is not None
    module = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(module)
    return module


def _make_task_context(payload: dict[str, Any] | None = None) -> TaskContext:
    return TaskContext(
        task_id="task-1",
        run_id="run-1",
        handler_name="external_systems.webhook_event_cleanup",
        payload=payload or {},
        triggered_by="scheduler",
        scheduled_at=utc_now_naive(),
    )


def _make_repos(
    deleted_count: int = 0,
    near_critical: list[Any] | None = None,
    near_warning: list[Any] | None = None,
    expiring_quotas: list[Any] | None = None,
    expiring_subscriptions: list[Any] | None = None,
    audit_log_deleted_count: int = 0,
    system_ids: list[int] | None = None,
) -> MagicMock:
    repos = MagicMock()
    repos.webhook_event = AsyncMock()
    repos.webhook_event.delete_old_events = AsyncMock(return_value=deleted_count)
    repos.webhook_subscription = AsyncMock()
    repos.webhook_subscription.list_expiring = AsyncMock(return_value=expiring_subscriptions or [])
    repos.webhook_subscription.update_renewal_result = AsyncMock()
    repos.quota_usage = AsyncMock()
    repos.quota_usage.list_near_critical = AsyncMock(return_value=near_critical or [])
    repos.quota_usage.list_near_warning = AsyncMock(return_value=near_warning or [])
    repos.quota_usage.list = AsyncMock(return_value=expiring_quotas or [])
    repos.quota_usage.reset_window = AsyncMock()
    repos.audit_log = AsyncMock()
    repos.audit_log.delete_old_logs = AsyncMock(return_value=audit_log_deleted_count)
    repos.system = AsyncMock()
    repos.system.list_ids = AsyncMock(return_value=system_ids or [])
    repos.alert = AsyncMock()
    return repos


class _QuotaStub:
    """配额记录存根，用于 quota handler 测试。"""

    def __init__(
        self,
        quota_id: int | None = None,
        system_id: int = 1,
        env_key: str = "default",
        quota_key: str = "api_calls",
        quota_name: str = "API Calls",
        quota_window: str = "daily",
        limit_value: int = 1000,
        used_value: int = 0,
        usage_ratio: float = 0.0,
        warning_threshold: float = 80.0,
        critical_threshold: float = 95.0,
        window_start: Any | None = None,
        window_end: Any | None = None,
    ) -> None:
        self.id = quota_id
        self.system_id = system_id
        self.env_key = env_key
        self.quota_key = quota_key
        self.quota_name = quota_name
        self.quota_window = quota_window
        self.limit_value = limit_value
        self.used_value = used_value
        self.usage_ratio = usage_ratio
        self.warning_threshold = warning_threshold
        self.critical_threshold = critical_threshold
        self.window_start = window_start
        self.window_end = window_end


class _FakeSession:
    pass


class _FakeSessionFactory:
    async def __aenter__(self) -> _FakeSession:
        return _FakeSession()

    async def __aexit__(self, exc_type: Any, exc: Any, tb: Any) -> None:
        pass


@pytest.fixture
def scheduler_module(monkeypatch: pytest.MonkeyPatch) -> Any:
    """加载 scheduler 模块，并将其内部 create_repositories 替换为返回 mock repos 的工厂。"""
    module = _load_scheduler_module()
    repos = _make_repos(deleted_count=5)
    monkeypatch.setattr(
        module,
        "create_repositories",
        lambda _db: repos,
    )
    return module


@pytest.mark.unit
@pytest.mark.asyncio
async def test_cleanupHandler_uses_default_retention_days(scheduler_module: Any) -> None:
    handler = scheduler_module.WebhookEventCleanupHandler(
        session_factory=_FakeSessionFactory,
        default_retention_days=30,
    )

    result = await handler.execute(_make_task_context())

    assert isinstance(result, TaskResult)
    assert result.success is True
    assert result.output["deleted_count"] == 5
    assert result.output["retention_days"] == 30
    repos = scheduler_module.create_repositories(None)
    repos.webhook_event.delete_old_events.assert_called_once()
    before = repos.webhook_event.delete_old_events.call_args.args[0]
    assert before < utc_now_naive()
    assert before > utc_now_naive() - timedelta(days=31)


@pytest.mark.unit
@pytest.mark.asyncio
async def test_cleanupHandler_uses_payload_retention_days(scheduler_module: Any) -> None:
    repos = _make_repos(deleted_count=0)
    scheduler_module.create_repositories = lambda _db: repos
    handler = scheduler_module.WebhookEventCleanupHandler(
        session_factory=_FakeSessionFactory,
        default_retention_days=30,
    )

    result = await handler.execute(_make_task_context({"retention_days": 7}))

    assert result.success is True
    assert result.output["retention_days"] == 7


@pytest.mark.unit
@pytest.mark.asyncio
async def test_cleanupHandler_returns_failure_on_exception(scheduler_module: Any) -> None:
    repos = _make_repos()
    repos.webhook_event.delete_old_events = AsyncMock(side_effect=RuntimeError("db down"))
    scheduler_module.create_repositories = lambda _db: repos
    handler = scheduler_module.WebhookEventCleanupHandler(session_factory=_FakeSessionFactory)

    result = await handler.execute(_make_task_context())

    assert result.success is False
    assert "db down" in (result.error or "")


@pytest.mark.unit
def test_registerSchedulerHandlers_registers_cleanup_handler() -> None:
    module = _load_scheduler_module()
    registry = HandlerRegistry()
    module.register_scheduler_handlers(registry)

    handler = registry.get("external_systems.webhook_event_cleanup")
    assert handler is not None
    assert handler.name == "external_systems.webhook_event_cleanup"
    assert handler.description == "清理超过保留期的 Webhook 事件"


# ─── PendingWebhookEventConsumerHandler ────────────────────────────────────


def _patch_container_for_pending_consumer(
    monkeypatch: pytest.MonkeyPatch,
    consume_results: list[dict[str, Any]],
) -> AsyncMock:
    """向 sys.modules 注入 fake container，mock webhook_event_handler.consume_pending_event。"""
    fake_service = AsyncMock()
    fake_service.consume_pending_event = AsyncMock(side_effect=consume_results)
    fake_use_cases = MagicMock()
    fake_use_cases.webhook_event_handler = fake_service

    fake_container = types.ModuleType("yuxi.external_systems.infrastructure.container")
    fake_container.create_use_cases_from_db = lambda _db: fake_use_cases
    monkeypatch.setitem(
        sys.modules,
        "yuxi.external_systems.infrastructure.container",
        fake_container,
    )
    return fake_service


class _EventStub:
    """WebhookEvent 存根，用于 pending consumer 测试。"""

    def __init__(self, event_id: int | None, subscription_slug: str = "sub") -> None:
        self.id = event_id
        self.subscription_slug = subscription_slug


@pytest.mark.unit
@pytest.mark.asyncio
async def test_pendingConsumerHandler_consumes_pending_events(
    scheduler_module: Any,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    events = [_EventStub(1), _EventStub(2)]
    repos = _make_repos()
    repos.webhook_event.list_pending = AsyncMock(return_value=events)
    scheduler_module.create_repositories = lambda _db: repos
    fake_service = _patch_container_for_pending_consumer(monkeypatch, [None, None])

    handler = scheduler_module.PendingWebhookEventConsumerHandler(
        session_factory=_FakeSessionFactory,
        default_batch_size=100,
    )
    result = await handler.execute(_make_task_context())

    assert isinstance(result, TaskResult)
    assert result.success is True
    assert result.output["consumed_count"] == 2
    assert result.output["failed_count"] == 0
    fake_service.consume_pending_event.assert_awaited()
    assert repos.webhook_event.list_pending.call_args.kwargs["limit"] == 100


@pytest.mark.unit
@pytest.mark.asyncio
async def test_pendingConsumerHandler_uses_payload_batch_size(
    scheduler_module: Any,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    events = [_EventStub(1)]
    repos = _make_repos()
    repos.webhook_event.list_pending = AsyncMock(return_value=events)
    scheduler_module.create_repositories = lambda _db: repos
    _patch_container_for_pending_consumer(monkeypatch, [None])

    handler = scheduler_module.PendingWebhookEventConsumerHandler(
        session_factory=_FakeSessionFactory,
        default_batch_size=100,
    )
    result = await handler.execute(_make_task_context({"batch_size": 50}))

    assert result.success is True
    assert repos.webhook_event.list_pending.call_args.kwargs["limit"] == 50


@pytest.mark.unit
@pytest.mark.asyncio
async def test_pendingConsumerHandler_isolates_single_failure(
    scheduler_module: Any,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    events = [_EventStub(1), _EventStub(2)]
    repos = _make_repos()
    repos.webhook_event.list_pending = AsyncMock(return_value=events)
    scheduler_module.create_repositories = lambda _db: repos
    fake_service = _patch_container_for_pending_consumer(
        monkeypatch,
        [RuntimeError("consume failed"), None],
    )

    handler = scheduler_module.PendingWebhookEventConsumerHandler(
        session_factory=_FakeSessionFactory,
    )
    result = await handler.execute(_make_task_context())

    assert result.success is True
    assert result.output["consumed_count"] == 1
    assert result.output["failed_count"] == 1
    assert len(result.output["errors"]) == 1
    fake_service.consume_pending_event.assert_awaited()


@pytest.mark.unit
@pytest.mark.asyncio
async def test_pendingConsumerHandler_skips_event_without_id(
    scheduler_module: Any,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    events = [_EventStub(None), _EventStub(2)]
    repos = _make_repos()
    repos.webhook_event.list_pending = AsyncMock(return_value=events)
    scheduler_module.create_repositories = lambda _db: repos
    fake_service = _patch_container_for_pending_consumer(monkeypatch, [None])

    handler = scheduler_module.PendingWebhookEventConsumerHandler(
        session_factory=_FakeSessionFactory,
    )
    result = await handler.execute(_make_task_context())

    assert result.success is True
    assert result.output["consumed_count"] == 1
    assert result.output["failed_count"] == 1
    assert fake_service.consume_pending_event.await_count == 1


@pytest.mark.unit
@pytest.mark.asyncio
async def test_pendingConsumerHandler_returns_failure_on_exception(
    scheduler_module: Any,
) -> None:
    repos = _make_repos()
    repos.webhook_event.list_pending = AsyncMock(side_effect=RuntimeError("db down"))
    scheduler_module.create_repositories = lambda _db: repos

    handler = scheduler_module.PendingWebhookEventConsumerHandler(
        session_factory=_FakeSessionFactory,
    )
    result = await handler.execute(_make_task_context())

    assert result.success is False
    assert "db down" in (result.error or "")


@pytest.mark.unit
def test_registerSchedulerHandlers_registers_pending_consumer_handler() -> None:
    module = _load_scheduler_module()
    registry = HandlerRegistry()
    module.register_scheduler_handlers(registry)

    handler = registry.get("external_systems.webhook_pending_consumer")
    assert handler is not None
    assert handler.name == "external_systems.webhook_pending_consumer"
    assert handler.description == "定时消费 pending 状态的 Webhook 事件"


# ─── ProcessingTimeoutRecoveryHandler ──────────────────────────────────────


@pytest.mark.unit
@pytest.mark.asyncio
async def test_processingTimeoutRecoveryHandler_resets_stuck_events(
    scheduler_module: Any,
) -> None:
    events = [_EventStub(1), _EventStub(2)]
    repos = _make_repos()
    repos.webhook_event.list_processing_stuck = AsyncMock(return_value=events)
    repos.webhook_event.update_processing = AsyncMock(return_value=None)
    scheduler_module.create_repositories = lambda _db: repos

    handler = scheduler_module.ProcessingTimeoutRecoveryHandler(
        session_factory=_FakeSessionFactory,
        default_timeout_seconds=300,
        default_batch_size=100,
    )
    result = await handler.execute(_make_task_context())

    assert isinstance(result, TaskResult)
    assert result.success is True
    assert result.output["recovered_count"] == 2
    assert result.output["failed_count"] == 0
    repos.webhook_event.list_processing_stuck.assert_called_once_with(
        300,
        limit=100,
    )
    assert repos.webhook_event.update_processing.await_count == 2
    call = repos.webhook_event.update_processing.call_args
    assert call.kwargs["status"] == "pending"


@pytest.mark.unit
@pytest.mark.asyncio
async def test_processingTimeoutRecoveryHandler_uses_payload_params(
    scheduler_module: Any,
) -> None:
    events = [_EventStub(1)]
    repos = _make_repos()
    repos.webhook_event.list_processing_stuck = AsyncMock(return_value=events)
    repos.webhook_event.update_processing = AsyncMock(return_value=None)
    scheduler_module.create_repositories = lambda _db: repos

    handler = scheduler_module.ProcessingTimeoutRecoveryHandler(
        session_factory=_FakeSessionFactory,
    )
    result = await handler.execute(_make_task_context({"timeout_seconds": 60, "batch_size": 10}))

    assert result.success is True
    repos.webhook_event.list_processing_stuck.assert_called_once_with(
        60,
        limit=10,
    )


@pytest.mark.unit
@pytest.mark.asyncio
async def test_processingTimeoutRecoveryHandler_isolates_single_failure(
    scheduler_module: Any,
) -> None:
    events = [_EventStub(1), _EventStub(2)]
    repos = _make_repos()
    repos.webhook_event.list_processing_stuck = AsyncMock(return_value=events)
    repos.webhook_event.update_processing = AsyncMock(side_effect=[RuntimeError("locked"), None])
    scheduler_module.create_repositories = lambda _db: repos

    handler = scheduler_module.ProcessingTimeoutRecoveryHandler(
        session_factory=_FakeSessionFactory,
    )
    result = await handler.execute(_make_task_context())

    assert result.success is True
    assert result.output["recovered_count"] == 1
    assert result.output["failed_count"] == 1
    assert len(result.output["errors"]) == 1


@pytest.mark.unit
@pytest.mark.asyncio
async def test_processingTimeoutRecoveryHandler_skips_event_without_id(
    scheduler_module: Any,
) -> None:
    events = [_EventStub(None), _EventStub(2)]
    repos = _make_repos()
    repos.webhook_event.list_processing_stuck = AsyncMock(return_value=events)
    repos.webhook_event.update_processing = AsyncMock(return_value=None)
    scheduler_module.create_repositories = lambda _db: repos

    handler = scheduler_module.ProcessingTimeoutRecoveryHandler(
        session_factory=_FakeSessionFactory,
    )
    result = await handler.execute(_make_task_context())

    assert result.success is True
    assert result.output["recovered_count"] == 1
    assert result.output["failed_count"] == 1
    assert repos.webhook_event.update_processing.await_count == 1


@pytest.mark.unit
@pytest.mark.asyncio
async def test_processingTimeoutRecoveryHandler_returns_failure_on_exception(
    scheduler_module: Any,
) -> None:
    repos = _make_repos()
    repos.webhook_event.list_processing_stuck = AsyncMock(side_effect=RuntimeError("db down"))
    scheduler_module.create_repositories = lambda _db: repos

    handler = scheduler_module.ProcessingTimeoutRecoveryHandler(
        session_factory=_FakeSessionFactory,
    )
    result = await handler.execute(_make_task_context())

    assert result.success is False
    assert "db down" in (result.error or "")


@pytest.mark.unit
def test_registerSchedulerHandlers_registers_processing_recovery_handler() -> None:
    module = _load_scheduler_module()
    registry = HandlerRegistry()
    module.register_scheduler_handlers(registry)

    handler = registry.get("external_systems.webhook_processing_recovery")
    assert handler is not None
    assert handler.name == "external_systems.webhook_processing_recovery"
    assert handler.description == "将 processing 状态超时的 Webhook 事件重置为 pending"


# ─── SecretRotationHandler ─────────────────────────────────────────────────


def _patch_container_for_secret_rotation(
    monkeypatch: pytest.MonkeyPatch,
    scheduled_rotation_output: Any,
) -> AsyncMock:
    """向 sys.modules 注入 fake container，避免本地非容器环境导入 app_config 失败。"""
    fake_service = AsyncMock()
    fake_service.scheduled_rotation = AsyncMock(return_value=scheduled_rotation_output)
    fake_use_cases = MagicMock()
    fake_use_cases.secret_rotation_service = fake_service

    fake_container = types.ModuleType("yuxi.external_systems.infrastructure.container")
    fake_container.create_use_cases_from_db = lambda _db: fake_use_cases
    monkeypatch.setitem(
        sys.modules,
        "yuxi.external_systems.infrastructure.container",
        fake_container,
    )
    return fake_service


@pytest.mark.unit
@pytest.mark.asyncio
async def test_secretRotationHandler_executes_scheduled_rotation(
    scheduler_module: Any,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    from yuxi.external_systems.use_cases.dto.secret_rotation import ScheduledRotationOutput

    fake_service = _patch_container_for_secret_rotation(
        monkeypatch,
        ScheduledRotationOutput(affected=2, succeeded=2, failed=0),
    )

    handler = scheduler_module.SecretRotationHandler(session_factory=_FakeSessionFactory)
    result = await handler.execute(_make_task_context())

    assert isinstance(result, TaskResult)
    assert result.success is True
    assert result.output["affected"] == 2
    assert result.output["succeeded"] == 2
    fake_service.scheduled_rotation.assert_awaited_once()


@pytest.mark.unit
@pytest.mark.asyncio
async def test_secretRotationHandler_returns_failure_on_exception(
    scheduler_module: Any,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    fake_service = AsyncMock()
    fake_service.scheduled_rotation = AsyncMock(side_effect=RuntimeError("rotation failed"))
    fake_use_cases = MagicMock()
    fake_use_cases.secret_rotation_service = fake_service

    fake_container = types.ModuleType("yuxi.external_systems.infrastructure.container")
    fake_container.create_use_cases_from_db = lambda _db: fake_use_cases
    monkeypatch.setitem(
        sys.modules,
        "yuxi.external_systems.infrastructure.container",
        fake_container,
    )

    handler = scheduler_module.SecretRotationHandler(session_factory=_FakeSessionFactory)
    result = await handler.execute(_make_task_context())

    assert result.success is False
    assert "rotation failed" in (result.error or "")


# ─── QuotaThresholdAlertHandler ────────────────────────────────────────────


@pytest.mark.unit
@pytest.mark.asyncio
async def test_quotaThresholdAlertHandler_fires_critical_and_warning(
    scheduler_module: Any,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    critical_quota = _QuotaStub(quota_id=1, quota_name="Q1", usage_ratio=0.96)
    warning_quota = _QuotaStub(quota_id=2, quota_name="Q2", usage_ratio=0.85)
    repos = _make_repos(
        near_critical=[critical_quota],
        near_warning=[warning_quota],
    )
    scheduler_module.create_repositories = lambda _db: repos

    class _FakeAlertManager:
        fired: list[dict[str, Any]] = []

        async def fire(
            self,
            system_id: Any,
            env_key: Any,
            alert_type: Any,
            **kwargs: Any,
        ) -> None:
            self.fired.append(
                {
                    "system_id": system_id,
                    "env_key": env_key,
                    "alert_type": alert_type,
                    **kwargs,
                }
            )

        def build_dedup_key(self, *args: Any, **kwargs: Any) -> str:
            return "dedup"

    fake_alert_manager = _FakeAlertManager()
    monkeypatch.setattr(
        scheduler_module,
        "AlertManagerImpl",
        lambda _alert_repo: fake_alert_manager,
    )

    handler = scheduler_module.QuotaThresholdAlertHandler(
        session_factory=_FakeSessionFactory,
    )

    result = await handler.execute(_make_task_context())

    assert result.success is True
    assert result.output["critical_count"] == 1
    assert result.output["warning_count"] == 1
    assert len(fake_alert_manager.fired) == 2


@pytest.mark.unit
@pytest.mark.asyncio
async def test_quotaThresholdAlertHandler_dedup_warning_when_critical(
    scheduler_module: Any,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    quota = _QuotaStub(quota_id=1, quota_name="Q1", usage_ratio=0.97)
    repos = _make_repos(
        near_critical=[quota],
        near_warning=[quota],
    )
    scheduler_module.create_repositories = lambda _db: repos

    class _FakeAlertManager:
        fired: list[dict[str, Any]] = []

        async def fire(
            self,
            system_id: Any,
            env_key: Any,
            alert_type: Any,
            **kwargs: Any,
        ) -> None:
            self.fired.append(
                {
                    "system_id": system_id,
                    "env_key": env_key,
                    "alert_type": alert_type,
                    **kwargs,
                }
            )

    fake_alert_manager = _FakeAlertManager()
    monkeypatch.setattr(
        scheduler_module,
        "AlertManagerImpl",
        lambda _alert_repo: fake_alert_manager,
    )

    handler = scheduler_module.QuotaThresholdAlertHandler(
        session_factory=_FakeSessionFactory,
    )
    result = await handler.execute(_make_task_context())

    assert result.success is True
    assert result.output["warning_count"] == 0
    assert len(fake_alert_manager.fired) == 1


@pytest.mark.unit
@pytest.mark.asyncio
async def test_quotaThresholdAlertHandler_returns_failure_on_exception(scheduler_module: Any) -> None:
    repos = _make_repos()
    repos.quota_usage.list_near_critical = AsyncMock(side_effect=RuntimeError("db down"))
    scheduler_module.create_repositories = lambda _db: repos

    handler = scheduler_module.QuotaThresholdAlertHandler(
        session_factory=_FakeSessionFactory,
    )
    result = await handler.execute(_make_task_context())

    assert result.success is False
    assert "db down" in (result.error or "")


# ─── QuotaWindowResetHandler ───────────────────────────────────────────────


@pytest.mark.unit
@pytest.mark.asyncio
async def test_quotaWindowResetHandler_resets_expired_windows(scheduler_module: Any) -> None:
    now = utc_now_naive()
    quota = _QuotaStub(
        system_id=1,
        env_key="default",
        quota_key="api_calls",
        quota_window="daily",
        window_start=now - timedelta(days=2),
        window_end=now - timedelta(hours=1),
    )
    repos = _make_repos(expiring_quotas=[quota])
    scheduler_module.create_repositories = lambda _db: repos

    handler = scheduler_module.QuotaWindowResetHandler(
        session_factory=_FakeSessionFactory,
    )
    result = await handler.execute(_make_task_context())

    assert result.success is True
    assert result.output["reset_count"] == 1
    assert result.output["failed_count"] == 0
    repos.quota_usage.reset_window.assert_called_once()
    call = repos.quota_usage.reset_window.call_args
    assert call.kwargs["used_value"] == 0
    assert call.kwargs["updated_by"] == "scheduler"


@pytest.mark.unit
@pytest.mark.asyncio
async def test_quotaWindowResetHandler_rolling_24h_window(scheduler_module: Any) -> None:
    now = utc_now_naive()
    quota = _QuotaStub(
        quota_window="rolling_24h",
        window_start=now - timedelta(days=2),
        window_end=now - timedelta(hours=1),
    )
    repos = _make_repos(expiring_quotas=[quota])
    scheduler_module.create_repositories = lambda _db: repos

    handler = scheduler_module.QuotaWindowResetHandler(
        session_factory=_FakeSessionFactory,
    )
    result = await handler.execute(_make_task_context())

    assert result.success is True
    assert result.output["reset_count"] == 1
    repos.quota_usage.reset_window.assert_called_once()


@pytest.mark.unit
@pytest.mark.asyncio
async def test_quotaWindowResetHandler_isolates_single_failure(scheduler_module: Any) -> None:
    now = utc_now_naive()
    quota = _QuotaStub(
        system_id=1,
        env_key="default",
        quota_key="api_calls",
        quota_window="daily",
        window_end=now - timedelta(hours=1),
    )
    repos = _make_repos(expiring_quotas=[quota])
    repos.quota_usage.reset_window = AsyncMock(side_effect=RuntimeError("locked"))
    scheduler_module.create_repositories = lambda _db: repos

    handler = scheduler_module.QuotaWindowResetHandler(
        session_factory=_FakeSessionFactory,
    )
    result = await handler.execute(_make_task_context())

    assert result.success is True
    assert result.output["reset_count"] == 0
    assert result.output["failed_count"] == 1
    assert len(result.output["errors"]) == 1


@pytest.mark.unit
@pytest.mark.asyncio
async def test_quotaWindowResetHandler_returns_failure_on_exception(scheduler_module: Any) -> None:
    repos = _make_repos()
    repos.quota_usage.list = AsyncMock(side_effect=RuntimeError("db down"))
    scheduler_module.create_repositories = lambda _db: repos

    handler = scheduler_module.QuotaWindowResetHandler(
        session_factory=_FakeSessionFactory,
    )
    result = await handler.execute(_make_task_context())

    assert result.success is False
    assert "db down" in (result.error or "")


@pytest.mark.unit
def test_registerSchedulerHandlers_registers_quota_handlers() -> None:
    module = _load_scheduler_module()
    registry = HandlerRegistry()
    module.register_scheduler_handlers(registry)

    assert registry.get("external_systems.quota_threshold_alert") is not None
    assert registry.get("external_systems.quota_window_reset") is not None


@pytest.mark.unit
def test_registerSchedulerHandlers_registers_secret_rotation_handler() -> None:
    module = _load_scheduler_module()
    registry = HandlerRegistry()
    module.register_scheduler_handlers(registry)

    handler = registry.get("external_systems.secret_rotation")
    assert handler is not None
    assert handler.name == "external_systems.secret_rotation"
    assert handler.description == "检查到期密钥轮换策略并执行轮换/告警"


# ─── TrashPurgeHandler ────────────────────────────────────────────────────


def _patch_container_for_trash_purge(
    monkeypatch: pytest.MonkeyPatch,
    purge_output: Any,
) -> AsyncMock:
    """向 sys.modules 注入 fake container，mock trash_service.purge。"""
    fake_service = AsyncMock()
    fake_service.purge = AsyncMock(return_value=purge_output)
    fake_use_cases = MagicMock()
    fake_use_cases.trash_service = fake_service

    fake_container = types.ModuleType("yuxi.external_systems.infrastructure.container")
    fake_container.create_use_cases_from_db = lambda _db: fake_use_cases
    monkeypatch.setitem(
        sys.modules,
        "yuxi.external_systems.infrastructure.container",
        fake_container,
    )
    return fake_service


@pytest.mark.unit
@pytest.mark.asyncio
async def test_trashPurgeHandler_executes_purge_with_default_days(
    scheduler_module: Any,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    from yuxi.external_systems.use_cases.dto.trash import PurgeTrashOutput

    fake_service = _patch_container_for_trash_purge(
        monkeypatch,
        PurgeTrashOutput(purged_count=3, by_type={"system": 2, "tool": 1}, cutoff_at=utc_now_naive()),
    )

    handler = scheduler_module.TrashPurgeHandler(
        session_factory=_FakeSessionFactory,
        default_older_than_days=90,
    )
    result = await handler.execute(_make_task_context())

    assert isinstance(result, TaskResult)
    assert result.success is True
    assert result.output["purged_count"] == 3
    assert result.output["by_type"] == {"system": 2, "tool": 1}
    fake_service.purge.assert_awaited_once()
    call = fake_service.purge.call_args
    assert call.args[0].older_than_days == 90
    assert call.args[0].purged_by == "scheduler"


@pytest.mark.unit
@pytest.mark.asyncio
async def test_trashPurgeHandler_uses_payload_older_than_days(
    scheduler_module: Any,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    from yuxi.external_systems.use_cases.dto.trash import PurgeTrashOutput

    fake_service = _patch_container_for_trash_purge(
        monkeypatch,
        PurgeTrashOutput(purged_count=0, by_type={}, cutoff_at=utc_now_naive()),
    )

    handler = scheduler_module.TrashPurgeHandler(
        session_factory=_FakeSessionFactory,
        default_older_than_days=90,
    )
    result = await handler.execute(_make_task_context({"older_than_days": 30}))

    assert result.success is True
    assert result.output["purged_count"] == 0
    call = fake_service.purge.call_args
    assert call.args[0].older_than_days == 30


@pytest.mark.unit
@pytest.mark.asyncio
async def test_trashPurgeHandler_returns_failure_on_exception(
    scheduler_module: Any,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    fake_service = AsyncMock()
    fake_service.purge = AsyncMock(side_effect=RuntimeError("db down"))
    fake_use_cases = MagicMock()
    fake_use_cases.trash_service = fake_service

    fake_container = types.ModuleType("yuxi.external_systems.infrastructure.container")
    fake_container.create_use_cases_from_db = lambda _db: fake_use_cases
    monkeypatch.setitem(
        sys.modules,
        "yuxi.external_systems.infrastructure.container",
        fake_container,
    )

    handler = scheduler_module.TrashPurgeHandler(session_factory=_FakeSessionFactory)
    result = await handler.execute(_make_task_context())

    assert result.success is False
    assert "db down" in (result.error or "")


@pytest.mark.unit
def test_registerSchedulerHandlers_registers_trash_purge_handler() -> None:
    module = _load_scheduler_module()
    registry = HandlerRegistry()
    module.register_scheduler_handlers(registry)

    handler = registry.get("external_systems.trash_purge")
    assert handler is not None
    assert handler.name == "external_systems.trash_purge"
    assert handler.description == "定时清理回收站中超过保留期的软删除资源"


# ─── TestRegressionSchedulerHandler ────────────────────────────────────────


def _patch_container_for_test_regression(
    monkeypatch: pytest.MonkeyPatch,
    regression_output: Any,
) -> AsyncMock:
    """向 sys.modules 注入 fake container，mock test_regression_service.scheduled_regression。"""
    fake_service = AsyncMock()
    fake_service.scheduled_regression = AsyncMock(return_value=regression_output)
    fake_use_cases = MagicMock()
    fake_use_cases.test_regression_service = fake_service

    fake_container = types.ModuleType("yuxi.external_systems.infrastructure.container")
    fake_container.create_use_cases_from_db = lambda _db: fake_use_cases
    monkeypatch.setitem(
        sys.modules,
        "yuxi.external_systems.infrastructure.container",
        fake_container,
    )
    return fake_service


@pytest.mark.unit
@pytest.mark.asyncio
async def test_testRegressionSchedulerHandler_executes_scheduled_regression(
    scheduler_module: Any,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    from yuxi.external_systems.use_cases.dto.test_case import ScheduledRegressionOutput

    fake_service = _patch_container_for_test_regression(
        monkeypatch,
        ScheduledRegressionOutput(affected=3, succeeded=2, failed=1, errors=["e1"]),
    )

    handler = scheduler_module.TestRegressionSchedulerHandler(session_factory=_FakeSessionFactory)
    result = await handler.execute(_make_task_context())

    assert isinstance(result, TaskResult)
    assert result.success is True
    assert result.output["affected"] == 3
    assert result.output["succeeded"] == 2
    assert result.output["failed"] == 1
    fake_service.scheduled_regression.assert_awaited_once()


@pytest.mark.unit
@pytest.mark.asyncio
async def test_testRegressionSchedulerHandler_returns_failure_on_exception(
    scheduler_module: Any,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    fake_service = AsyncMock()
    fake_service.scheduled_regression = AsyncMock(side_effect=RuntimeError("regression failed"))
    fake_use_cases = MagicMock()
    fake_use_cases.test_regression_service = fake_service

    fake_container = types.ModuleType("yuxi.external_systems.infrastructure.container")
    fake_container.create_use_cases_from_db = lambda _db: fake_use_cases
    monkeypatch.setitem(
        sys.modules,
        "yuxi.external_systems.infrastructure.container",
        fake_container,
    )

    handler = scheduler_module.TestRegressionSchedulerHandler(session_factory=_FakeSessionFactory)
    result = await handler.execute(_make_task_context())

    assert result.success is False
    assert "regression failed" in (result.error or "")


@pytest.mark.unit
def test_registerSchedulerHandlers_registers_test_regression_handler() -> None:
    module = _load_scheduler_module()
    registry = HandlerRegistry()
    module.register_scheduler_handlers(registry)

    handler = registry.get("external_systems.test_regression")
    assert handler is not None
    assert handler.name == "external_systems.test_regression"
    assert handler.description == "定时执行到期的外部系统工具测试用例回归"


# ─── _compute_next_window ──────────────────────────────────────────────────


def test_computeNextWindow_rolling_24h() -> None:
    module = _load_scheduler_module()
    now = utc_now_naive()
    start, end = module._compute_next_window("rolling_24h", None, None, now)
    assert start == now
    assert end == now + timedelta(hours=24)


def test_computeNextWindow_daily() -> None:
    module = _load_scheduler_module()
    now = utc_now_naive()
    last_end = now - timedelta(hours=1)
    start, end = module._compute_next_window("daily", None, last_end, now)
    assert start == last_end
    assert end == last_end + timedelta(days=1)


def test_computeNextWindow_hourly() -> None:
    module = _load_scheduler_module()
    now = utc_now_naive()
    last_end = now - timedelta(minutes=30)
    start, end = module._compute_next_window("hourly", None, last_end, now)
    assert start == last_end
    assert end == last_end + timedelta(hours=1)


def test_computeNextWindow_minute() -> None:
    module = _load_scheduler_module()
    now = utc_now_naive()
    last_end = now - timedelta(seconds=30)
    start, end = module._compute_next_window("minute", None, last_end, now)
    assert start == last_end
    assert end == last_end + timedelta(minutes=1)


def test_computeNextWindow_custom_preserves_previous_range() -> None:
    module = _load_scheduler_module()
    now = utc_now_naive()
    window_start = now - timedelta(hours=2)
    window_end = now - timedelta(minutes=30)
    start, end = module._compute_next_window("custom", window_start, window_end, now)
    assert start == window_end
    assert end == window_end + (window_end - window_start)


def test_computeNextWindow_custom_invalid_range_falls_back_to_one_day() -> None:
    module = _load_scheduler_module()
    now = utc_now_naive()
    window_start = now
    window_end = now - timedelta(hours=1)  # end < start
    start, end = module._compute_next_window("custom", window_start, window_end, now)
    assert start == window_end
    assert end == window_end + timedelta(days=1)


def test_computeNextWindow_custom_without_window_range_falls_back_to_one_day() -> None:
    module = _load_scheduler_module()
    now = utc_now_naive()
    last_end = now - timedelta(hours=1)
    start, end = module._compute_next_window("custom", None, last_end, now)
    assert start == last_end
    assert end == last_end + timedelta(days=1)


def test_computeNextWindow_uses_now_when_window_end_none() -> None:
    module = _load_scheduler_module()
    now = utc_now_naive()
    start, end = module._compute_next_window("daily", None, None, now)
    assert start == now
    assert end == now + timedelta(days=1)


# ─── WebhookRenewalHandler ─────────────────────────────────────────────────


class _SubscriptionStub:
    """WebhookSubscription 存根，用于 renewal handler 测试。"""

    def __init__(
        self,
        subscription_id: int | None = 1,
        slug: str = "sub-1",
        retention_days: int = 30,
    ) -> None:
        self.id = subscription_id
        self.slug = slug
        self.retention_days = retention_days


@pytest.mark.unit
@pytest.mark.asyncio
async def test_webhookRenewalHandler_renews_expiring_subscriptions(
    scheduler_module: Any,
) -> None:
    subscriptions = [_SubscriptionStub(subscription_id=1, retention_days=30)]
    repos = _make_repos(expiring_subscriptions=subscriptions)
    scheduler_module.create_repositories = lambda _db: repos

    handler = scheduler_module.WebhookRenewalHandler(session_factory=_FakeSessionFactory)
    result = await handler.execute(_make_task_context())

    assert isinstance(result, TaskResult)
    assert result.success is True
    assert result.output["renewed_count"] == 1
    assert result.output["failed_count"] == 0
    assert result.output["threshold_days"] == 7
    repos.webhook_subscription.list_expiring.assert_called_once()
    before = repos.webhook_subscription.list_expiring.call_args.kwargs["before"]
    assert before > utc_now_naive()
    repos.webhook_subscription.update_renewal_result.assert_called_once()
    call = repos.webhook_subscription.update_renewal_result.call_args
    assert call.args[0] == 1
    assert call.kwargs["success"] is True
    assert call.kwargs["new_expires_at"] is not None


@pytest.mark.unit
@pytest.mark.asyncio
async def test_webhookRenewalHandler_uses_payload_threshold_days(
    scheduler_module: Any,
) -> None:
    subscriptions = [_SubscriptionStub(subscription_id=1)]
    repos = _make_repos(expiring_subscriptions=subscriptions)
    scheduler_module.create_repositories = lambda _db: repos

    handler = scheduler_module.WebhookRenewalHandler(session_factory=_FakeSessionFactory)
    result = await handler.execute(_make_task_context({"threshold_days": 14}))

    assert result.success is True
    assert result.output["threshold_days"] == 14
    before = repos.webhook_subscription.list_expiring.call_args.kwargs["before"]
    assert before > utc_now_naive() + timedelta(days=13)
    assert before < utc_now_naive() + timedelta(days=15)


@pytest.mark.unit
@pytest.mark.asyncio
async def test_webhookRenewalHandler_skips_subscription_without_id(
    scheduler_module: Any,
) -> None:
    subscriptions = [
        _SubscriptionStub(subscription_id=None),
        _SubscriptionStub(subscription_id=2),
    ]
    repos = _make_repos(expiring_subscriptions=subscriptions)
    scheduler_module.create_repositories = lambda _db: repos

    handler = scheduler_module.WebhookRenewalHandler(session_factory=_FakeSessionFactory)
    result = await handler.execute(_make_task_context())

    assert result.success is True
    assert result.output["renewed_count"] == 1
    assert result.output["failed_count"] == 1
    assert repos.webhook_subscription.update_renewal_result.await_count == 1


@pytest.mark.unit
@pytest.mark.asyncio
async def test_webhookRenewalHandler_isolates_single_failure(
    scheduler_module: Any,
) -> None:
    subscriptions = [
        _SubscriptionStub(subscription_id=1),
        _SubscriptionStub(subscription_id=2),
    ]
    repos = _make_repos(expiring_subscriptions=subscriptions)
    repos.webhook_subscription.update_renewal_result = AsyncMock(
        side_effect=[RuntimeError("locked"), None],
    )
    scheduler_module.create_repositories = lambda _db: repos

    handler = scheduler_module.WebhookRenewalHandler(session_factory=_FakeSessionFactory)
    result = await handler.execute(_make_task_context())

    assert result.success is True
    assert result.output["renewed_count"] == 1
    assert result.output["failed_count"] == 1


@pytest.mark.unit
@pytest.mark.asyncio
async def test_webhookRenewalHandler_returns_failure_on_exception(
    scheduler_module: Any,
) -> None:
    repos = _make_repos()
    repos.webhook_subscription.list_expiring = AsyncMock(side_effect=RuntimeError("db down"))
    scheduler_module.create_repositories = lambda _db: repos

    handler = scheduler_module.WebhookRenewalHandler(session_factory=_FakeSessionFactory)
    result = await handler.execute(_make_task_context())

    assert result.success is False
    assert "db down" in (result.error or "")


@pytest.mark.unit
def test_registerSchedulerHandlers_registers_webhook_renewal_handler() -> None:
    module = _load_scheduler_module()
    registry = HandlerRegistry()
    module.register_scheduler_handlers(registry)

    handler = registry.get("external_systems.webhook_renewal")
    assert handler is not None
    assert handler.name == "external_systems.webhook_renewal"
    assert handler.description == "自动续期即将过期的 Webhook 订阅"


# ─── AuditLogRetentionHandler ──────────────────────────────────────────────


@pytest.mark.unit
@pytest.mark.asyncio
async def test_auditLogRetentionHandler_uses_default_retention_days(
    scheduler_module: Any,
) -> None:
    repos = _make_repos(audit_log_deleted_count=7)
    scheduler_module.create_repositories = lambda _db: repos

    handler = scheduler_module.AuditLogRetentionHandler(
        session_factory=_FakeSessionFactory,
        default_retention_days=90,
    )
    result = await handler.execute(_make_task_context())

    assert isinstance(result, TaskResult)
    assert result.success is True
    assert result.output["deleted_count"] == 7
    assert result.output["retention_days"] == 90
    repos.audit_log.delete_old_logs.assert_called_once()
    before = repos.audit_log.delete_old_logs.call_args.args[0]
    assert before < utc_now_naive()
    assert before > utc_now_naive() - timedelta(days=91)
    assert repos.audit_log.delete_old_logs.call_args.kwargs["updated_by"] == "scheduler"


@pytest.mark.unit
@pytest.mark.asyncio
async def test_auditLogRetentionHandler_uses_payload_retention_days(
    scheduler_module: Any,
) -> None:
    repos = _make_repos(audit_log_deleted_count=0)
    scheduler_module.create_repositories = lambda _db: repos

    handler = scheduler_module.AuditLogRetentionHandler(
        session_factory=_FakeSessionFactory,
        default_retention_days=90,
    )
    result = await handler.execute(_make_task_context({"retention_days": 30}))

    assert result.success is True
    assert result.output["retention_days"] == 30


@pytest.mark.unit
@pytest.mark.asyncio
async def test_auditLogRetentionHandler_returns_failure_on_exception(
    scheduler_module: Any,
) -> None:
    repos = _make_repos()
    repos.audit_log.delete_old_logs = AsyncMock(side_effect=RuntimeError("db down"))
    scheduler_module.create_repositories = lambda _db: repos

    handler = scheduler_module.AuditLogRetentionHandler(session_factory=_FakeSessionFactory)
    result = await handler.execute(_make_task_context())

    assert result.success is False
    assert "db down" in (result.error or "")


@pytest.mark.unit
def test_registerSchedulerHandlers_registers_audit_log_retention_handler() -> None:
    module = _load_scheduler_module()
    registry = HandlerRegistry()
    module.register_scheduler_handlers(registry)

    handler = registry.get("external_systems.audit_log_retention")
    assert handler is not None
    assert handler.name == "external_systems.audit_log_retention"
    assert handler.description == "清理超过保留期的外部系统审计日志"


# ─── HealthCheckSchedulerHandler ───────────────────────────────────────────


def _patch_container_for_health_check(
    monkeypatch: pytest.MonkeyPatch,
) -> AsyncMock:
    """向 sys.modules 注入 fake container，mock tool_service.trigger_system_health_check。"""
    fake_service = AsyncMock()
    fake_service.trigger_system_health_check = AsyncMock()
    fake_use_cases = MagicMock()
    fake_use_cases.tool_service = fake_service

    fake_container = types.ModuleType("yuxi.external_systems.infrastructure.container")
    fake_container.create_use_cases_from_db = lambda _db: fake_use_cases
    monkeypatch.setitem(
        sys.modules,
        "yuxi.external_systems.infrastructure.container",
        fake_container,
    )
    return fake_service


@pytest.mark.unit
@pytest.mark.asyncio
async def test_healthCheckSchedulerHandler_checks_systems(
    scheduler_module: Any,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    from yuxi.external_systems.use_cases.dto.health_check import (
        TriggerSystemHealthCheckInput,
    )

    repos = _make_repos(system_ids=[1, 2, 3])
    scheduler_module.create_repositories = lambda _db: repos
    fake_service = _patch_container_for_health_check(monkeypatch)

    handler = scheduler_module.HealthCheckSchedulerHandler(
        session_factory=_FakeSessionFactory,
        default_batch_size=50,
    )
    result = await handler.execute(_make_task_context())

    assert isinstance(result, TaskResult)
    assert result.success is True
    assert result.output["checked_count"] == 3
    assert result.output["failed_count"] == 0
    assert result.output["batch_size"] == 50
    assert fake_service.trigger_system_health_check.await_count == 3
    call = fake_service.trigger_system_health_check.call_args
    assert isinstance(call.args[0], TriggerSystemHealthCheckInput)
    assert call.args[0].triggered_by == "scheduler"
    repos.system.list_ids.assert_called_once_with(
        enabled=True,
        sort_by="created_at",
        sort_order="asc",
    )


@pytest.mark.unit
@pytest.mark.asyncio
async def test_healthCheckSchedulerHandler_uses_payload_batch_size(
    scheduler_module: Any,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    repos = _make_repos(system_ids=[1, 2, 3, 4])
    scheduler_module.create_repositories = lambda _db: repos
    fake_service = _patch_container_for_health_check(monkeypatch)

    handler = scheduler_module.HealthCheckSchedulerHandler(
        session_factory=_FakeSessionFactory,
        default_batch_size=50,
    )
    result = await handler.execute(_make_task_context({"batch_size": 2}))

    assert result.success is True
    assert result.output["batch_size"] == 2
    assert result.output["checked_count"] == 2
    assert fake_service.trigger_system_health_check.await_count == 2


@pytest.mark.unit
@pytest.mark.asyncio
async def test_healthCheckSchedulerHandler_isolates_single_failure(
    scheduler_module: Any,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    repos = _make_repos(system_ids=[1, 2])
    scheduler_module.create_repositories = lambda _db: repos
    fake_service = _patch_container_for_health_check(monkeypatch)
    fake_service.trigger_system_health_check = AsyncMock(
        side_effect=[RuntimeError("probe failed"), None],
    )

    handler = scheduler_module.HealthCheckSchedulerHandler(
        session_factory=_FakeSessionFactory,
    )
    result = await handler.execute(_make_task_context())

    assert result.success is True
    assert result.output["checked_count"] == 1
    assert result.output["failed_count"] == 1
    assert len(result.output["errors"]) == 1
    assert result.output["errors"][0]["system_id"] == 1


@pytest.mark.unit
@pytest.mark.asyncio
async def test_healthCheckSchedulerHandler_returns_failure_on_exception(
    scheduler_module: Any,
) -> None:
    repos = _make_repos()
    repos.system.list_ids = AsyncMock(side_effect=RuntimeError("db down"))
    scheduler_module.create_repositories = lambda _db: repos

    handler = scheduler_module.HealthCheckSchedulerHandler(
        session_factory=_FakeSessionFactory,
    )
    result = await handler.execute(_make_task_context())

    assert result.success is False
    assert "db down" in (result.error or "")


@pytest.mark.unit
def test_registerSchedulerHandlers_registers_health_check_handler() -> None:
    module = _load_scheduler_module()
    registry = HandlerRegistry()
    module.register_scheduler_handlers(registry)

    handler = registry.get("external_systems.health_check")
    assert handler is not None
    assert handler.name == "external_systems.health_check"
    assert handler.description == "定时健康检查探测"


# ─── register_scheduler_handlers 完整性 ────────────────────────────────────


@pytest.mark.unit
def test_registerSchedulerHandlers_registers_all_handlers() -> None:
    """验证 register_scheduler_handlers 注册了 external_systems 全部 11 个 handler。"""
    module = _load_scheduler_module()
    registry = HandlerRegistry()
    module.register_scheduler_handlers(registry)

    expected = {
        "external_systems.webhook_event_cleanup",
        "external_systems.webhook_pending_consumer",
        "external_systems.webhook_processing_recovery",
        "external_systems.webhook_renewal",
        "external_systems.quota_threshold_alert",
        "external_systems.quota_window_reset",
        "external_systems.secret_rotation",
        "external_systems.audit_log_retention",
        "external_systems.trash_purge",
        "external_systems.health_check",
        "external_systems.test_regression",
    }
    assert set(registry.list_names()) == expected