""" 分类任务 API 集成测试。 覆盖端点： - POST /classification-jobs — 提交分类任务 - GET /classification-jobs/{job_id} — 获取分类任务详情 使用 FastAPI TestClient + dependency_overrides 模式， 导入真实路由模块，mock Celery 和 repository。 """ from __future__ import annotations import os import sys from unittest.mock import MagicMock # ── 环境变量 & sys.path（必须在导入 app.* 之前设置） ────────────────────────── os.environ.setdefault("JWT_SECRET_KEY", "unit-test-secret-key-for-testing") os.environ.setdefault("DATABASE_URL", "sqlite:///test.db") import pytest from fastapi import FastAPI from fastapi.testclient import TestClient sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "..", "apps", "api")) # mock celery_app 以避免实际发送任务 import app.api.routes.classification_jobs as classification_routes from app.api.routes.classification_jobs import router from app.dependencies import get_classification_job_repository from packages.adapters.in_memory import InMemoryClassificationJobRepository from packages.domain import ClassificationJob, ClassificationJobStatus classification_routes.celery_app = MagicMock() # --------------------------------------------------------------------------- # 1. 辅助函数 # --------------------------------------------------------------------------- def _make_job( project_id: str = "proj-1", asset_id: str = "asset-1", status: ClassificationJobStatus = ClassificationJobStatus.PENDING, ) -> ClassificationJob: job = ClassificationJob.create(project_id=project_id, asset_id=asset_id) if status == ClassificationJobStatus.PROCESSING: job.status = ClassificationJobStatus.PROCESSING elif status == ClassificationJobStatus.COMPLETED: job.status = ClassificationJobStatus.COMPLETED job.classification = "scenic" job.confidence = 0.92 elif status == ClassificationJobStatus.FAILED: job.status = ClassificationJobStatus.FAILED job.error_message = "AI 服务不可用" return job # --------------------------------------------------------------------------- # 2. Fixtures # --------------------------------------------------------------------------- @pytest.fixture def repo(): return InMemoryClassificationJobRepository() @pytest.fixture def client(repo): """创建带有依赖覆盖的 TestClient。""" test_app = FastAPI() test_app.include_router(router, prefix="/classification-jobs") def _override_repo(): return repo test_app.dependency_overrides[get_classification_job_repository] = _override_repo yield TestClient(test_app) test_app.dependency_overrides.clear() # --------------------------------------------------------------------------- # 3. POST / — 提交分类任务 # --------------------------------------------------------------------------- class TestSubmitClassificationJob: """提交分类任务端点测试。""" def test_submit_with_valid_data(self, client): """使用有效数据提交分类任务应成功。""" resp = client.post( "/classification-jobs", json={ "project_id": "proj-123", "asset_id": "asset-456", }, ) assert resp.status_code == 200 data = resp.json() assert data["project_id"] == "proj-123" assert data["asset_id"] == "asset-456" assert data["status"] == "pending" assert data["classification"] == "" assert data["confidence"] == 0.0 assert data["error_message"] == "" assert "id" in data assert len(data["id"]) > 0 def test_submit_generates_unique_id(self, client): """每次提交应生成不同的任务 ID。""" resp1 = client.post("/classification-jobs", json={"project_id": "p1", "asset_id": "a1"}) resp2 = client.post("/classification-jobs", json={"project_id": "p1", "asset_id": "a2"}) assert resp1.json()["id"] != resp2.json()["id"] def test_submit_missing_project_id_returns_422(self, client): """缺少 project_id 应返回 422。""" resp = client.post("/classification-jobs", json={"asset_id": "asset-1"}) assert resp.status_code == 422 def test_submit_missing_asset_id_returns_422(self, client): """缺少 asset_id 应返回 422。""" resp = client.post("/classification-jobs", json={"project_id": "proj-1"}) assert resp.status_code == 422 def test_submit_empty_project_id_returns_422(self, client): """空 project_id 应返回 422。""" resp = client.post("/classification-jobs", json={"project_id": "", "asset_id": "asset-1"}) assert resp.status_code == 422 def test_submit_empty_asset_id_returns_422(self, client): """空 asset_id 应返回 422。""" resp = client.post("/classification-jobs", json={"project_id": "proj-1", "asset_id": ""}) assert resp.status_code == 422 def test_submit_sends_celery_task(self, client): """提交任务后应触发 Celery 异步任务。""" classification_routes.celery_app.send_task.reset_mock() resp = client.post("/classification-jobs", json={"project_id": "p1", "asset_id": "a1"}) assert resp.status_code == 200 job_id = resp.json()["id"] classification_routes.celery_app.send_task.assert_called_once_with( "worker.classify_asset", args=[job_id], ) def test_submit_persists_to_repository(self, client, repo): """提交后任务应保存到 repository。""" resp = client.post("/classification-jobs", json={"project_id": "p1", "asset_id": "a1"}) job_id = resp.json()["id"] saved = repo.get(job_id) assert saved is not None assert saved.project_id == "p1" assert saved.asset_id == "a1" assert saved.status == ClassificationJobStatus.PENDING # --------------------------------------------------------------------------- # 4. GET /{job_id} — 获取分类任务详情 # --------------------------------------------------------------------------- class TestGetClassificationJob: """获取分类任务详情端点测试。""" def test_get_pending_job(self, client, repo): """获取 pending 状态的任务。""" job = _make_job(status=ClassificationJobStatus.PENDING) repo.create(job) resp = client.get(f"/classification-jobs/{job.id}") assert resp.status_code == 200 data = resp.json() assert data["id"] == job.id assert data["status"] == "pending" assert data["classification"] == "" assert data["confidence"] == 0.0 def test_get_processing_job(self, client, repo): """获取 processing 状态的任务。""" job = _make_job(status=ClassificationJobStatus.PROCESSING) repo.create(job) resp = client.get(f"/classification-jobs/{job.id}") assert resp.status_code == 200 assert resp.json()["status"] == "processing" def test_get_completed_job(self, client, repo): """获取已完成的任务应包含分类结果和置信度。""" job = _make_job(status=ClassificationJobStatus.COMPLETED) repo.create(job) resp = client.get(f"/classification-jobs/{job.id}") assert resp.status_code == 200 data = resp.json() assert data["status"] == "completed" assert data["classification"] == "scenic" assert data["confidence"] == 0.92 assert data["error_message"] == "" def test_get_failed_job(self, client, repo): """获取失败的任务应包含错误信息。""" job = _make_job(status=ClassificationJobStatus.FAILED) repo.create(job) resp = client.get(f"/classification-jobs/{job.id}") assert resp.status_code == 200 data = resp.json() assert data["status"] == "failed" assert "AI 服务不可用" in data["error_message"] def test_get_nonexistent_job_returns_404(self, client): """获取不存在的任务应返回 404。""" resp = client.get("/classification-jobs/nonexistent-job-id") assert resp.status_code == 404 def test_response_contains_all_required_fields(self, client, repo): """响应应包含所有必需字段。""" job = _make_job() repo.create(job) resp = client.get(f"/classification-jobs/{job.id}") data = resp.json() for field in ["id", "project_id", "asset_id", "status", "classification", "confidence", "error_message"]: assert field in data, f"缺少字段: {field}" # --------------------------------------------------------------------------- # 5. 跨端点场景 # --------------------------------------------------------------------------- class TestClassificationApiScenarios: """分类任务 API 跨端点集成场景。""" def test_submit_then_get_pending(self, client, repo): """提交任务后立即查询应为 pending 状态。""" submit_resp = client.post( "/classification-jobs", json={"project_id": "proj-scenario", "asset_id": "asset-scenario"}, ) assert submit_resp.status_code == 200 job_id = submit_resp.json()["id"] get_resp = client.get(f"/classification-jobs/{job_id}") assert get_resp.status_code == 200 assert get_resp.json()["status"] == "pending" assert get_resp.json()["project_id"] == "proj-scenario" assert get_resp.json()["asset_id"] == "asset-scenario" def test_submit_simulate_complete_then_get(self, client, repo): """模拟 worker 完成任务后查询应返回结果。""" submit_resp = client.post("/classification-jobs", json={"project_id": "p1", "asset_id": "a1"}) job_id = submit_resp.json()["id"] # 模拟 worker 处理完成 job = repo.get(job_id) assert job is not None job.status = ClassificationJobStatus.COMPLETED job.classification = "product" job.confidence = 0.88 repo.update(job) # 查询结果 get_resp = client.get(f"/classification-jobs/{job_id}") assert get_resp.status_code == 200 data = get_resp.json() assert data["status"] == "completed" assert data["classification"] == "product" assert data["confidence"] == 0.88 def test_submit_simulate_failure_then_get(self, client, repo): """模拟 worker 失败后查询应返回错误信息。""" submit_resp = client.post("/classification-jobs", json={"project_id": "p1", "asset_id": "a1"}) job_id = submit_resp.json()["id"] # 模拟处理失败 job = repo.get(job_id) assert job is not None job.status = ClassificationJobStatus.FAILED job.error_message = "网络超时" repo.update(job) get_resp = client.get(f"/classification-jobs/{job_id}") assert get_resp.status_code == 200 data = get_resp.json() assert data["status"] == "failed" assert "网络超时" in data["error_message"] if __name__ == "__main__": pytest.main([__file__, "-v"])