wecom_it_smart_desk/backend/tests/automation/test_timeout_cleaner.py

# =============================================================================
# 复杂场景重构 — 超时清理定时任务单元测试
# =============================================================================
# 测试范围：timeout_cleaner.py
#   - run_once 24h 超时自动关闭
#   - run_once 无超时会话时不操作
#   - run_once Redis 恢复点清理
#   - run_once 推送 WS 超时关闭事件
#
# 技术说明：TimeoutCleaner.run_once() 内部使用 db_factory 创建独立 DB 会话并
#   调用 commit()。测试中通过 MockDbFactory 复用 db_session（SAVEPOINT），
#   并将 commit 替换为 flush，避免干扰 conftest 的事务回滚机制。
# =============================================================================

import pytest
from datetime import datetime, timedelta, timezone
from unittest.mock import AsyncMock, patch

from app.constants import (
    PAUSE_TIMEOUT_HOURS,
    REDIS_KEY_PAUSED_SESSIONS,
    REDIS_KEY_RESUME_POINT,
)
from app.models.automation import AutoSession
from app.services.automation.timeout_cleaner import TimeoutCleaner


class MockDbFactory:
    """模拟 DB 会话工厂，复用 db_session 的 SAVEPOINT 事务。

    run_once() 内部调用 await db.commit()，在 SAVEPOINT 环境下会导致
    事务提前提交。此处将 commit 替换为 flush，确保数据写入 SAVEPOINT
    但不提交外层事务，测试结束后由 conftest 回滚清理。
    """

    def __init__(self, db_session):
        self._db = db_session

    def __call__(self):
        return self

    async def __aenter__(self):
        self._db.commit = AsyncMock(side_effect=self._db.flush)
        return self._db

    async def __aexit__(self, *args):
        pass


