xiaoxia-saas/tests/integration/test_error_scenarios.py

"""
错误场景集成测试

覆盖：
- 401 未授权（无 token、无效 token、过期 token）
- 403 禁止访问（无权限资源）
- 404 不存在资源
- 422 参数校验失败（缺少字段、类型错误、格式错误）
- 并发请求处理
- 大数据量请求

使用内存数据库（USE_IN_MEMORY_DB=True）即可运行，无需外部 PostgreSQL。
"""

from __future__ import annotations

import os
import sys
import uuid
from concurrent.futures import ThreadPoolExecutor, as_completed
from pathlib import Path

import pytest

ROOT = Path(__file__).resolve().parents[2]
if str(ROOT) not in sys.path:
    sys.path.insert(0, str(ROOT))

os.environ.setdefault("JWT_SECRET_KEY", "test-secret-key-for-all-tests")
os.environ.setdefault("USE_IN_MEMORY_DB", "True")

from fastapi.testclient import TestClient

from apps.api.main import app

client = TestClient(app)


def _reset_rate_limiter():
    """重置限流中间件状态，避免测试间互相影响。"""
    node = getattr(app, "middleware_stack", None)
    while node is not None:
        if hasattr(node, "requests"):
            node.requests.clear()
            break
        node = getattr(node, "app", None)


@pytest.fixture(autouse=True)
def _clear_rate_limit_between_tests():
    """每个测试前清空限流计数器。"""
    _reset_rate_limiter()
    yield


# ---------------------------------------------------------------------------
# Fixtures
# ---------------------------------------------------------------------------


@pytest.fixture
def auth_headers():
    """创建测试用户并返回认证 headers。"""
    import time

    unique = uuid.uuid4().hex[:8]
    email = f"errtest-{unique}@example.com"
    username = f"errtest-{unique}"

    reg = client.post(
        "/api/v1/auth/register",
        json={
            "email": email,
            "password": "SecurePass123",
            "username": username,
            "display_name": "Error Test User",
        },
    )
    assert reg.status_code in (200, 201), f"注册失败: {reg.text}"

    # 登录可能触发限流（429），最多重试 5 次，每次等待更久
    login = None
    for _attempt in range(5):
        login = client.post(
            "/api/v1/auth/login",
            json={"email": email, "password": "SecurePass123"},
        )
        if login.status_code != 429:
            break
        time.sleep(8)
    if login.status_code == 429:
        pytest.skip("登录端点限流，跳过需要认证的测试")
    assert login.status_code == 200, f"登录失败: {login.text}"

    token = login.json()["access_token"]
    return {"Authorization": f"Bearer {token}"}


@pytest.fixture
def other_auth_headers():
    """创建第二个测试用户（用于权限隔离测试）。"""
    import time

    unique = uuid.uuid4().hex[:8]
    email = f"errtest-other-{unique}@example.com"
    username = f"errother-{unique}"

    client.post(
        "/api/v1/auth/register",
        json={
            "email": email,
            "password": "SecurePass123",
            "username": username,
            "display_name": "Other User",
        },
    )

    login = None
    for _attempt in range(5):
        login = client.post(
            "/api/v1/auth/login",
            json={"email": email, "password": "SecurePass123"},
        )
        if login.status_code != 429:
            break
        time.sleep(8)
    if login.status_code == 429:
        pytest.skip("登录端点限流，跳过需要认证的测试")
    token = login.json()["access_token"]
    return {"Authorization": f"Bearer {token}"}


# ---------------------------------------------------------------------------
# 401 未授权
# ---------------------------------------------------------------------------


