""" 一键生成链路日志最小集 单元测试 覆盖： - GenerationTask.append_log() 正确追加结构化日志 - GenerationTask.append_log() 超过 200 条时截断 - GenerationTask.get_logs() 正确解析 JSON - GenerationTask.get_logs() 异常 JSON 不抛异常 - GenerationTaskResponse logs 字段 validator 解析 JSON 字符串 - GenerationTaskResponse logs 字段 validator 处理非法输入 - Worker 日志格式 [task_id=xxx] [阶段] 消息 - _flush_logs 异常不抛出 """ from __future__ import annotations import json import sys from pathlib import Path from unittest.mock import MagicMock, patch sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "apps" / "api")) sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "apps" / "worker")) # ── 预注入 mock 模块，防止 worker_app.db 触发真实数据库连接 ── _mock_db_module = MagicMock() _mock_db_module.SessionLocal = MagicMock() sys.modules.setdefault("worker_app.db", _mock_db_module) if "worker_app" in sys.modules: sys.modules["worker_app"].db = _mock_db_module from app.schemas.generation_task import GenerationTaskResponse from packages.domain.generation_task import GenerationTask, GenerationTaskStatus def _make_task(**kwargs) -> GenerationTask: """创建测试用 GenerationTask。""" defaults = { "id": "task-001", "project_id": "proj-001", "asset_library_id": "lib-001", "strategy_id": "one_take", "voice_library_id": "", "template_id": "tpl-001", "asset_ids": ["asset-1", "asset-2"], "title_ids": [], "voice_ids": [], "status": GenerationTaskStatus.PENDING, "progress": 0.0, "result_count": 0, "error_message": "", "created_by_user_id": "user-001", "source_edit_plan_id": "", "asset_select_mode": "all", "batch_id": "", } defaults.update(kwargs) return GenerationTask(**defaults) class TestAppendLog: """GenerationTask.append_log() 单元测试。""" def test_append_single_log(self): task = _make_task() task.append_log("接收任务", "任务开始", mode="one_take") logs = task.get_logs() assert len(logs) == 1 entry = logs[0] assert entry["level"] == "INFO" assert entry["stage"] == "接收任务" assert entry["message"] == "任务开始" assert entry["mode"] == "one_take" assert "ts" in entry def test_append_multiple_logs(self): task = _make_task() task.append_log("接收任务", "任务开始") task.append_log("下载素材", "下载完成", count=3) task.append_log("渲染", "渲染完成", duration=12.5) logs = task.get_logs() assert len(logs) == 3 assert logs[0]["stage"] == "接收任务" assert logs[1]["stage"] == "下载素材" assert logs[1]["count"] == 3 assert logs[2]["stage"] == "渲染" assert logs[2]["duration"] == 12.5 def test_append_log_with_error_level(self): task = _make_task() task.append_log("任务失败", "OSS上传失败", level="ERROR", error_type="RuntimeError") logs = task.get_logs() assert len(logs) == 1 assert logs[0]["level"] == "ERROR" assert logs[0]["error_type"] == "RuntimeError" def test_append_log_truncates_at_200(self): task = _make_task() for i in range(250): task.append_log("阶段", f"消息{i}") logs = task.get_logs() assert len(logs) == 200 # 保留最后 200 条 assert logs[0]["message"] == "消息50" assert logs[-1]["message"] == "消息249" def test_append_log_handles_corrupted_json(self): task = _make_task(logs="not-valid-json") task.append_log("接收任务", "任务开始") logs = task.get_logs() assert len(logs) == 1 assert logs[0]["message"] == "任务开始" def test_append_log_handles_empty_string(self): task = _make_task(logs="") task.append_log("接收任务", "任务开始") logs = task.get_logs() assert len(logs) == 1 class TestGetLogs: """GenerationTask.get_logs() 单元测试。""" def test_get_logs_empty(self): task = _make_task() assert task.get_logs() == [] def test_get_logs_parses_json(self): entries = [{"ts": "2026-01-01T00:00:00Z", "level": "INFO", "stage": "test", "message": "hello"}] task = _make_task(logs=json.dumps(entries, ensure_ascii=False)) logs = task.get_logs() assert len(logs) == 1 assert logs[0]["message"] == "hello" def test_get_logs_handles_invalid_json(self): task = _make_task(logs="{broken") assert task.get_logs() == [] def test_get_logs_handles_none(self): task = _make_task(logs=None) assert task.get_logs() == [] class TestGenerationTaskResponseLogs: """GenerationTaskResponse logs 字段 validator 测试。""" def _make_response_data(self, logs_value) -> dict: return { "id": "task-001", "project_id": "proj-001", "asset_library_id": "lib-001", "strategy_id": "one_take", "voice_library_id": "", "template_id": "", "asset_ids": [], "title_ids": [], "voice_ids": [], "source_edit_plan_id": "", "asset_select_mode": "all", "batch_id": "", "status": "completed", "progress": 1.0, "result_count": 1, "error_message": "", "logs": logs_value, } def test_logs_json_string_parsed(self): entries = [{"ts": "2026-01-01T00:00:00Z", "level": "INFO", "stage": "test", "message": "ok"}] data = self._make_response_data(json.dumps(entries, ensure_ascii=False)) resp = GenerationTaskResponse(**data) assert isinstance(resp.logs, list) assert len(resp.logs) == 1 assert resp.logs[0]["message"] == "ok" def test_logs_list_passthrough(self): entries = [{"ts": "2026-01-01T00:00:00Z", "level": "INFO", "stage": "test", "message": "ok"}] data = self._make_response_data(entries) resp = GenerationTaskResponse(**data) assert resp.logs == entries def test_logs_invalid_json_returns_empty(self): data = self._make_response_data("{broken") resp = GenerationTaskResponse(**data) assert resp.logs == [] def test_logs_empty_string_returns_empty(self): data = self._make_response_data("") resp = GenerationTaskResponse(**data) assert resp.logs == [] def test_logs_default_empty(self): data = self._make_response_data("[]") resp = GenerationTaskResponse(**data) assert resp.logs == [] class TestWorkerLogFormat: """Worker 日志格式 [task_id=xxx] [阶段] 消息 测试。""" def test_log_format_pattern(self): """验证日志格式匹配 [task_id=xxx] [阶段] 消息。""" import re task_id = "abc123" stage = "下载素材" message = "完成: 成功=3个, 耗时=1.5s" formatted = f"[task_id={task_id}] [{stage}] {message}" pattern = r"^\[task_id=[\w-]+\] \[.+\] .+$" assert re.match(pattern, formatted) def test_log_entries_contain_required_fields(self): """验证 append_log 生成的条目包含所有必需字段。""" task = _make_task() task.append_log("OSS上传", "上传成功", file_size=1024000, duration=2.5) logs = task.get_logs() entry = logs[0] assert "ts" in entry assert "level" in entry assert "stage" in entry assert "message" in entry assert entry["file_size"] == 1024000 assert entry["duration"] == 2.5 class TestFlushLogs: """_flush_logs 异常安全测试。""" def test_flush_logs_exception_not_raised(self): """_flush_logs 在 DB 异常时不应抛出。""" # 模拟 worker 环境 sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "apps" / "worker")) from worker_app.tasks.generation import _flush_logs task = _make_task() task.append_log("测试", "消息") # Mock SessionLocal 抛异常 with patch("worker_app.tasks.generation.SessionLocal", side_effect=RuntimeError("DB error")): # 不应抛出 _flush_logs("task-001", task)