wecom_it_smart_desk/backend/tests/test_invite_link.py

# =============================================================================
# 邀请链接回归测试 — 邀请 URL 绝对路径 & join_conversation 状态校验
# =============================================================================
# 测试覆盖（对应本次 3 个关联 Bug 修复）：
#
#   一、join_conversation 状态校验（Bug 2 修复核心）
#     1. ai_handling 状态 → join 成功（本次修复核心）
#     2. serving 状态 → join 成功（原有功能不回归）
#     3. resolved 状态 → join 失败（3033）
#     4. queued 状态 → join 失败（3033）
#     5. pending_close 状态 → join 失败（3033）
#     6. 未被邀请的员工 → join 失败（3034）
#
#   二、邀请 URL 绝对路径校验（Bug 1 修复核心）
#     7. invite_url 以 https:// 开头（包含完整域名）
#        修复前：/h5/?invite=...（无域名，企微卡片无法跳转）
#        修复后：https://itsupport.servyou.com.cn/h5/?invite=...
#
# 设计原则：
#   - mock SessionService 的内部辅助方法和 DB，聚焦业务逻辑
#   - 参考 backend/tests/test_invite_status.py 的测试风格
#   - 不依赖真实数据库 / 真实企微 API
# =============================================================================

from unittest.mock import AsyncMock, MagicMock, patch

import pytest

from app.services.session_service import SessionService
from app.utils.response import AppException


# =============================================================================
# 辅助：构造 mock 会话对象与 SessionService 实例
# =============================================================================

def _mock_conversation(
    status: str = "serving",
    conv_id: str = "conv-001",
    participants: list | None = None,
    assigned_agent_id: str = "agent_001",
    employee_id: str = "emp_001",
):
    """构造一个模拟会话对象，用于测试 join_conversation / invite_participants。

    Args:
        status: 会话状态
        conv_id: 会话ID
        participants: 参与者列表
        assigned_agent_id: 主责坐席ID（invite_participants 权限校验需要）
        employee_id: 会话发起人ID
    """
    conv = MagicMock()
    conv.id = conv_id
    conv.status = status
    conv.assigned_agent_id = assigned_agent_id
    conv.employee_id = employee_id
    conv.participants = participants if participants is not None else []
    conv.updated_at = MagicMock()
    return conv


def _invited_participants():
    """构造被邀请的参与者列表（1 个员工，未加入状态）。"""
    return [
        {"id": "emp_002", "name": "张三", "type": "employee", "joined": False},
    ]


def _make_service():
    """构造 SessionService 实例，所有外部依赖已 mock。

    返回的 service 对象已 mock 以下方法/属性：
      - _get_conversation: AsyncMock（需在测试中设置 return_value）
      - _create_system_message: AsyncMock
      - _broadcast_participant_change: AsyncMock
      - db: MagicMock（add/flush 已 mock）
      - wecom_service: None（join_conversation 不需要）
    """
    db = MagicMock()
    db.add = MagicMock()
    db.flush = AsyncMock()

    service = SessionService(db=db, wecom_service=None)
    service._get_conversation = AsyncMock()
    service._create_system_message = AsyncMock()
    service._broadcast_participant_change = AsyncMock()
    return service


def _make_service_with_wecom():
    """构造带 mock 企微服务的 SessionService 实例（用于邀请 URL 测试）。

    额外 mock：
      - _get_employee_avatar: AsyncMock，返回 None（跳过头像获取）
      - wecom_service: MagicMock，send_card_message 为 AsyncMock
    """
    db = MagicMock()
    db.add = MagicMock()
    db.flush = AsyncMock()

    wecom_service = MagicMock()
    wecom_service.send_card_message = AsyncMock()

    service = SessionService(db=db, wecom_service=wecom_service)
    service._get_conversation = AsyncMock()
    service._get_employee_avatar = AsyncMock(return_value=None)
    service._create_system_message = AsyncMock()
    service._broadcast_participant_change = AsyncMock()
    return service