class TestUnauthorized401:
    """测试 401 未授权场景。"""

    def test_access_protected_endpoint_without_token(self):
        """无 token 访问受保护端点应返回 401 或 403。"""
        response = client.get("/api/v1/auth/me")
        assert response.status_code in [401, 403]

    def test_access_projects_without_token(self):
        """无 token 访问项目列表应返回 401 或 403。"""
        response = client.get("/api/v1/projects")
        assert response.status_code in [401, 403]

    def test_access_with_invalid_token(self):
        """无效 token 应返回 401。"""
        response = client.get(
            "/api/v1/auth/me",
            headers={"Authorization": "Bearer invalid.token.value"},
        )
        assert response.status_code in [401, 403]

    def test_access_with_malformed_bearer(self):
        """格式错误的 Bearer 应返回 401 或 403。"""
        response = client.get(
            "/api/v1/auth/me",
            headers={"Authorization": "NotBearer token"},
        )
        assert response.status_code in [401, 403]

    def test_access_with_empty_token(self):
        """空 token 应返回 401 或 403。"""
        response = client.get(
            "/api/v1/auth/me",
            headers={"Authorization": "Bearer "},
        )
        assert response.status_code in [401, 403]

    def test_login_with_wrong_password(self):
        """错误密码登录应返回 401。"""
        import time

        unique = uuid.uuid4().hex[:8]
        client.post(
            "/api/v1/auth/register",
            json={
                "email": f"wrongpwd-{unique}@example.com",
                "password": "SecurePass123",
                "username": f"wrongpwd-{unique}",
            },
        )
        response = None
        for _attempt in range(3):
            response = client.post(
                "/api/v1/auth/login",
                json={
                    "email": f"wrongpwd-{unique}@example.com",
                    "password": "WrongPassword999!",
                },
            )
            if response.status_code != 429:
                break
            time.sleep(8)
        if response.status_code == 429:
            pytest.skip("登录端点限流")
        assert response.status_code == 401

    def test_login_with_nonexistent_email(self):
        """不存在的用户登录应返回 401。"""
        import time

        response = None
        for _attempt in range(3):
            response = client.post(
                "/api/v1/auth/login",
                json={
                    "email": f"ghost-{uuid.uuid4().hex[:8]}@nonexist.com",
                    "password": "AnyPassword123",
                },
            )
            if response.status_code != 429:
                break
            time.sleep(8)
        if response.status_code == 429:
            pytest.skip("登录端点限流")
        assert response.status_code == 401

    def test_create_project_without_auth(self):
        """未认证创建项目应返回 401 或 403。"""
        response = client.post(
            "/api/v1/projects",
            json={"name": "Unauthorized Project"},
        )
        assert response.status_code in [401, 403]


# ---------------------------------------------------------------------------
# 403 禁止访问
# ---------------------------------------------------------------------------


class TestForbidden403:
    """测试 403 禁止访问场景。"""

    def test_access_other_user_project(self, auth_headers, other_auth_headers):
        """访问他人项目应返回 403。"""
        # 用户 A 创建项目
        created = client.post(
            "/api/v1/projects",
            json={"name": "Private Project"},
            headers=auth_headers,
        )
        assert created.status_code == 200, f"创建项目失败: {created.text}"
        project_id = created.json()["id"]

        # 用户 B 尝试访问
        response = client.get(
            f"/api/v1/projects/{project_id}",
            headers=other_auth_headers,
        )
        assert response.status_code == 403, f"访问他人项目应返回 403，实际: {response.status_code}"

    def test_delete_other_user_project(self, auth_headers, other_auth_headers):
        """删除他人项目应返回 403。"""
        created = client.post(
            "/api/v1/projects",
            json={"name": "Do Not Delete"},
            headers=auth_headers,
        )
        assert created.status_code == 200
        project_id = created.json()["id"]

        response = client.delete(
            f"/api/v1/projects/{project_id}",
            headers=other_auth_headers,
        )
        assert response.status_code == 403, f"删除他人项目应返回 403，实际: {response.status_code}"


# ---------------------------------------------------------------------------
# 404 不存在资源
# ---------------------------------------------------------------------------


