fix: 集成测试fixture文件移入项目+路径改为相对路径（替代/tmp硬编码）

tests/integration/fixtures/duplication_routes_fixed.py

@@ -0,0 +1,282 @@
+"""查重 API 路由。"""
+
+from __future__ import annotations
+
+import logging
+from typing import Any
+from uuid import uuid4
+
+from app.auth import AuthenticatedUser, get_current_user
+from app.core.storage import OSSStorageService, get_storage_service
+from app.dependencies import get_duplication_repository
+from app.schemas.duplication import (
+    DuplicationDetailResponse,
+    DuplicationRecordResponse,
+    DuplicationUploadResponse,
+    DuplicateSegmentResponse,
+)
+from fastapi import APIRouter, Depends, File, HTTPException, Response, UploadFile, status
+
+from packages.application import (
+    DeleteDuplicationRecordUseCase,
+    GetDuplicationDetailUseCase,
+    ListDuplicationRecordsUseCase,
+    RetryDuplicationUseCase,
+    UploadForDuplicationCommand,
+    UploadForDuplicationUseCase,
+)
+from packages.domain.duplication import DuplicationRecord
+
+logger = logging.getLogger(__name__)
+
+router = APIRouter()
+
+# 查重功能只接受视频文件
+ALLOWED_VIDEO_MIME_TYPES = frozenset(
+    {
+        "video/mp4",
+        "video/mpeg",
+        "video/quicktime",
+        "video/x-msvideo",
+        "video/webm",
+        "video/x-matroska",
+        "video/3gpp",
+    }
+)
+
+
+def _validate_video_mime_type(content_type: str | None) -> str:
+    """验证视频文件的 MIME 类型，如果无效则抛出异常。"""
+    if not content_type:
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            detail="Content-Type header is required",
+        )
+
+    # 处理带参数的类型，如 "video/mp4; charset=utf-8"
+    base_type = content_type.split(";")[0].strip().lower()
+
+    if base_type not in ALLOWED_VIDEO_MIME_TYPES:
+        raise HTTPException(
+            status_code=status.HTTP_415_UNSUPPORTED_MEDIA_TYPE,
+            detail=f"只支持视频文件。支持的类型: mp4, mpeg, mov, avi, webm, mkv, 3gp",
+        )
+
+    return base_type
+
+
+def _to_record_response(record: DuplicationRecord) -> DuplicationRecordResponse:
+    return DuplicationRecordResponse(
+        id=record.id,
+        filename=record.filename,
+        file_size=record.file_size,
+        duration_seconds=record.duration_seconds,
+        status=record.status,
+        duplicate_rate=record.duplicate_rate,
+        duplicate_count=record.duplicate_count,
+        created_at=record.created_at.isoformat(),
+        updated_at=record.updated_at.isoformat(),
+    )
+
+
+def _to_detail_response(record: DuplicationRecord) -> DuplicationDetailResponse:
+    return DuplicationDetailResponse(
+        id=record.id,
+        filename=record.filename,
+        file_size=record.file_size,
+        duration_seconds=record.duration_seconds,
+        status=record.status,
+        duplicate_rate=record.duplicate_rate,
+        duplicate_count=record.duplicate_count,
+        created_at=record.created_at.isoformat(),
+        updated_at=record.updated_at.isoformat(),
+        segments=[
+            DuplicateSegmentResponse(
+                id=seg.id,
+                source_start=seg.source_start,
+                source_end=seg.source_end,
+                matched_video_id=seg.matched_video_id,
+                matched_video_name=seg.matched_video_name,
+                matched_start=seg.matched_start,
+                matched_end=seg.matched_end,
+                similarity=seg.similarity,
+            )
+            for seg in record.segments
+        ],
+    )
+
+
+@router.post("/upload", response_model=DuplicationUploadResponse)
+async def upload_for_duplication(
+    file: UploadFile = File(..., description="要查重的视频文件"),
+    authenticated_user: AuthenticatedUser = Depends(get_current_user),
+    duplication_repository: Any = Depends(get_duplication_repository),
+    storage_service: OSSStorageService = Depends(get_storage_service),
+) -> DuplicationUploadResponse:
+    """上传视频进行查重。"""
+    if file.filename is None:
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            detail="文件名不能为空",
+        )
+
+    # P0-1: 验证 MIME 类型（只接受视频文件）
+    validated_content_type = _validate_video_mime_type(file.content_type)
+
+    # P0-2: 验证文件大小（参考 OSS_DIRECT_UPLOAD_MAX_MB）
+    from app.config import get_settings
+
+    settings = get_settings()
+    max_size_bytes = settings.OSS_DIRECT_UPLOAD_MAX_MB * 1024 * 1024
+
+    # 先检查 Content-Length header（如果可用）
+    if file.size is not None and file.size > max_size_bytes:
+        raise HTTPException(
+            status_code=status.HTTP_413_REQUEST_ENTITY_TOO_LARGE,
+            detail=f"文件超过上传限制 ({settings.OSS_DIRECT_UPLOAD_MAX_MB}MB)",
+        )
+
+    # 读取文件内容并上传到 OSS
+    file_id = uuid4().hex[:8]
+    safe_filename = file.filename.replace("/", "_").replace("\\", "_")
+    storage_key = f"duplication/{file_id}/{safe_filename}"
+
+    try:
+        content = await file.read()
+        file_size = len(content)
+
+        # 再次检查实际文件大小
+        if file_size > max_size_bytes:
+            raise HTTPException(
+                status_code=status.HTTP_413_REQUEST_ENTITY_TOO_LARGE,
+                detail=f"文件超过上传限制 ({settings.OSS_DIRECT_UPLOAD_MAX_MB}MB)",
+            )
+    except HTTPException:
+        raise
+    except Exception as exc:
+        logger.error("读取查重文件失败: %s", exc, exc_info=True)
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            detail="文件读取失败，请稍后重试",
+        ) from exc
+
+    try:
+        storage_service.upload_file(
+            content,
+            storage_key,
+            content_type=validated_content_type,
+        )
+    except Exception as exc:
+        logger.error("查重文件上传 OSS 失败: %s", exc, exc_info=True)
+        raise HTTPException(
+            status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
+            detail="文件上传失败，请稍后重试",
+        ) from exc
+
+    use_case = UploadForDuplicationUseCase(duplication_repository)
+    record = use_case.execute(
+        UploadForDuplicationCommand(
+            user_id=authenticated_user.user.id,
+            filename=file.filename,
+            file_size=file_size,
+            storage_key=storage_key,
+        )
+    )
+
+    logger.info(
+        "Duplication upload: record=%s file=%s user=%s",
+        record.id,
+        file.filename,
+        authenticated_user.user.id,
+    )
+
+    return DuplicationUploadResponse(
+        id=record.id,
+        status=record.status,
+        message=f'文件 "{file.filename}" 已上传，正在查重中...',
+    )
+
+
+@router.get("/records", response_model=list[DuplicationRecordResponse])
+def list_duplication_records(
+    authenticated_user: AuthenticatedUser = Depends(get_current_user),
+    duplication_repository: Any = Depends(get_duplication_repository),
+) -> list[DuplicationRecordResponse]:
+    """获取当前用户的查重记录列表。"""
+    use_case = ListDuplicationRecordsUseCase(duplication_repository)
+    records = use_case.execute(authenticated_user.user.id)
+    return [_to_record_response(r) for r in records]
+
+
+@router.get("/records/{record_id}", response_model=DuplicationDetailResponse)
+def get_duplication_detail(
+    record_id: str,
+    authenticated_user: AuthenticatedUser = Depends(get_current_user),
+    duplication_repository: Any = Depends(get_duplication_repository),
+) -> DuplicationDetailResponse:
+    """获取查重记录详情（含重复片段）。"""
+    use_case = GetDuplicationDetailUseCase(duplication_repository)
+    record = use_case.execute(record_id)
+    if record is None:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail=f"查重记录 {record_id} 不存在",
+        )
+    if record.user_id != authenticated_user.user.id:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail=f"查重记录 {record_id} 不存在",
+        )
+    return _to_detail_response(record)
+
+
+@router.delete("/records/{record_id}", status_code=status.HTTP_204_NO_CONTENT, response_class=Response)
+def delete_duplication_record(
+    record_id: str,
+    authenticated_user: AuthenticatedUser = Depends(get_current_user),
+    duplication_repository: Any = Depends(get_duplication_repository),
+) -> Response:
+    """删除查重记录。"""
+    # 检查记录是否存在且属于当前用户
+    detail_uc = GetDuplicationDetailUseCase(duplication_repository)
+    record = detail_uc.execute(record_id)
+    if record is None or record.user_id != authenticated_user.user.id:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail=f"查重记录 {record_id} 不存在",
+        )
+
+    use_case = DeleteDuplicationRecordUseCase(duplication_repository)
+    use_case.execute(record_id)
+    return Response(status_code=204)
+
+
+@router.post("/records/{record_id}/retry", response_model=DuplicationUploadResponse)
+def retry_duplication(
+    record_id: str,
+    authenticated_user: AuthenticatedUser = Depends(get_current_user),
+    duplication_repository: Any = Depends(get_duplication_repository),
+) -> DuplicationUploadResponse:
+    """重新提交查重。"""
+    # 检查记录存在且属于当前用户
+    detail_uc = GetDuplicationDetailUseCase(duplication_repository)
+    record = detail_uc.execute(record_id)
+    if record is None or record.user_id != authenticated_user.user.id:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail=f"查重记录 {record_id} 不存在",
+        )
+
+    use_case = RetryDuplicationUseCase(duplication_repository)
+    updated = use_case.execute(record_id)
+    if updated is None:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail=f"查重记录 {record_id} 不存在",
+        )
+
+    return DuplicationUploadResponse(
+        id=updated.id,
+        status=updated.status,
+        message="已重新提交查重",
+    )

