MemeMeow/tests/test_api_activity.py

"""任务摘要批量装配 Agent 活跃度字段的 API 边界测试。"""

from __future__ import annotations

import asyncio
import sqlite3
from pathlib import Path
from types import SimpleNamespace

from api import get_task, list_tasks
from backend.opencode_activity import AgentActivity, OpenCodeActivityReader
from backend.tasks import TaskRecord


class FakeActivityReader:
    """记录批量调用并返回固定领域值的测试 reader。"""

    def __init__(self, values=None, error: Exception | None = None):
        self.values = values or {}
        self.error = error
        self.calls: list[list[str]] = []

    def read_many(self, task_ids):
        """模拟一次批量 SQLite 观测。"""
        self.calls.append(list(task_ids))
        if self.error:
            raise self.error
        return {task_id: self.values[task_id] for task_id in task_ids if task_id in self.values}


class FakeTasks:
    """为路由函数提供任务列表和详情快照。"""

    def __init__(self, records: list[TaskRecord]):
        self.records = records

    def list(self, **_kwargs):
        """返回固定任务页，不执行额外查询。"""
        return self.records, None

    def get(self, task_id: str):
        """按任务 ID 返回固定详情。"""
        return next((record for record in self.records if record.task_id == task_id), None)


def _request(records: list[TaskRecord], reader: FakeActivityReader):
    """构造只包含摘要装配依赖的伪 Request。"""
    metadata = SimpleNamespace(image_for_meme=lambda _meme_id: (None, Path("sample.png")))
    state = SimpleNamespace(tasks=FakeTasks(records), agent_activity=reader, metadata=metadata)
    return SimpleNamespace(app=SimpleNamespace(state=state))


def _context_record(task_id: str) -> TaskRecord:
    """创建带图片身份的语境任务快照。"""
    return TaskRecord(task_id=task_id, task_type="meme_context_generation", payload={"meme_id": f"meme-{task_id}"}, status="running")


def test_task_list_reads_activity_once_and_omits_non_context_fields():
    """任务列表对多个语境任务只批量读取一次，其他类型不带 Agent 字段。"""
    records = [_context_record("one"), _context_record("two"), TaskRecord(task_id="cache", task_type="cache_generation")]
    reader = FakeActivityReader({"one": AgentActivity(3, True, "2026-08-13T12:23:57Z"), "two": AgentActivity(4, False, "2026-08-13T12:24:57Z")})
    request = _request(records, reader)

    response = asyncio.run(list_tasks(request, status=[], task_type=[], cursor=None, limit=50))

    assert reader.calls == [["one", "two"]]
    assert response["items"][0]["agent_completed_turns"] == 3
    assert response["items"][0]["agent_turn_running"] is True
    assert response["items"][1]["agent_turn_running"] is False
    assert "agent_completed_turns" not in response["items"][2]


def test_task_detail_reuses_single_activity_boundary():
    """单任务详情复用同一 reader 边界并返回完整字段。"""
    record = _context_record("detail")
    reader = FakeActivityReader({"detail": AgentActivity(18, True, "2026-08-13T12:23:57Z")})
    request = _request([record], reader)

    response = asyncio.run(get_task(request, "detail"))

    assert reader.calls == [["detail"]]
    assert response["agent_completed_turns"] == 18
    assert response["agent_turn_running"] is True
    assert response["agent_last_activity_at"].endswith("Z")


def test_reader_error_keeps_task_api_successful_and_omits_all_activity():
    """reader 异常只会隐藏活动字段，不改变既有任务响应。"""
    record = _context_record("failed-reader")
    reader = FakeActivityReader(error=PermissionError("database unavailable"))
    request = _request([record], reader)

    response = asyncio.run(get_task(request, "failed-reader"))

    assert response["task_id"] == "failed-reader"
    assert response["status"] == "running"
    assert not {"agent_completed_turns", "agent_turn_running", "agent_last_activity_at"} & response.keys()


def test_incomplete_activity_is_not_exposed_as_partial_summary():
    """不完整活动值不会被误报为零轮或半个活动摘要。"""
    record = _context_record("partial")
    reader = FakeActivityReader({"partial": {"agent_completed_turns": 0, "agent_turn_running": False}})
    request = _request([record], reader)

    response = asyncio.run(get_task(request, "partial"))

    assert not {"agent_completed_turns", "agent_turn_running", "agent_last_activity_at"} & response.keys()


def test_fault_injected_reader_keeps_list_and_detail_successful(tmp_path: Path):
    """数据库缺失、schema 不兼容和临时锁忙时两个任务路由仍返回成功。"""
    runtimes: list[tuple[Path, sqlite3.Connection | None]] = [(tmp_path / "missing", None)]
    incompatible = tmp_path / "incompatible"
    incompatible.mkdir()
    connection = sqlite3.connect(incompatible / "opencode.db")
    connection.execute("CREATE TABLE session(id TEXT)")
    connection.commit()
    connection.close()
    runtimes.append((incompatible, None))

    locked = tmp_path / "locked"
    locked.mkdir()
    connection = sqlite3.connect(locked / "opencode.db")
    connection.executescript(
        """
        CREATE TABLE session(id TEXT PRIMARY KEY, title TEXT NOT NULL, time_created INTEGER NOT NULL);
        CREATE TABLE part(id TEXT PRIMARY KEY, session_id TEXT NOT NULL, time_updated INTEGER NOT NULL, data TEXT NOT NULL);
        """
    )
    connection.commit()
    connection.execute("BEGIN EXCLUSIVE")
    runtimes.append((locked, connection))

    try:
        for runtime, _writer in runtimes:
            record = _context_record(f"fault-{runtime.name}")
            reader = OpenCodeActivityReader(runtime, busy_timeout_ms=1)
            request = _request([record], reader)
            listed = asyncio.run(list_tasks(request, status=[], task_type=[], cursor=None, limit=50))
            detail = asyncio.run(get_task(request, record.task_id))
            assert listed["items"][0]["status"] == "running"
            assert detail["status"] == "running"
            assert not {"agent_completed_turns", "agent_turn_running", "agent_last_activity_at"} & detail.keys()
    finally:
        connection.rollback()
        connection.close()