class TestTimeoutCleaner:
    """超时清理器测试用例"""

    @pytest.mark.asyncio
    async def test_timeout_cleaner(self, db_session, mock_redis):
        """测试 24h 超时自动关闭"""
        # Arrange — 创建一个 25 小时前暂停的会话
        session = AutoSession(
            employee_id="emp_001",
            title="超时任务",
            status="paused",
            paused_at=datetime.now(timezone.utc) - timedelta(hours=25),
        )
        db_session.add(session)
        await db_session.flush()

        # Act
        cleaner = TimeoutCleaner(MockDbFactory(db_session), mock_redis)
        closed_count = await cleaner.run_once()

        # Assert
        assert closed_count == 1
        await db_session.refresh(session)
        assert session.status == "closed"
        assert session.closed_by == "system(timeout)"

    @pytest.mark.asyncio
    async def test_timeout_cleaner_no_expired(self, db_session, mock_redis):
        """测试无超时会话时不操作"""
        # Arrange — 创建一个刚暂停的会话（未超时）
        session = AutoSession(
            employee_id="emp_001",
            title="近期任务",
            status="paused",
            paused_at=datetime.now(timezone.utc) - timedelta(hours=1),
        )
        db_session.add(session)
        await db_session.flush()

        # Act
        cleaner = TimeoutCleaner(MockDbFactory(db_session), mock_redis)
        closed_count = await cleaner.run_once()

        # Assert
        assert closed_count == 0
        await db_session.refresh(session)
        assert session.status == "paused"

    @pytest.mark.asyncio
    async def test_timeout_cleaner_redis_cleanup(self, db_session, mock_redis):
        """测试超时关闭时清理 Redis 恢复点"""
        # Arrange
        session = AutoSession(
            employee_id="emp_redis_test",
            title="Redis清理测试",
            status="paused",
            paused_at=datetime.now(timezone.utc) - timedelta(hours=25),
        )
        db_session.add(session)
        await db_session.flush()
        session_id = session.id
        employee_id = session.employee_id

        # 预置 Redis 恢复点和暂停集合
        resume_key = REDIS_KEY_RESUME_POINT.format(session_id=session_id)
        await mock_redis.setex(resume_key, 90000, '{"title":"test"}')
        paused_key = REDIS_KEY_PAUSED_SESSIONS.format(employee_id=employee_id)
        await mock_redis.sadd(paused_key, session_id)

        # Act
        cleaner = TimeoutCleaner(MockDbFactory(db_session), mock_redis)
        await cleaner.run_once()

        # Assert — Redis 恢复点已删除
        raw = await mock_redis.get(resume_key)
        assert raw is None

        # Assert — 暂停集合中已移除
        members = await mock_redis.smembers(paused_key)
        assert session_id not in members

    @pytest.mark.asyncio
    async def test_timeout_cleaner_multiple_sessions(self, db_session, mock_redis):
        """测试多个超时会话一次性关闭"""
        # Arrange
        sessions = []
        for i in range(3):
            s = AutoSession(
                employee_id=f"emp_{i}",
                title=f"超时任务{i}",
                status="paused",
                paused_at=datetime.now(timezone.utc) - timedelta(hours=26),
            )
            db_session.add(s)
            sessions.append(s)
        await db_session.flush()

        # Act
        cleaner = TimeoutCleaner(MockDbFactory(db_session), mock_redis)
        closed_count = await cleaner.run_once()

        # Assert
        assert closed_count == 3
        for s in sessions:
            await db_session.refresh(s)
            assert s.status == "closed"
            assert s.closed_by == "system(timeout)"

    @pytest.mark.asyncio
    async def test_timeout_cleaner_mixed_sessions(self, db_session, mock_redis):
        """测试混合场景 — 只关闭超时的，不关闭未超时的"""
        # Arrange
        expired = AutoSession(
            employee_id="emp_001",
            title="超时",
            status="paused",
            paused_at=datetime.now(timezone.utc) - timedelta(hours=25),
        )
        recent = AutoSession(
            employee_id="emp_002",
            title="近期",
            status="paused",
            paused_at=datetime.now(timezone.utc) - timedelta(minutes=30),
        )
        db_session.add_all([expired, recent])
        await db_session.flush()

        # Act
        cleaner = TimeoutCleaner(MockDbFactory(db_session), mock_redis)
        closed_count = await cleaner.run_once()

        # Assert
        assert closed_count == 1
        await db_session.refresh(expired)
        assert expired.status == "closed"
        await db_session.refresh(recent)
        assert recent.status == "paused"

    @pytest.mark.asyncio
    async def test_timeout_cleaner_without_redis(self, db_session):
        """测试无 Redis 时超时关闭仍成功"""
        # Arrange
        session = AutoSession(
            employee_id="emp_001",
            title="无Redis超时",
            status="paused",
            paused_at=datetime.now(timezone.utc) - timedelta(hours=25),
        )
        db_session.add(session)
        await db_session.flush()

        # Act — redis=None
        cleaner = TimeoutCleaner(MockDbFactory(db_session), redis=None)
        closed_count = await cleaner.run_once()

        # Assert
        assert closed_count == 1
        await db_session.refresh(session)
        assert session.status == "closed"

    @pytest.mark.asyncio
    async def test_timeout_cleaner_boundary_just_expired(self, db_session, mock_redis):
        """测试边界 — 刚好超过 24 小时的会话被关闭"""
        # Arrange — 24小时 + 1分钟前
        session = AutoSession(
            employee_id="emp_001",
            title="边界测试",
            status="paused",
            paused_at=datetime.now(timezone.utc) - timedelta(hours=24, minutes=1),
        )
        db_session.add(session)
        await db_session.flush()

        # Act
        cleaner = TimeoutCleaner(MockDbFactory(db_session), mock_redis)
        closed_count = await cleaner.run_once()

        # Assert
        assert closed_count == 1

    @pytest.mark.asyncio
    async def test_timeout_cleaner_boundary_just_under(self, db_session, mock_redis):
        """测试边界 — 差 1 分钟到 24 小时的会话不被关闭"""
        # Arrange — 23小时59分钟前
        session = AutoSession(
            employee_id="emp_001",
            title="未超时边界",
            status="paused",
            paused_at=datetime.now(timezone.utc) - timedelta(hours=23, minutes=59),
        )
        db_session.add(session)
        await db_session.flush()

        # Act
        cleaner = TimeoutCleaner(MockDbFactory(db_session), mock_redis)
        closed_count = await cleaner.run_once()

        # Assert
        assert closed_count == 0