wecom_it_smart_desk/backend/tests/test_agents.py

# =============================================================================
# 企微智能IT支持服务台 — 坐席降级登录测试
# =============================================================================
# 覆盖 P0 修复 Fix-4: 企微 API 不可达时,已注册坐席必须验证本地密码
# 创建日期: 2026-06-15 (Claude Code 补最小测试,因 WB 提交时未含此测试)
# =============================================================================

import pytest
import pytest_asyncio
from unittest.mock import AsyncMock, patch

from app.models.agent import Agent
from app.utils.error_codes import ErrorCode
from tests.conftest import create_test_agent


class TestAgentDegradedLogin:
    """P0 修复 Fix-4: 降级登录密码验证"""

    @pytest.mark.asyncio
    async def test_degraded_login_wrong_password_rejected(
        self, client, db_session, mock_redis, mock_wecom_instance
    ):
        """场景: 企微 API 不可达,坐席有 password_hash,登录用错密码 → 拒绝

        验证:
        - 状态码非 200(或响应 code 非 0)
        - 错误码属于 AUTH_PASSWORD_WRONG 类(1011 当前,2006 改完后)
        """
        # 1. 预置坐席:有 password_hash
        import bcrypt

        correct_pw = "CorrectP@ss123"
        pw_hash = bcrypt.hashpw(correct_pw.encode("utf-8"), bcrypt.gensalt()).decode(
            "utf-8"
        )

        agent = create_test_agent(
            user_id="degraded_agent_001",
            name="降级坐席",
        )
        agent.password_hash = pw_hash
        db_session.add(agent)
        await db_session.flush()

        # 2. 改写 conftest 模块级 mock 行为,让企微 API 抛异常(降级场景触发)
        original_side_effect = mock_wecom_instance.get_user_info.side_effect

        async def fail_get_user_info(*args, **kwargs):
            raise Exception("企微 API 不可达 - 验证降级路径")

        mock_wecom_instance.get_user_info.side_effect = fail_get_user_info

        try:
            # 3. 用错误密码登录
            response = await client.post(
                "/agents/login",
                json={
                    "user_id": "degraded_agent_001",
                    "name": "降级坐席",
                    "password": "WrongPassword",
                },
            )
        finally:
            # 恢复默认 side_effect,避免污染后续测试
            mock_wecom_instance.get_user_info.side_effect = original_side_effect

        # 4. 断言:被拒绝
        assert response.status_code in (200, 401, 403), (
            f"预期被拒绝,实际 status={response.status_code}, body={response.text}"
        )
        body = response.json()
        # 业务 code 应该非 0
        assert body.get("code") != 0, f"预期失败 code,实际成功: {body}"

        # 错误码: WB 修复后是 AUTH_PASSWORD_WRONG=2006,旧码 1011 也接受
        error_code = body.get("code")
        assert error_code in (
            ErrorCode.AUTH_PASSWORD_WRONG.value,  # 2006
            1011,  # 旧数字码,WB 接入 ErrorCode 前的过渡
        ), f"错误码不匹配: {error_code}, body={body}"

    @pytest.mark.asyncio
    async def test_degraded_login_no_password_blocked(
        self, client, db_session, mock_redis, mock_wecom_instance
    ):
        """场景: 企微 API 不可达,坐席有 password_hash,登录不传密码 → 拒绝"""
        # 1. 预置坐席
        import bcrypt

        pw_hash = bcrypt.hashpw(b"AnyP@ss", bcrypt.gensalt()).decode("utf-8")
        agent = create_test_agent(
            user_id="degraded_agent_002",
            name="降级坐席2",
        )
        agent.password_hash = pw_hash
        db_session.add(agent)
        await db_session.flush()

        # 2. 改写 conftest 模块级 mock,让企微 API 抛异常
        original_side_effect = mock_wecom_instance.get_user_info.side_effect

        async def fail_get_user_info(*args, **kwargs):
            raise Exception("企微 API 不可达 - 验证降级路径")

        mock_wecom_instance.get_user_info.side_effect = fail_get_user_info

        try:
            # 3. 不传 password 登录
            response = await client.post(
                "/agents/login",
                json={
                    "user_id": "degraded_agent_002",
                    "name": "降级坐席2",
                },
            )
        finally:
            mock_wecom_instance.get_user_info.side_effect = original_side_effect

        # 4. 断言:被拒绝
        body = response.json()
        assert body.get("code") != 0, f"预期被拒绝: {body}"
        error_code = body.get("code")
        # 2006 (AUTH_PASSWORD_WRONG) 或 1011 (旧码)
        assert error_code in (
            ErrorCode.AUTH_PASSWORD_WRONG.value,
            1011,
        ), f"错误码不匹配: {error_code}, body={body}"

    @pytest.mark.asyncio
    async def test_degraded_login_correct_password_succeeds(
        self, client, db_session, mock_redis, mock_wecom_instance
    ):
        """场景: 企微 API 不可达,坐席有 password_hash,登录用对密码 → 成功

        验证降级路径正常工作时,正确密码可以登录
        """
        import bcrypt

        correct_pw = "CorrectP@ss456"
        pw_hash = bcrypt.hashpw(correct_pw.encode("utf-8"), bcrypt.gensalt()).decode(
            "utf-8"
        )

        agent = create_test_agent(
            user_id="degraded_agent_003",
            name="降级坐席3",
        )
        agent.password_hash = pw_hash
        db_session.add(agent)
        await db_session.flush()

        # 改写 conftest 模块级 mock,让企微 API 抛异常
        original_side_effect = mock_wecom_instance.get_user_info.side_effect

        async def fail_get_user_info(*args, **kwargs):
            raise Exception("企微 API 不可达 - 验证降级路径")

        mock_wecom_instance.get_user_info.side_effect = fail_get_user_info

        try:
            response = await client.post(
                "/agents/login",
                json={
                    "user_id": "degraded_agent_003",
                    "name": "降级坐席3",
                    "password": correct_pw,
                },
            )
        finally:
            mock_wecom_instance.get_user_info.side_effect = original_side_effect

        # 降级 + 正确密码应能登录
        body = response.json()
        assert body.get("code") == 0, (
            f"预期降级登录成功,实际失败: {body}"
        )
        assert "token" in body.get("data", {}), (
            f"响应缺 token: {body}"
        )