From 64f2f8ab3d1369e07f62e0536a23e79b611e4cc0 Mon Sep 17 00:00:00 2001 From: xiaoxia Date: Fri, 4 Sep 2026 11:31:31 +0800 Subject: [PATCH 1/2] =?UTF-8?q?fix:=20=E8=A1=A5=E6=9F=A5=E9=87=8D=20API=20?= =?UTF-8?q?enqueue=20=E5=8D=95=E6=B5=8B=20+=20=E4=BF=AE=E5=A4=8D=20duplica?= =?UTF-8?q?tion=5Fcheck=20=E6=97=A0=E9=99=90=E9=87=8D=E8=AF=95=20bug?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 新增 test_duplication_api_enqueue.py（3 个测试）： - 上传接口成功 enqueue celery task - 重试接口成功 enqueue celery task - repository.update 调用 session.commit - 修复 duplication_check.py 中 retry 无条件执行导致的无限重试： 仅在 retries < max_retries 时 retry，超限后直接 raise - 同步更新 test_duplication_check_worker.py 的 _run 辅助函数 以兼容新增的非 retry 异常路径 --- .../worker_app/tasks/duplication_check.py | 6 +- tests/unit/test_duplication_api_enqueue.py | 166 ++++++++++++++++++ tests/unit/test_duplication_check_worker.py | 4 + 3 files changed, 175 insertions(+), 1 deletion(-) create mode 100644 tests/unit/test_duplication_api_enqueue.py 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..ef1fe0704 --- /dev/null +++ b/tests/unit/test_duplication_api_enqueue.py @@ -0,0 +1,166 @@ +"""#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, patch as _patch + + from fastapi import FastAPI + + 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 + + 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 -- 2.54.0 From 481f8776b8ab674d33152d74421d95a27d61d731 Mon Sep 17 00:00:00 2001 From: CI Bot Date: Fri, 4 Sep 2026 03:34:57 +0000 Subject: [PATCH 2/2] style: auto-format with black + isort + prettier [skip ci-format-check] --- tests/unit/test_duplication_api_enqueue.py | 7 ++++--- 1 file changed, 4 insertions(+), 3 deletions(-) diff --git a/tests/unit/test_duplication_api_enqueue.py b/tests/unit/test_duplication_api_enqueue.py index ef1fe0704..b4a25c21f 100644 --- a/tests/unit/test_duplication_api_enqueue.py +++ b/tests/unit/test_duplication_api_enqueue.py @@ -25,6 +25,7 @@ sys.path.insert(0, str(ROOT / "packages")) 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") @@ -59,14 +60,14 @@ def mock_storage(): @pytest.fixture() def app(mock_repo, mock_storage): """构造带依赖 override 的 FastAPI app，避免拉起完整 DB / 认证栈。""" - from unittest.mock import MagicMock as _MagicMock, patch as _patch - - from fastapi import FastAPI + 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") -- 2.54.0