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:

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()

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