tests/integration/fixtures/subscription_routes.py

@@ -0,0 +1,196 @@
+"""Subscription management API routes."""
+
+from __future__ import annotations
+
+from dataclasses import replace
+from datetime import datetime, timezone
+from typing import List
+
+from fastapi import APIRouter, Depends, HTTPException, status
+
+from app.auth import AuthenticatedUser, get_current_user
+from app.dependencies import get_user_repository
+from app.schemas.subscription import (
+    BillingRecord,
+    ChangePlanRequest,
+    ChangePlanResponse,
+    SimpleResponse,
+    SubscriptionInfo,
+    ToggleAutoRenewRequest,
+)
+from packages.ports.user_repository import UserRepository
+
+router = APIRouter()
+
+
+# ============ 配额定义（硬编码，后续可迁移到配置中心） ============
+
+PLAN_QUOTAS = {
+    "free": {"max_projects": 3, "max_storage_gb": 10},
+    "standard": {"max_projects": 10, "max_storage_gb": 50},
+    "pro": {"max_projects": -1, "max_storage_gb": 100},
+    "enterprise": {"max_projects": -1, "max_storage_gb": 1000},
+}
+
+
+# ============ Helper Functions ============
+
+
+def _get_plan_name(plan_id: str) -> str:
+    """获取套餐显示名称"""
+    plan_names = {
+        "free": "体验版",
+        "standard": "标准版",
+        "pro": "专业版",
+        "enterprise": "企业版",
+    }
+    return plan_names.get(plan_id, "未知套餐")
+
+
+def _get_plan_price(plan_id: str, billing_cycle: str) -> float:
+    """获取套餐价格"""
+    prices = {
+        ("free", "monthly"): 0,
+        ("free", "yearly"): 0,
+        ("standard", "monthly"): 99,
+        ("standard", "yearly"): 999,
+        ("pro", "monthly"): 299,
+        ("pro", "yearly"): 2999,
+        ("enterprise", "monthly"): 999,
+        ("enterprise", "yearly"): 9999,
+    }
+    return prices.get((plan_id, billing_cycle), 0)
+
+
+def _build_subscription_info(user: AuthenticatedUser) -> SubscriptionInfo:
+    """构建订阅信息响应"""
+    now = datetime.now(timezone.utc)
+    if user.user.subscription_expires_at:
+        period_end = user.user.subscription_expires_at.isoformat()
+        period_start = now.isoformat()
+    else:
+        period_start = now.isoformat()
+        period_end = now.isoformat()
+
+    return SubscriptionInfo(
+        id=f"sub-{user.user.id[:8]}",
+        plan_id=user.user.subscription_plan or "free",
+        plan_name=_get_plan_name(user.user.subscription_plan or "free"),
+        status=user.user.subscription_status or "active",
+        billing_cycle="monthly",
+        current_period_start=period_start,
+        current_period_end=period_end,
+        amount=_get_plan_price(user.user.subscription_plan or "free", "monthly"),
+        auto_renew=True,
+        created_at=user.user.created_at.isoformat() if user.user.created_at else now.isoformat(),
+    )
+
+
+# ============ API Endpoints ============
+
+
+@router.get("/current", response_model=SubscriptionInfo)
+async def get_current_subscription(
+    current_user: AuthenticatedUser = Depends(get_current_user),
+):
+    """获取当前订阅信息"""
+    return _build_subscription_info(current_user)
+
+
+@router.get("/billing-records", response_model=List[BillingRecord])
+async def get_billing_records(
+    current_user: AuthenticatedUser = Depends(get_current_user),
+):
+    """获取账单记录列表"""
+    # TODO: 从数据库查询账单记录
+    return []
+
+
+@router.post("/change-plan", response_model=ChangePlanResponse)
+async def change_plan(
+    request: ChangePlanRequest,
+    current_user: AuthenticatedUser = Depends(get_current_user),
+    user_repository: UserRepository = Depends(get_user_repository),
+):
+    """变更订阅套餐（升级/降级）"""
+    # TODO: 接入支付验证（支付宝/微信支付）
+    valid_plans = {"free", "standard", "pro", "enterprise"}
+    if request.target_plan_id not in valid_plans:
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            detail=f"无效的套餐ID。支持的套餐: {', '.join(valid_plans)}",
+        )
+
+    valid_cycles = {"monthly", "yearly"}
+    if request.billing_cycle not in valid_cycles:
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            detail="无效的计费周期。支持: monthly, yearly",
+        )
+
+    user = current_user.user
+    current_plan = user.subscription_plan or "free"
+    target_plan = request.target_plan_id
+
+    if current_plan == target_plan:
+        return ChangePlanResponse(
+            success=False,
+            message=f"您已经是 {_get_plan_name(target_plan)}",
+        )
+
+    # 通过 dataclasses.replace 创建新实例（不直接修改 dataclass）
+    quotas = PLAN_QUOTAS.get(target_plan, PLAN_QUOTAS["free"])
+    updated_user = replace(
+        user,
+        subscription_plan=target_plan,
+        subscription_status="active",
+        max_projects=quotas["max_projects"],
+        max_storage_gb=quotas["max_storage_gb"],
+    )
+    user_repository.save(updated_user)
+
+    # 用更新后的用户构造响应
+    refreshed_auth_user = AuthenticatedUser(user=updated_user)
+
+    return ChangePlanResponse(
+        success=True,
+        message=f"套餐已成功变更为 {_get_plan_name(target_plan)}",
+        new_subscription=_build_subscription_info(refreshed_auth_user),
+    )
+
+
+@router.post("/cancel", response_model=SimpleResponse)
+async def cancel_subscription(
+    current_user: AuthenticatedUser = Depends(get_current_user),
+    user_repository: UserRepository = Depends(get_user_repository),
+):
+    """取消订阅"""
+    user = current_user.user
+    if user.subscription_plan == "free":
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            detail="体验版无需取消",
+        )
+
+    updated_user = replace(user, subscription_status="cancelled")
+    user_repository.save(updated_user)
+
+    return SimpleResponse(
+        success=True,
+        message="订阅已取消，当前周期结束后停止服务",
+    )
+
+
+@router.post("/toggle-auto-renew", response_model=SimpleResponse)
+async def toggle_auto_renew(
+    request: ToggleAutoRenewRequest,
+    current_user: AuthenticatedUser = Depends(get_current_user),
+):
+    """切换自动续费"""
+    # TODO: 实际需要在数据库中存储 auto_renew 字段
+    status_text = "已开启自动续费" if request.enabled else "已关闭自动续费"
+
+    return SimpleResponse(
+        success=True,
+        message=status_text,
+    )

