xiaoxia-saas/tests/unit/test_duplication_api_enqueue.py

"""#1679 查重 API enqueue + repository.update 单测。

覆盖：
1. POST /duplication/upload 上传成功后调用 send_task 入队 worker。
2. POST /duplication/records/{id}/retry 重置后调用 send_task 入队 worker。
3. SQLAlchemyDuplicationRecordRepository.update 会调用 session.commit。
"""

from __future__ import annotations

import io
import sys
from pathlib import Path
from unittest.mock import MagicMock, patch

import pytest

ROOT = Path(__file__).resolve().parents[2]
sys.path.insert(0, str(ROOT / "apps" / "api"))
sys.path.insert(0, str(ROOT / "packages"))


# ── 通用 fixtures ────────────────────────────────────────────────


def _make_auth_user(user_id: str = "user-1"):
    from app.auth import AuthenticatedUser

    from packages.domain.entities import User

    user = User(id=user_id, email="u@example.com", display_name="Test User", username="u")
    return AuthenticatedUser(user=user)


def _make_record(record_id: str = "rec-1", status: str = "pending", user_id: str = "user-1"):
    from packages.domain.duplication import DuplicationRecord

    record = DuplicationRecord.create(
        user_id=user_id,
        filename="demo.mp4",
        file_size=2048,
        storage_key="duplication/abc/demo.mp4",
    )
    # 覆盖生成的 id，方便断言
    record.id = record_id
    record.status = status
    return record


@pytest.fixture()
def mock_repo():
    return MagicMock()


@pytest.fixture()
def mock_storage():
    return MagicMock()


@pytest.fixture()
def app(mock_repo, mock_storage):
    """构造带依赖 override 的 FastAPI app，避免拉起完整 DB / 认证栈。"""
    from unittest.mock import MagicMock as _MagicMock
    from unittest.mock import patch as _patch

    from app.api.routes.duplication import router as duplication_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 fastapi import FastAPI

    test_app = FastAPI()
    test_app.include_router(duplication_router, prefix="/duplication")
    test_app.dependency_overrides[get_current_user] = lambda: _make_auth_user()
    test_app.dependency_overrides[get_duplication_repository] = lambda: mock_repo
    test_app.dependency_overrides[get_storage_service] = lambda: mock_storage

    # Mock get_settings so OSS_DIRECT_UPLOAD_MAX_MB is available
    mock_settings = _MagicMock()
    mock_settings.OSS_DIRECT_UPLOAD_MAX_MB = 100
    with _patch("app.config.get_settings", return_value=mock_settings):
        yield test_app


# ── 1. 上传接口 enqueue ──────────────────────────────────────────


def test_upload_enqueue_calls_celery_task(app, mock_repo):
    """POST /duplication/upload 成功创建记录后，必须调用 send_task 入队 worker。"""
    from fastapi.testclient import TestClient

    record = _make_record("rec-upload", status="pending")
    mock_repo.create.return_value = record

    with patch("app.api.routes.duplication.celery_app") as mock_celery:
        mock_celery.send_task = MagicMock(return_value=MagicMock(id="task-xyz"))

        client = TestClient(app)
        # 使用 in-memory bytes，避免真实写盘
        response = client.post(
            "/duplication/upload",
            files={"file": ("demo.mp4", io.BytesIO(b"\x00\x00\x00\x00fake"), "video/mp4")},
        )

        assert response.status_code == 200, response.text
        body = response.json()
        assert body["id"] == "rec-upload"
        # 关键断言：send_task 被调用，且参数包含 worker.process_duplication_check
        mock_celery.send_task.assert_called_once_with("worker.process_duplication_check", args=["rec-upload"])


# ── 2. 重试接口 enqueue ──────────────────────────────────────────


def test_retry_enqueue_calls_celery_task(app, mock_repo):
    """POST /duplication/records/{id}/retry 重置后必须调用 send_task 入队 worker。"""
    from fastapi.testclient import TestClient

    record = _make_record("rec-retry", status="failed")
    mock_repo.get.return_value = record
    mock_repo.update.return_value = record

    with patch("app.api.routes.duplication.celery_app") as mock_celery:
        mock_celery.send_task = MagicMock(return_value=MagicMock(id="task-xyz"))

        client = TestClient(app)
        response = client.post("/duplication/records/rec-retry/retry")

        assert response.status_code == 200, response.text
        body = response.json()
        assert body["id"] == "rec-retry"
        # 关键断言：send_task 被调用
        mock_celery.send_task.assert_called_once_with("worker.process_duplication_check", args=["rec-retry"])


# ── 3. repository.update 调用 session.commit ─────────────────────


def test_repository_update_calls_session_commit():
    """SQLAlchemyDuplicationRecordRepository.update 必须在末尾调用 session.commit。"""
    from packages.adapters.sqlalchemy_impl.duplication_repository import (
        SQLAlchemyDuplicationRecordRepository,
    )
    from packages.domain.duplication import DuplicationRecord

    session = MagicMock()
    # 模拟 query().filter().first() 返回一个 model 实例
    model = MagicMock()
    model.id = "rec-1"
    query_proxy = MagicMock()
    query_proxy.filter.return_value.first.return_value = model
    session.query.return_value = query_proxy

    repo = SQLAlchemyDuplicationRecordRepository(session)

    record = DuplicationRecord.create(
        user_id="user-1",
        filename="demo.mp4",
        file_size=1024,
        storage_key="duplication/abc/demo.mp4",
    )
    record.id = "rec-1"

    repo.update(record)

    # 关键断言：session.commit 被调用一次
    session.commit.assert_called_once()