xiaoxia-saas/tests/unit/test_generation_logging.py

"""
一键生成链路日志最小集 单元测试

覆盖：
- 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)