# =============================================================================
# 一、join_conversation 状态校验测试
# =============================================================================

class TestJoinConversationStatusCheck:
    """join_conversation 状态校验测试。

    修复前：if conversation.status != "serving": → ai_handling 被拒
    修复后：if conversation.status not in ("serving", "ai_handling"):
    """

    # ----- 1.1 ai_handling 状态 → join 成功（本次修复核心） -----

    @pytest.mark.asyncio
    async def test_ai_handling_status_join_succeeds(self):
        """ai_handling 状态下被邀请人加入应成功（本次修复核心）。"""
        # Arrange
        service = _make_service()
        conv = _mock_conversation(
            status="ai_handling",
            participants=_invited_participants(),
        )
        service._get_conversation.return_value = conv

        # Act
        result = await service.join_conversation(
            conversation_id="conv-001",
            employee_id="emp_002",
        )

        # Assert
        assert result is conv
        # joined 状态已更新
        participants = conv.participants
        emp = next(p for p in participants if p["id"] == "emp_002")
        assert emp["joined"] is True
        assert "joined_at" in emp
        # DB 写入被调用
        service.db.add.assert_called_once_with(conv)
        service.db.flush.assert_awaited_once()
        # 系统消息 + WebSocket 广播
        service._create_system_message.assert_awaited_once()
        service._broadcast_participant_change.assert_awaited_once()

    # ----- 1.2 serving 状态 → join 成功（原有功能不回归） -----

    @pytest.mark.asyncio
    async def test_serving_status_join_succeeds(self):
        """serving 状态下被邀请人加入应成功（原有功能不回归）。"""
        # Arrange
        service = _make_service()
        conv = _mock_conversation(
            status="serving",
            participants=_invited_participants(),
        )
        service._get_conversation.return_value = conv

        # Act
        result = await service.join_conversation(
            conversation_id="conv-001",
            employee_id="emp_002",
        )

        # Assert
        assert result is conv
        emp = next(p for p in conv.participants if p["id"] == "emp_002")
        assert emp["joined"] is True
        assert "joined_at" in emp
        service.db.add.assert_called_once_with(conv)
        service.db.flush.assert_awaited_once()

    # ----- 1.3 resolved 状态 → join 失败 -----

    @pytest.mark.asyncio
    async def test_resolved_status_join_fails(self):
        """resolved 状态下加入应失败（已结单不应允许加入）。"""
        # Arrange
        service = _make_service()
        conv = _mock_conversation(
            status="resolved",
            participants=_invited_participants(),
        )
        service._get_conversation.return_value = conv

        # Act & Assert
        with pytest.raises(AppException) as exc_info:
            await service.join_conversation(
                conversation_id="conv-001",
                employee_id="emp_002",
            )
        assert exc_info.value.code == 3033
        # 不应写 DB
        service.db.add.assert_not_called()
        service.db.flush.assert_not_awaited()
        # 不应创建系统消息 / 广播
        service._create_system_message.assert_not_awaited()
        service._broadcast_participant_change.assert_not_awaited()

    # ----- 1.4 queued 状态 → join 失败 -----

    @pytest.mark.asyncio
    async def test_queued_status_join_fails(self):
        """queued 状态下加入应失败（排队中不应允许加入）。"""
        # Arrange
        service = _make_service()
        conv = _mock_conversation(
            status="queued",
            participants=_invited_participants(),
        )
        service._get_conversation.return_value = conv

        # Act & Assert
        with pytest.raises(AppException) as exc_info:
            await service.join_conversation(
                conversation_id="conv-001",
                employee_id="emp_002",
            )
        assert exc_info.value.code == 3033
        service.db.add.assert_not_called()
        service.db.flush.assert_not_awaited()

    # ----- 1.5 pending_close 状态 → join 失败 -----

    @pytest.mark.asyncio
    async def test_pending_close_status_join_fails(self):
        """pending_close 状态下加入应失败（待关单不应允许加入）。"""
        # Arrange
        service = _make_service()
        conv = _mock_conversation(
            status="pending_close",
            participants=_invited_participants(),
        )
        service._get_conversation.return_value = conv

        # Act & Assert
        with pytest.raises(AppException) as exc_info:
            await service.join_conversation(
                conversation_id="conv-001",
                employee_id="emp_002",
            )
        assert exc_info.value.code == 3033
        service.db.add.assert_not_called()
        service.db.flush.assert_not_awaited()


