MemeMeow/tests/test_task_http.py

"""公共核心任务控制 HTTP 边界测试。"""

from __future__ import annotations

import ast
import asyncio
from pathlib import Path
from types import SimpleNamespace

import pytest
from fastapi import HTTPException

import api
import backend.task_http as task_http
from backend.tasks import TaskRecord


def _request(*, tasks: object | None = None, metadata: object | None = None, settings: object | None = None) -> SimpleNamespace:
    """构造不启动 lifespan 的最小任务请求。"""
    state = SimpleNamespace(tasks=tasks, metadata=metadata, settings=settings, agent_activity=None)
    return SimpleNamespace(app=SimpleNamespace(state=state), state=SimpleNamespace())


def _error(status: int, code: str, message: str) -> HTTPException:
    """构造任务入口使用的稳定错误 detail。"""
    return HTTPException(status_code=status, detail={"error": code, "message": message})


def _repository(_request: object) -> SimpleNamespace:
    """构造没有图片处理父 Job 的 repository。"""
    return SimpleNamespace(snapshot=lambda _task_id: None)


def test_task_http_module_keeps_one_way_dependency_and_legacy_aliases() -> None:
    """新模块不反向导入入口，旧任务符号仍保留。"""
    source = Path(task_http.__file__).read_text(encoding="utf-8")
    tree = ast.parse(source)
    imported_modules = {
        alias.name.split(".", 1)[0]
        for node in ast.walk(tree)
        if isinstance(node, ast.Import)
        for alias in node.names
    }
    imported_modules.update(
        node.module.split(".", 1)[0]
        for node in ast.walk(tree)
        if isinstance(node, ast.ImportFrom) and node.module
    )
    assert "api" not in imported_modules
    assert "server_api" not in imported_modules
    assert api.list_tasks.__name__ == "list_tasks"
    assert api.get_task.__name__ == "get_task"
    assert api.cancel_task.__name__ == "cancel_task"
    assert api.retry_task.__name__ == "retry_task"
    assert api._task_summary.__name__ == "_task_summary"


def test_task_routes_keep_metadata_and_order() -> None:
    """四个任务 route 保持 tasks tag、method、status 和相对顺序。"""
    paths = {"/search", "/generate-cache", "/tasks", "/tasks/{task_id}", "/tasks/{task_id}/cancel", "/tasks/{task_id}/retry"}
    routes = [route for route in api.app.routes if getattr(route, "path", None) in paths]
    assert [route.path for route in routes] == [
        "/search",
        "/generate-cache",
        "/tasks",
        "/tasks/{task_id}",
        "/tasks/{task_id}/cancel",
        "/tasks/{task_id}/retry",
    ]
    assert [route.methods for route in routes] == [{"POST"}, {"POST"}, {"GET"}, {"GET"}, {"POST"}, {"POST"}]
    assert [route.status_code for route in routes] == [None, 202, None, None, None, 202]
    assert all(route.tags == (["search"] if route.path == "/search" else ["tasks"]) for route in routes)


class _Tasks:
    """记录任务 service 的读取、取消和重试调用。"""

    def __init__(self, records: list[TaskRecord]) -> None:
        self.records = {record.task_id: record for record in records}
        self.calls: list[tuple[str, object]] = []

    def list(self, **kwargs: object) -> tuple[list[TaskRecord], str | None]:
        """返回固定任务页并记录分页参数。"""
        self.calls.append(("list", kwargs))
        return list(self.records.values()), "next-page"

    def get(self, task_id: str) -> TaskRecord | None:
        """按任务 ID 返回当前 scope 任务。"""
        self.calls.append(("get", task_id))
        return self.records.get(task_id)

    def cancel(self, task_id: str) -> bool:
        """记录取消并把任务收束为失败状态。"""
        self.calls.append(("cancel", task_id))
        record = self.records.get(task_id)
        if record is None:
            return False
        record.status = "failed"
        return True

    def retry(self, task_id: str) -> TaskRecord:
        """返回重试后的任务，或抛出稳定错误。"""
        self.calls.append(("retry", task_id))
        record = self.records.get(task_id)
        if record is None:
            raise RuntimeError("task_not_found")
        if record.status != "failed":
            raise RuntimeError("task_not_failed: current status")
        record.status = "queued"
        return record


