diff --git a/apps/worker/worker_app/tasks/duplication_check.py b/apps/worker/worker_app/tasks/duplication_check.py index dcfa49c17..81472e323 100644 --- a/apps/worker/worker_app/tasks/duplication_check.py +++ b/apps/worker/worker_app/tasks/duplication_check.py @@ -187,7 +187,11 @@ def process_duplication_check(self: Task, record_id: str) -> dict: except Exception as inner: logger.error("Failed to mark duplication record %s as failed: %s", record_id, inner) session.rollback() - raise self.retry(exc=e, countdown=60) from e + # 仅在还有重试次数时才重新入队;超过 max_retries 时上面已标记 failed, + # 直接抛出异常让 Celery 记录错误,不再无限重试。 + if self.request.retries < self.max_retries: + raise self.retry(exc=e, countdown=60) from e + raise finally: if session is not None: diff --git a/tests/unit/test_duplication_api_enqueue.py b/tests/unit/test_duplication_api_enqueue.py new file mode 100644 index 000000000..b4a25c21f --- /dev/null +++ b/tests/unit/test_duplication_api_enqueue.py @@ -0,0 +1,167 @@ +"""#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() diff --git a/tests/unit/test_duplication_check_worker.py b/tests/unit/test_duplication_check_worker.py index b6dd9f901..de8cb9417 100644 --- a/tests/unit/test_duplication_check_worker.py +++ b/tests/unit/test_duplication_check_worker.py @@ -42,6 +42,8 @@ def _run(mod, record_id, retries=0): result = func(record_id) except CeleryRetry as e: raised = e + except Exception as e: + raised = e return result, raised, None mock_self = MagicMock() mock_self.request.retries = retries @@ -51,6 +53,8 @@ def _run(mod, record_id, retries=0): result = func(mock_self, record_id) except CeleryRetry as e: raised = e + except Exception as e: + raised = e return result, raised, mock_self