xiaoxia-saas/tests/unit/test_duplication_api_enqueue.py

"""#1661 查重 API enqueue 及仓储 commit 覆盖测试。

覆盖：
- upload 接口在成功后调用 celery_app.send_task
- retry 接口在成功后调用 celery_app.send_task
- duplication_repository.update() 正确调用 session.commit()
"""

from __future__ import annotations

import os
import sys
from unittest.mock import MagicMock, patch

import pytest

os.environ.setdefault("JWT_SECRET_KEY", "unit-test-secret-key-for-testing")
os.environ.setdefault("DATABASE_URL", "sqlite:///test.db")

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

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 fastapi import FastAPI
from fastapi.testclient import TestClient

from packages.domain.duplication import DuplicationRecord
from packages.domain.entities import User

# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------


def _make_test_user():
    return User(id="user-1", username="testuser", email="test@example.com", display_name="Test User")


def _make_auth_user():
    return AuthenticatedUser(user=_make_test_user(), session_id="test-session", token_type="bearer")


def _make_record(status="pending"):
    record = DuplicationRecord.create(
        user_id="user-1",
        filename="test.mp4",
        file_size=1024,
        storage_key="duplication/abc/test.mp4",
    )
    if status != "pending":
        record.status = status
    return record


def _build_client(auth_user, repo, storage=None):
    """构建带 dependency_overrides 的 TestClient。"""
    app = FastAPI()
    app.include_router(router, prefix="/duplication")
    app.dependency_overrides[get_current_user] = lambda: auth_user
    app.dependency_overrides[get_duplication_repository] = lambda: repo
    if storage is not None:
        app.dependency_overrides[get_storage_service] = lambda: storage
    return TestClient(app)


# ---------------------------------------------------------------------------
# 1. Upload endpoint enqueues celery task
# ---------------------------------------------------------------------------


def test_upload_enqueue_calls_celery_task():
    """POST /duplication/upload 成功创建记录后必须调用 send_task。"""
    record = _make_record()

    fake_repo = MagicMock()
    fake_repo.create.return_value = record

    fake_storage = MagicMock()
    fake_auth = _make_auth_user()

    client = _build_client(fake_auth, fake_repo, fake_storage)

    with patch("app.api.routes.duplication.celery_app") as mock_celery:
        response = client.post(
            "/duplication/upload",
            files={"file": ("test.mp4", b"fake-video-content", "video/mp4")},
        )

    assert response.status_code == 200, response.text
    mock_celery.send_task.assert_called_once_with(
        "worker.process_duplication_check",
        args=[record.id],
    )


# ---------------------------------------------------------------------------
# 2. Retry endpoint enqueues celery task
# ---------------------------------------------------------------------------


def test_retry_enqueue_calls_celery_task():
    """POST /duplication/records/{id}/retry 成功后必须调用 send_task。"""
    record = _make_record(status="failed")

    fake_repo = MagicMock()
    fake_repo.get.return_value = record

    # RetryDuplicationUseCase.execute 内部调用 repo.get → record.reset_for_retry → repo.update
    updated = _make_record()
    updated.id = record.id
    updated.status = "pending"
    fake_repo.update.return_value = updated

    fake_auth = _make_auth_user()

    client = _build_client(fake_auth, fake_repo)

    with patch("app.api.routes.duplication.celery_app") as mock_celery:
        response = client.post(f"/duplication/records/{record.id}/retry")

    assert response.status_code == 200, response.text
    mock_celery.send_task.assert_called_once_with(
        "worker.process_duplication_check",
        args=[record.id],
    )


# ---------------------------------------------------------------------------
# 3. Repository update calls session.commit()
# ---------------------------------------------------------------------------


def test_repository_update_calls_session_commit():
    """duplication_repository 的 update 方法必须调用 session.commit()。"""
    from packages.adapters.sqlalchemy_impl.duplication_repository import (
        SQLAlchemyDuplicationRecordRepository,
    )
    from packages.adapters.sqlalchemy_impl.models import DuplicationRecordModel

    mock_session = MagicMock()
    mock_model = MagicMock(spec=DuplicationRecordModel)
    mock_model.id = "rec-1"

    mock_session.query.return_value.filter.return_value.first.return_value = mock_model

    repo = SQLAlchemyDuplicationRecordRepository(mock_session)

    record = DuplicationRecord.create(
        user_id="user-1",
        filename="test.mp4",
        file_size=1024,
        storage_key="duplication/abc/test.mp4",
    )
    record.status = "completed"
    record.duplicate_rate = 42.0
    record.duplicate_count = 1
    record.visual_similarity = 0.85
    record.match_count = 2

    result = repo.update(record)

    mock_session.commit.assert_called()
    assert result.visual_similarity == 0.85
    assert result.match_count == 2