class TestNotFound404:
    """测试 404 不存在资源场景。"""

    def test_get_nonexistent_project(self, auth_headers):
        """获取不存在的项目应返回 404。"""
        response = client.get(
            "/api/v1/projects/nonexistent-project-id-99999",
            headers=auth_headers,
        )
        assert response.status_code == 404

    def test_get_nonexistent_asset(self, auth_headers):
        """获取不存在的资产应返回 404。"""
        response = client.get(
            "/api/v1/assets/nonexistent-asset-id-99999",
            headers=auth_headers,
        )
        assert response.status_code == 404

    def test_unknown_api_endpoint(self, auth_headers):
        """访问不存在的 API 端点应返回 404。"""
        response = client.get(
            "/api/v1/nonexistent-endpoint",
            headers=auth_headers,
        )
        assert response.status_code == 404

    def test_get_nonexistent_user_profile(self, auth_headers):
        """获取不存在的用户信息应返回 404。"""
        response = client.get(
            "/api/v1/users/nonexistent-user-id",
            headers=auth_headers,
        )
        assert response.status_code in [404, 405]


# ---------------------------------------------------------------------------
# 422 参数校验失败
# ---------------------------------------------------------------------------


class TestValidation422:
    """测试 422 参数校验失败场景。"""

    def test_register_with_invalid_email_format(self):
        """无效邮箱格式注册应返回 422。"""
        response = client.post(
            "/api/v1/auth/register",
            json={
                "email": "not-an-email",
                "password": "SecurePass123",
                "username": "bademail",
            },
        )
        assert response.status_code in [400, 422]

    def test_register_with_weak_password(self):
        """弱密码注册应返回 400 或 422。"""
        response = client.post(
            "/api/v1/auth/register",
            json={
                "email": f"weakpwd-{uuid.uuid4().hex[:8]}@example.com",
                "password": "123",
                "username": f"weakpwd-{uuid.uuid4().hex[:8]}",
            },
        )
        assert response.status_code in [400, 422]

    def test_register_with_empty_body(self):
        """空注册请求体应返回 422。"""
        response = client.post(
            "/api/v1/auth/register",
            json={},
        )
        assert response.status_code == 422

    def test_login_with_missing_fields(self):
        """登录缺少字段应返回 422。"""
        response = client.post(
            "/api/v1/auth/login",
            json={"email": "test@example.com"},
        )
        assert response.status_code == 422

    def test_create_project_with_empty_name(self, auth_headers):
        """创建项目空名称应返回 422。"""
        response = client.post(
            "/api/v1/projects",
            json={"name": ""},
            headers=auth_headers,
        )
        assert response.status_code in [400, 422]

    def test_create_project_with_missing_name(self, auth_headers):
        """创建项目缺少名称应返回 422。"""
        response = client.post(
            "/api/v1/projects",
            json={"description": "No name provided"},
            headers=auth_headers,
        )
        assert response.status_code in [400, 422]

    def test_change_subscription_with_invalid_plan(self, auth_headers):
        """变更无效套餐应返回 400 或 422。"""
        response = client.post(
            "/api/v1/subscription/change-plan",
            json={
                "target_plan_id": "invalid_plan_xyz",
                "billing_cycle": "monthly",
            },
            headers=auth_headers,
        )
        assert response.status_code in [400, 422]

    def test_toggle_auto_renew_missing_field(self, auth_headers):
        """切换自动续费缺少 enabled 字段应返回 422。"""
        response = client.post(
            "/api/v1/subscription/toggle-auto-renew",
            json={},
            headers=auth_headers,
        )
        assert response.status_code == 422

    def test_register_with_duplicate_email(self):
        """重复邮箱注册应返回 400。"""
        unique = uuid.uuid4().hex[:8]
        email = f"dup-{unique}@example.com"

        client.post(
            "/api/v1/auth/register",
            json={
                "email": email,
                "password": "SecurePass123",
                "username": f"user1-{unique}",
            },
        )

        response = client.post(
            "/api/v1/auth/register",
            json={
                "email": email,
                "password": "SecurePass123",
                "username": f"user2-{unique}",
            },
        )
        assert response.status_code in [400, 409]