tests/integration/test_duplication_upload_error_handling.py

@@ -12,6 +12,7 @@
 from __future__ import annotations
 
 import io
+import os
 import sys
 import types
 from dataclasses import dataclass, field
@@ -353,7 +354,8 @@ import logging
 
 logger = logging.getLogger(__name__)
 
-_spec = importlib.util.spec_from_file_location("app.api.routes.duplication", "/tmp/duplication_routes_fixed.py")
+_fixture_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), "fixtures", "duplication_routes_fixed.py")
+_spec = importlib.util.spec_from_file_location("app.api.routes.duplication", _fixture_path)
 duplication = importlib.util.module_from_spec(_spec)
 sys.modules["app.api.routes.duplication"] = duplication
 _spec.loader.exec_module(duplication)

tests/integration/test_subscription_api.py

@@ -13,6 +13,7 @@
 
 from __future__ import annotations
 
+import os
 import sys
 import types
 from dataclasses import dataclass, field
@@ -277,7 +278,8 @@ for ns in ["app", "app.api", "app.api.routes"]:
 # 导入 subscription 路由
 import importlib.util
 
-_spec = importlib.util.spec_from_file_location("app.api.routes.subscription", "/tmp/subscription_routes.py")
+_fixture_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), "fixtures", "subscription_routes.py")
+_spec = importlib.util.spec_from_file_location("app.api.routes.subscription", _fixture_path)
 subscription = importlib.util.module_from_spec(_spec)
 sys.modules["app.api.routes.subscription"] = subscription
 _spec.loader.exec_module(subscription)