xiaoxia-saas/tests/integration/test_duplication_api.py

"""查重 API 集成测试。

覆盖端点：
- GET  /records — 列表查询（含分页）
- GET  /records/{record_id} — 详情查询
- DELETE /records/{record_id} — 删除记录
- POST /records/{record_id}/retry — 重试查重

使用 FastAPI TestClient + dependency_overrides 模式，
导入真实模块，不创建 fake namespace packages，避免 sys.modules 污染。
"""

from __future__ import annotations

import os
import sys
from datetime import datetime, timezone
from typing import Any
from unittest.mock import MagicMock
from uuid import uuid4

# ── 环境变量 & sys.path（必须在导入 app.* 之前设置） ──────────────────────────
os.environ.setdefault("JWT_SECRET_KEY", "unit-test-secret-key-for-testing")
os.environ.setdefault("DATABASE_URL", "sqlite:///test.db")

import pytest
from fastapi import FastAPI
from fastapi.testclient import TestClient

sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "..", "apps", "api"))

from app.api.routes.duplication import router
from app.auth import AuthenticatedUser, get_current_user
from app.core.storage import get_storage_service
from app.dependencies import get_duplication_repository

from packages.domain.duplication import DuplicateSegment, DuplicationRecord

# ── 导入真实模块（不创建 fake module） ────────────────────────────────────────
from packages.domain.entities import User

# ---------------------------------------------------------------------------
# 1. 内存 Repository + 辅助函数
# ---------------------------------------------------------------------------


class InMemoryDuplicationRepo:
    """内存中的查重记录 Repository，模拟持久化行为。"""

    def __init__(self):
        self.records: dict[str, DuplicationRecord] = {}

    def create(self, record: DuplicationRecord) -> DuplicationRecord:
        self.records[record.id] = record
        return record

    def get(self, record_id: str) -> DuplicationRecord | None:
        return self.records.get(record_id)

    def list_by_user(self, user_id: str, *, offset: int = 0, limit: int = 50) -> list[DuplicationRecord]:
        all_records = [r for r in self.records.values() if r.user_id == user_id]
        return all_records[offset : offset + limit]

    def update(self, record: DuplicationRecord) -> DuplicationRecord:
        self.records[record.id] = record
        return record

    def delete(self, record_id: str) -> bool:
        if record_id in self.records:
            del self.records[record_id]
            return True
        return False


def _make_user(**overrides) -> User:
    defaults = dict(
        id="user-test-001",
        email="test@example.com",
        display_name="Test User",
        username="testuser",
        subscription_plan="free",
        subscription_status="active",
        max_projects=3,
        max_storage_gb=10,
        created_at=datetime(2026, 1, 1, tzinfo=timezone.utc),
    )
    defaults.update(overrides)
    return User(**defaults)


def _make_record(
    user_id: str = "user-test-001",
    status: str = "pending",
    filename: str = "test.mp4",
    **kw: Any,
) -> DuplicationRecord:
    """创建测试用 DuplicationRecord 并设置状态。"""
    record = DuplicationRecord(
        id=uuid4().hex,
        user_id=user_id,
        filename=filename,
        file_size=kw.get("file_size", 1024),
        storage_key=kw.get("storage_key", "oss/key"),
        duration_seconds=kw.get("duration", 30.0),
    )
    if status == "processing":
        record.mark_processing()
    elif status == "completed":
        record.mark_processing()
        record.mark_completed(duplicate_rate=15.0, duplicate_count=1, segments=[])
    elif status == "failed":
        record.mark_processing()
        record.mark_failed("处理失败")
    return record


# ---------------------------------------------------------------------------
# 2. Fixtures
# ---------------------------------------------------------------------------


@pytest.fixture
def repo():
    return InMemoryDuplicationRepo()


@pytest.fixture
def client(repo):
    """创建带有依赖覆盖的 TestClient。"""
    test_app = FastAPI()
    test_app.include_router(router)

    def _override_current_user():
        return AuthenticatedUser(user=_make_user())

    def _override_dup_repo():
        return repo

    def _override_storage():
        return MagicMock()

    test_app.dependency_overrides[get_current_user] = _override_current_user
    test_app.dependency_overrides[get_duplication_repository] = _override_dup_repo
    test_app.dependency_overrides[get_storage_service] = _override_storage

    yield TestClient(test_app)

    test_app.dependency_overrides.clear()