# =============================================================================
# 二、join_conversation 权限校验测试
# =============================================================================

class TestJoinConversationPermissionCheck:
    """join_conversation 权限校验测试。

    规则：只有在 participants 列表中的员工才能加入（被邀请过才能加入）。
    """

    # ----- 2.1 未被邀请的员工 → join 失败（3034） -----

    @pytest.mark.asyncio
    async def test_not_invited_employee_join_fails(self):
        """未被邀请的员工加入应失败（3034 错误码）。"""
        # Arrange
        service = _make_service()
        conv = _mock_conversation(
            status="serving",
            participants=[
                {"id": "emp_002", "name": "张三", "type": "employee", "joined": False},
            ],
        )
        service._get_conversation.return_value = conv

        # Act & Assert — emp_hacker 不在 participants 中
        with pytest.raises(AppException) as exc_info:
            await service.join_conversation(
                conversation_id="conv-001",
                employee_id="emp_hacker",
            )
        assert exc_info.value.code == 3034
        service.db.add.assert_not_called()
        service.db.flush.assert_not_awaited()
        service._create_system_message.assert_not_awaited()

    # ----- 2.2 空参与者列表 → join 失败（3034） -----

    @pytest.mark.asyncio
    async def test_empty_participants_join_fails(self):
        """participants 为空时任何员工加入应失败（3034）。"""
        # Arrange
        service = _make_service()
        conv = _mock_conversation(
            status="ai_handling",
            participants=[],
        )
        service._get_conversation.return_value = conv

        # Act & Assert
        with pytest.raises(AppException) as exc_info:
            await service.join_conversation(
                conversation_id="conv-001",
                employee_id="emp_002",
            )
        assert exc_info.value.code == 3034

    # ----- 2.3 participants 为 None → join 失败（3034） -----

    @pytest.mark.asyncio
    async def test_none_participants_join_fails(self):
        """participants 为 None 时加入应失败（3034）。"""
        # Arrange
        service = _make_service()
        conv = _mock_conversation(
            status="serving",
            participants=None,
        )
        service._get_conversation.return_value = conv

        # Act & Assert
        with pytest.raises(AppException) as exc_info:
            await service.join_conversation(
                conversation_id="conv-001",
                employee_id="emp_002",
            )
        assert exc_info.value.code == 3034


# =============================================================================
# 三、邀请 URL 绝对路径校验（Bug 1 修复核心）
# =============================================================================

