AstrBot/tests/unit/test_cron_tools.py

"""Tests for cron tool metadata."""

from types import SimpleNamespace
from unittest.mock import AsyncMock

import pytest

from astrbot.core.tools.cron_tools import (
    CreateActiveCronTool,
    DeleteCronJobTool,
    ListCronJobsTool,
)


def test_create_future_task_tool_has_correct_name():
    """The create tool should have correct name."""
    tool = CreateActiveCronTool()

    assert tool.name == "create_future_task"


def test_create_future_task_tool_requires_note():
    """The create tool should require note field."""
    tool = CreateActiveCronTool()

    assert "note" in tool.parameters["required"]


def test_create_future_task_tool_has_cron_guidance():
    """The create tool should have cron guidance in description."""
    tool = CreateActiveCronTool()

    description = tool.parameters["properties"]["cron_expression"]["description"]

    assert "mon-fri" in description
    assert "sat,sun" in description


def test_delete_future_task_tool_has_job_id_required():
    """The delete tool should require job_id."""
    tool = DeleteCronJobTool()

    assert tool.name == "delete_future_task"
    assert "job_id" in tool.parameters["required"]


def test_list_future_tasks_tool_name():
    """The list tool should have correct name."""
    tool = ListCronJobsTool()

    assert tool.name == "list_future_tasks"


@pytest.mark.asyncio
async def test_delete_future_task_requires_job_id():
    """Delete should require job_id."""
    tool = DeleteCronJobTool()
    context = SimpleNamespace(
        context=SimpleNamespace(
            context=SimpleNamespace(cron_manager=SimpleNamespace()),
            event=SimpleNamespace(
                unified_msg_origin="test:private:session",
                get_sender_id=lambda: "user-1",
            ),
        )
    )

    result = await tool.call(context, job_id=None)

    assert "job_id is required" in result


@pytest.mark.asyncio
async def test_delete_future_task_verifies_ownership():
    """Delete should verify the job belongs to current umo."""
    tool = DeleteCronJobTool()
    cron_mgr = SimpleNamespace(
        db=SimpleNamespace(get_cron_job=AsyncMock(return_value=SimpleNamespace(
            job_id="job-1",
            name="test job",
            payload={"session": "other:private:session"},
        ))),
    )
    context = SimpleNamespace(
        context=SimpleNamespace(
            context=SimpleNamespace(cron_manager=cron_mgr),
            event=SimpleNamespace(
                unified_msg_origin="test:private:session",
                get_sender_id=lambda: "user-1",
            ),
        )
    )

    result = await tool.call(context, job_id="job-1")

    assert "only delete" in result


@pytest.mark.asyncio
async def test_delete_future_task_success():
    """Delete should successfully delete job and return message."""
    tool = DeleteCronJobTool()
    cron_mgr = SimpleNamespace(
        db=SimpleNamespace(get_cron_job=AsyncMock(return_value=SimpleNamespace(
            job_id="job-1",
            name="test job",
            payload={"session": "test:private:session"},
        ))),
        delete_job=AsyncMock(return_value=True),
    )
    context = SimpleNamespace(
        context=SimpleNamespace(
            context=SimpleNamespace(cron_manager=cron_mgr),
            event=SimpleNamespace(
                unified_msg_origin="test:private:session",
                get_sender_id=lambda: "user-1",
            ),
        )
    )

    result = await tool.call(context, job_id="job-1")

    cron_mgr.delete_job.assert_awaited_once_with("job-1")
    assert "Deleted cron job job-1" in result


@pytest.mark.asyncio
async def test_list_future_tasks_returns_jobs():
    """List should return formatted job list."""
    tool = ListCronJobsTool()
    cron_mgr = SimpleNamespace(
        list_jobs=AsyncMock(return_value=[
            SimpleNamespace(
                job_id="job-1",
                name="test job",
                job_type="active_agent",
                run_once=False,
                enabled=True,
                next_run_time="2026-04-16T08:00:00",
                payload={"session": "test:private:session"},
            ),
        ]),
    )
    context = SimpleNamespace(
        context=SimpleNamespace(
            context=SimpleNamespace(cron_manager=cron_mgr),
            event=SimpleNamespace(
                unified_msg_origin="test:private:session",
                get_sender_id=lambda: "user-1",
            ),
        )
    )

    result = await tool.call(context)

    assert "job-1" in result
    assert "test job" in result


@pytest.mark.asyncio
async def test_list_future_tasks_empty():
    """List should return message when no jobs."""
    tool = ListCronJobsTool()
    cron_mgr = SimpleNamespace(
        list_jobs=AsyncMock(return_value=[]),
    )
    context = SimpleNamespace(
        context=SimpleNamespace(
            context=SimpleNamespace(cron_manager=cron_mgr),
            event=SimpleNamespace(
                unified_msg_origin="test:private:session",
                get_sender_id=lambda: "user-1",
            ),
        )
    )

    result = await tool.call(context)

    assert "No cron jobs found" in result