# ---------------------------------------------------------------------------
# 3. GET /records — 列表查询
# ---------------------------------------------------------------------------


class TestListDuplicationRecords:
    """列表查询端点测试。"""

    def test_empty_list(self, client):
        """无记录时返回空列表。"""
        resp = client.get("/records")
        assert resp.status_code == 200
        assert resp.json() == []

    def test_returns_records(self, client, repo):
        """有记录时返回列表。"""
        r1 = _make_record(filename="a.mp4")
        r2 = _make_record(filename="b.mp4")
        repo.create(r1)
        repo.create(r2)

        resp = client.get("/records")
        assert resp.status_code == 200
        data = resp.json()
        assert len(data) == 2
        filenames = {item["filename"] for item in data}
        assert filenames == {"a.mp4", "b.mp4"}

    def test_record_response_fields(self, client, repo):
        """返回的字段应包含所有必需字段。"""
        record = _make_record()
        repo.create(record)

        resp = client.get("/records")
        assert resp.status_code == 200
        data = resp.json()
        assert len(data) == 1

        item = data[0]
        assert "id" in item
        assert "filename" in item
        assert "file_size" in item
        assert "status" in item
        assert "created_at" in item
        assert "updated_at" in item

    def test_only_returns_current_user_records(self, client, repo):
        """只返回当前用户的记录。"""
        r1 = _make_record(user_id="user-test-001", filename="mine.mp4")
        r2 = _make_record(user_id="other-user", filename="other.mp4")
        repo.create(r1)
        repo.create(r2)

        resp = client.get("/records")
        assert resp.status_code == 200
        data = resp.json()
        assert len(data) == 1
        assert data[0]["filename"] == "mine.mp4"


# ---------------------------------------------------------------------------
# 4. GET /records/{record_id} — 详情查询
# ---------------------------------------------------------------------------


class TestGetDuplicationDetail:
    """详情查询端点测试。"""

    def test_returns_detail_with_segments(self, client, repo):
        """返回记录详情含片段列表。"""
        seg = DuplicateSegment(
            id=uuid4().hex,
            source_start=0.0,
            source_end=5.0,
            matched_video_id="vid-1",
            matched_video_name="existing.mp4",
            matched_start=0.0,
            matched_end=5.0,
            similarity=92.5,
        )
        record = _make_record(status="completed")
        record.segments = [seg]
        repo.create(record)

        resp = client.get(f"/records/{record.id}")
        assert resp.status_code == 200
        data = resp.json()
        assert data["id"] == record.id
        assert data["status"] == "completed"
        assert len(data["segments"]) == 1
        assert data["segments"][0]["similarity"] == 92.5

    def test_returns_404_for_nonexistent(self, client):
        """不存在的记录返回 404。"""
        resp = client.get("/records/nonexistent-id")
        assert resp.status_code == 404

    def test_returns_404_for_other_user_record(self, client, repo):
        """其他用户的记录返回 404（安全隔离）。"""
        record = _make_record(user_id="other-user")
        repo.create(record)

        resp = client.get(f"/records/{record.id}")
        assert resp.status_code == 404

    def test_detail_includes_all_segment_fields(self, client, repo):
        """片段响应包含所有必需字段。"""
        seg = DuplicateSegment(
            id="seg-1",
            source_start=1.0,
            source_end=10.0,
            matched_video_id="vid-1",
            matched_video_name="ref.mp4",
            matched_start=2.0,
            matched_end=11.0,
            similarity=85.0,
        )
        record = _make_record(status="completed")
        record.segments = [seg]
        repo.create(record)

        resp = client.get(f"/records/{record.id}")
        assert resp.status_code == 200
        seg_data = resp.json()["segments"][0]
        assert seg_data["id"] == "seg-1"
        assert seg_data["source_start"] == 1.0
        assert seg_data["source_end"] == 10.0
        assert seg_data["matched_video_id"] == "vid-1"
        assert seg_data["matched_video_name"] == "ref.mp4"
        assert seg_data["matched_start"] == 2.0
        assert seg_data["matched_end"] == 11.0
        assert seg_data["similarity"] == 85.0


# ---------------------------------------------------------------------------
# 5. DELETE /records/{record_id} — 删除记录
# ---------------------------------------------------------------------------