class TestInviteUrlAbsolute:
    """invite_participants 邀请 URL 绝对路径校验。

    修复前：invite_url = f"{getattr(self, '_h5_base_url', '')}/h5/" → /h5/?invite=...
    修复后：invite_url = f"{settings.wecom_sso_callback_base.rstrip('/')}/h5/" → https://...
    """

    # ----- 3.1 invite_url 以 https:// 开头（包含完整域名） -----

    @pytest.mark.asyncio
    async def test_invite_url_starts_with_https(self):
        """邀请 URL 应以 https:// 开头（包含完整域名）。

        企微卡片消息要求 URL 必须以 https:// 开头，否则无法跳转。
        """
        # Arrange
        service = _make_service_with_wecom()
        conv = _mock_conversation(status="serving", participants=[])
        service._get_conversation.return_value = conv

        with patch(
            "app.services.session_service.settings"
        ) as mock_settings:
            mock_settings.wecom_sso_callback_base = (
                "https://itsupport.servyou.com.cn"
            )

            # Act
            await service.invite_participants(
                conversation_id="conv-001",
                inviter_agent_id="agent_001",
                participants=[
                    {"id": "emp_002", "name": "张三", "type": "employee"},
                ],
            )

        # Assert
        service.wecom_service.send_card_message.assert_awaited_once()
        call_kwargs = service.wecom_service.send_card_message.call_args.kwargs
        invite_url = call_kwargs["url"]
        assert invite_url.startswith("https://"), (
            f"invite_url 应以 https:// 开头，实际: {invite_url}"
        )

    # ----- 3.2 invite_url 包含 invite 和 eid 参数 -----

    @pytest.mark.asyncio
    async def test_invite_url_contains_invite_and_eid_params(self):
        """邀请 URL 应包含 invite（会话ID）和 eid（员工ID）参数。"""
        # Arrange
        service = _make_service_with_wecom()
        conv = _mock_conversation(status="ai_handling", participants=[])
        service._get_conversation.return_value = conv

        with patch(
            "app.services.session_service.settings"
        ) as mock_settings:
            mock_settings.wecom_sso_callback_base = (
                "https://itsupport.servyou.com.cn"
            )

            # Act
            await service.invite_participants(
                conversation_id="conv-001",
                inviter_agent_id="agent_001",
                participants=[
                    {"id": "emp_002", "name": "张三", "type": "employee"},
                ],
            )

        # Assert
        call_kwargs = service.wecom_service.send_card_message.call_args.kwargs
        invite_url = call_kwargs["url"]
        assert "invite=conv-001" in invite_url
        assert "eid=emp_002" in invite_url

    # ----- 3.3 invite_url 不含尾部多余斜杠 -----

    @pytest.mark.asyncio
    async def test_invite_url_no_trailing_slash_in_domain(self):
        """wecom_sso_callback_base 带尾部斜杠时，invite_url 不应有双斜杠。"""
        # Arrange
        service = _make_service_with_wecom()
        conv = _mock_conversation(status="serving", participants=[])
        service._get_conversation.return_value = conv

        with patch(
            "app.services.session_service.settings"
        ) as mock_settings:
            # base 末尾带斜杠，rstrip 应去除
            mock_settings.wecom_sso_callback_base = (
                "https://itsupport.servyou.com.cn/"
            )

            # Act
            await service.invite_participants(
                conversation_id="conv-001",
                inviter_agent_id="agent_001",
                participants=[
                    {"id": "emp_002", "name": "张三", "type": "employee"},
                ],
            )

        # Assert
        call_kwargs = service.wecom_service.send_card_message.call_args.kwargs
        invite_url = call_kwargs["url"]
        # 不应出现 //h5（base 尾部斜杠未去除会导致双斜杠）
        assert "//h5" not in invite_url, (
            f"invite_url 不应含双斜杠，实际: {invite_url}"
        )
        assert invite_url == (
            "https://itsupport.servyou.com.cn/h5/"
            "?invite=conv-001&eid=emp_002"
        )

    # ----- 3.4 修复前回归：URL 不应为相对路径 -----

    @pytest.mark.asyncio
    async def test_invite_url_not_relative_path(self):
        """邀请 URL 不应为相对路径（/h5/ 开头）。

        回归保障：修复前 invite_url 为 /h5/?invite=...，企微无法跳转。
        """
        # Arrange
        service = _make_service_with_wecom()
        conv = _mock_conversation(status="serving", participants=[])
        service._get_conversation.return_value = conv

        with patch(
            "app.services.session_service.settings"
        ) as mock_settings:
            mock_settings.wecom_sso_callback_base = (
                "https://itsupport.servyou.com.cn"
            )

            # Act
            await service.invite_participants(
                conversation_id="conv-001",
                inviter_agent_id="agent_001",
                participants=[
                    {"id": "emp_002", "name": "张三", "type": "employee"},
                ],
            )

        # Assert
        call_kwargs = service.wecom_service.send_card_message.call_args.kwargs
        invite_url = call_kwargs["url"]
        # 不应以 /h5/ 开头（相对路径）
        assert not invite_url.startswith("/h5/"), (
            f"invite_url 不应为相对路径，实际: {invite_url}"
        )