灵应
23ef50ccc0
CI/CD Pipeline / Build & Push Staging (Watchtower auto-deploy) (push) Has been skipped
CI/CD Pipeline / Build Production Runtime Images (push) Has been skipped
CI/CD Pipeline / Staging E2E Tests (push) Has been skipped
CI/CD Pipeline / Staging API Integration Tests (push) Has been skipped
CI/CD Pipeline / Deploy Production (push) Has been skipped
CI/CD Pipeline / Production Browser E2E (push) Has been skipped
CI/CD Pipeline / Frontend Lint (push) Failing after 1h0m49s
CI/CD Pipeline / Validate Code Quality And Tests (push) Failing after 1h0m49s
302 lines
11 KiB
Python
302 lines
11 KiB
Python
"""
|
|
分类任务 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"])
|