# ---------------------------------------------------------------------------
# 并发请求处理
# ---------------------------------------------------------------------------


class TestConcurrentRequests:
    """测试并发请求处理。"""

    def test_concurrent_project_creation(self, auth_headers):
        """并发创建多个项目应都能成功。"""

        def create_project(idx: int):
            resp = client.post(
                "/api/v1/projects",
                json={"name": f"Concurrent Project {idx}-{uuid.uuid4().hex[:4]}"},
                headers=auth_headers,
            )
            return resp.status_code

        with ThreadPoolExecutor(max_workers=5) as executor:
            futures = [executor.submit(create_project, i) for i in range(5)]
            results = [f.result() for f in as_completed(futures)]

        success_count = sum(1 for s in results if s == 200)
        # 至少部分请求应成功（可能受配额限制）
        assert success_count >= 1, f"并发创建项目应至少成功 1 个，实际: {results}"

    def test_concurrent_login_same_user(self):
        """同一用户并发登录应都能成功。"""
        unique = uuid.uuid4().hex[:8]
        email = f"concurrent-{unique}@example.com"
        username = f"concurrent-{unique}"

        client.post(
            "/api/v1/auth/register",
            json={
                "email": email,
                "password": "SecurePass123",
                "username": username,
            },
        )

        def login():
            resp = client.post(
                "/api/v1/auth/login",
                json={"email": email, "password": "SecurePass123"},
            )
            return resp.status_code

        with ThreadPoolExecutor(max_workers=5) as executor:
            futures = [executor.submit(login) for _ in range(5)]
            results = [f.result() for f in as_completed(futures)]

        # 并发登录可能触发限流（429），应返回 200 或 429，不应 500
        assert all(s in (200, 429) for s in results), f"并发登录应返回 200 或 429，实际: {results}"


# ---------------------------------------------------------------------------
# 大数据量请求
# ---------------------------------------------------------------------------


class TestLargeDataRequests:
    """测试大数据量请求处理。"""

    def test_create_project_with_very_long_name(self, auth_headers):
        """超长项目名称应返回 422 或截断处理。"""
        long_name = "A" * 10000
        response = client.post(
            "/api/v1/projects",
            json={"name": long_name, "description": "Long name test"},
            headers=auth_headers,
        )
        # 应返回 422（超过长度限制）或 400
        assert response.status_code in [400, 413, 422], f"超长名称应被拒绝，实际: {response.status_code}"

    def test_create_project_with_large_description(self, auth_headers):
        """超大描述应能处理（或拒绝）。"""
        large_desc = "B" * 100000
        response = client.post(
            "/api/v1/projects",
            json={"name": "Large Desc Test", "description": large_desc},
            headers=auth_headers,
        )
        # 可能被接受或被拒绝，但不应 500
        assert response.status_code < 500, f"超大描述不应导致 500，实际: {response.status_code}"

    def test_register_with_oversized_payload(self):
        """超大注册请求体应返回 413 或 422，而非 500。"""
        huge_payload = {
            "email": f"huge-{uuid.uuid4().hex[:8]}@example.com",
            "password": "SecurePass123",
            "username": f"huge-{uuid.uuid4().hex[:8]}",
            "extra_field": "X" * 100000,
        }
        response = client.post(
            "/api/v1/auth/register",
            json=huge_payload,
        )
        assert response.status_code < 500, f"超大请求体不应导致 500，实际: {response.status_code}"

    def test_rapid_sequential_requests(self, auth_headers):
        """快速连续请求不应触发限流导致 500。"""
        statuses = []
        for _ in range(20):
            resp = client.get("/api/v1/projects", headers=auth_headers)
            statuses.append(resp.status_code)

        # 所有请求应返回正常状态码（200 或限流 429），不应 500
        assert all(s < 500 for s in statuses), f"快速连续请求不应产生 500，状态码: {statuses}"


if __name__ == "__main__":
    pytest.main([__file__, "-v", "--timeout=60"])