def test_list_reads_current_service_once_and_projects_metadata() -> None:
    """任务列表只读取当前 scope service，并不返回 payload。"""
    record = TaskRecord(task_id="task-1", task_type="cache_generation", payload={"secret": "hidden"})
    tasks = _Tasks([record])
    metadata = SimpleNamespace(image_for_meme=lambda _meme_id: (None, Path("sample.png")))
    request = _request(tasks=tasks, metadata=metadata)
    service_calls: list[str] = []

    def service(_request: object, name: str) -> object:
        """记录 service 名称并只暴露 fake scope service。"""
        service_calls.append(name)
        return tasks if name == "tasks" else metadata

    response = asyncio.run(
        task_http.list_tasks(
            request,
            status=["queued"],
            task_type=["cache_generation"],
            cursor="cursor-1",
            limit=10,
            service=service,
            processing_repository=_repository,
        )
    )
    assert response["next_cursor"] == "next-page"
    assert response["items"][0]["task_id"] == "task-1"
    assert "payload" not in response["items"][0]
    assert service_calls == ["tasks"]
    assert tasks.calls[0] == (
        "list",
        {"statuses": {"queued"}, "task_types": {"cache_generation"}, "cursor": "cursor-1", "limit": 10},
    )


def test_missing_task_uses_processing_fallback_before_stable_404() -> None:
    """普通任务缺失时查询图片处理回退，均缺失才返回 task_not_found。"""
    tasks = _Tasks([])
    request = _request(tasks=tasks)
    snapshots = {"job-1": SimpleNamespace(as_dict=lambda: {"task_id": "job-1", "status": "running"})}
    repository_calls: list[str] = []

    def repository(_request: object) -> SimpleNamespace:
        """记录父 Job 查询并返回受控快照。"""
        return SimpleNamespace(snapshot=lambda task_id: repository_calls.append(task_id) or snapshots.get(task_id))

    def service(_request: object, name: str) -> object:
        """只返回当前 scope task service。"""
        assert name == "tasks"
        return tasks

    fallback = asyncio.run(task_http.get_task(request, "job-1", service=service, error=_error, processing_repository=repository))
    assert fallback["task_type"] == "visual_embedding_generation"
    assert fallback["image_stage"] == "visual"
    assert repository_calls == ["job-1"]

    with pytest.raises(HTTPException) as caught:
        asyncio.run(task_http.get_task(request, "missing", service=service, error=_error, processing_repository=repository))
    assert caught.value.status_code == 404
    assert caught.value.detail["error"] == "task_not_found"


def test_cancel_calls_agent_adapter_only_for_running_context_task() -> None:
    """取消语境任务时只向适配器传递当前 task id。"""
    record = TaskRecord(task_id="agent-1", task_type="meme_context_generation", status="running")
    tasks = _Tasks([record])
    request = _request(tasks=tasks)
    cancelled: list[str] = []

    def service(_request: object, name: str) -> object:
        """返回当前 scope task service。"""
        assert name == "tasks"
        return tasks

    response = asyncio.run(
        task_http.cancel_task(
            request,
            "agent-1",
            service=service,
            error=_error,
            processing_repository=_repository,
            cancel_agent=lambda _request, task_id: cancelled.append(task_id),
        )
    )
    assert response["status"] == "failed"
    assert cancelled == ["agent-1"]
    assert [call[0] for call in tasks.calls] == ["get", "cancel", "get"]


@pytest.mark.parametrize(
    ("task_id", "status", "expected_status", "expected_code"),
    [("missing", "failed", 404, "task_not_found"), ("queued", "queued", 409, "task_not_failed")],
)
def test_retry_maps_stable_task_errors(task_id: str, status: str, expected_status: int, expected_code: str) -> None:
    """重试错误只返回稳定 HTTP code，不泄露底层诊断文本。"""
    records = [] if task_id == "missing" else [TaskRecord(task_id=task_id, task_type="cache_generation", status=status)]
    tasks = _Tasks(records)
    request = _request(tasks=tasks)

    def service(_request: object, name: str) -> object:
        """返回当前 scope task service。"""
        assert name == "tasks"
        return tasks

    with pytest.raises(HTTPException) as caught:
        asyncio.run(task_http.retry_task(request, task_id, service=service, error=_error, processing_repository=_repository))
    assert caught.value.status_code == expected_status
    assert caught.value.detail == {"error": expected_code, "message": "任务不存在" if expected_code == "task_not_found" else "只有失败任务可以重试"}