class TestDeleteDuplicationRecord:
    """删除端点测试。"""

    def test_delete_existing_record(self, client, repo):
        """删除存在的记录返回 204。"""
        record = _make_record()
        repo.create(record)

        resp = client.delete(f"/records/{record.id}")
        assert resp.status_code == 204
        assert repo.get(record.id) is None

    def test_delete_nonexistent_returns_404(self, client):
        """删除不存在的记录返回 404。"""
        resp = client.delete("/records/nonexistent-id")
        assert resp.status_code == 404

    def test_delete_other_user_record_returns_404(self, client, repo):
        """删除其他用户的记录返回 404（安全隔离）。"""
        record = _make_record(user_id="other-user")
        repo.create(record)

        resp = client.delete(f"/records/{record.id}")
        assert resp.status_code == 404
        assert repo.get(record.id) is not None

    def test_delete_idempotent(self, client, repo):
        """删除后再次删除返回 404。"""
        record = _make_record()
        repo.create(record)

        resp1 = client.delete(f"/records/{record.id}")
        assert resp1.status_code == 204

        resp2 = client.delete(f"/records/{record.id}")
        assert resp2.status_code == 404


# ---------------------------------------------------------------------------
# 6. POST /records/{record_id}/retry — 重试查重
# ---------------------------------------------------------------------------


class TestRetryDuplication:
    """重试端点测试。"""

    def test_retry_failed_record(self, client, repo):
        """重试失败记录应重置状态为 pending。"""
        record = _make_record(status="failed")
        repo.create(record)

        resp = client.post(f"/records/{record.id}/retry")
        assert resp.status_code == 200
        data = resp.json()
        assert data["id"] == record.id
        assert data["status"] == "pending"
        assert "重新提交" in data["message"]

        # 验证 repo 中的记录也被更新
        updated = repo.get(record.id)
        assert updated.status == "pending"
        assert updated.error_message == ""

    def test_retry_nonexistent_returns_404(self, client):
        """重试不存在的记录返回 404。"""
        resp = client.post("/records/nonexistent-id/retry")
        assert resp.status_code == 404

    def test_retry_other_user_record_returns_404(self, client, repo):
        """重试其他用户的记录返回 404。"""
        record = _make_record(user_id="other-user", status="failed")
        repo.create(record)

        resp = client.post(f"/records/{record.id}/retry")
        assert resp.status_code == 404

    def test_retry_completed_record_returns_400(self, client, repo):
        """重试已完成记录 — 真实 UseCase 校验状态，非 failed 返回 400。"""
        record = _make_record(status="completed")
        repo.create(record)

        resp = client.post(f"/records/{record.id}/retry")
        assert resp.status_code == 400
        assert "failed" in resp.json()["detail"]

    def test_retry_pending_record_returns_400(self, client, repo):
        """重试 pending 状态的记录 — 真实 UseCase 返回 400。"""
        record = _make_record(status="pending")
        repo.create(record)

        resp = client.post(f"/records/{record.id}/retry")
        assert resp.status_code == 400
        assert "failed" in resp.json()["detail"]


# ---------------------------------------------------------------------------
# 7. 跨端点场景
# ---------------------------------------------------------------------------


class TestCrossEndpointScenarios:
    """跨端点集成场景。"""

    def test_create_then_list_then_detail(self, client, repo):
        """创建 → 列表 → 详情 完整流程。"""
        record = _make_record(filename="flow.mp4")
        repo.create(record)

        # 列表
        list_resp = client.get("/records")
        assert list_resp.status_code == 200
        assert len(list_resp.json()) == 1

        # 详情
        detail_resp = client.get(f"/records/{record.id}")
        assert detail_resp.status_code == 200
        assert detail_resp.json()["filename"] == "flow.mp4"

    def test_create_then_delete_then_404(self, client, repo):
        """创建 → 删除 → 详情 404 流程。"""
        record = _make_record()
        repo.create(record)

        # 删除
        del_resp = client.delete(f"/records/{record.id}")
        assert del_resp.status_code == 204

        # 详情应 404
        detail_resp = client.get(f"/records/{record.id}")
        assert detail_resp.status_code == 404

    def test_failed_record_retry_then_detail(self, client, repo):
        """失败记录 → 重试 → 查看详情状态已重置。"""
        record = _make_record(status="failed")
        repo.create(record)

        # 重试
        retry_resp = client.post(f"/records/{record.id}/retry")
        assert retry_resp.status_code == 200

        # 详情确认状态
        detail_resp = client.get(f"/records/{record.id}")
        assert detail_resp.status_code == 200
        assert detail_resp.json()["status"] == "pending"