meshcore-bot/tests/characterization/test_execution_accounting.py

"""Characterization: how command executions and responses are accounted for.

Pins, with an advancing clock, when ``execute_commands`` records a cooldown,
what a second ``record_execution`` inside ``execute`` does to the boundary,
which stats rows are written for each outcome, and that local-plugin commands
with zero-argument cooldown methods keep working. Some of this is defective
today (``_last_response`` is never reset, so a silent command after any reply
is recorded as having responded); it is recorded as-is so the refactor cannot
change it by accident. The accounting fix will update these expectations.
"""

from __future__ import annotations

import configparser
from pathlib import Path
from unittest.mock import AsyncMock, Mock, patch

import pytest

from modules.command_manager import CommandManager
from modules.commands.base_command import BaseCommand
from modules.models import MeshMessage


class _Clock:
    def __init__(self, start: float = 1_000_000.0) -> None:
        self.now = start

    def __call__(self) -> float:
        return self.now

    def advance(self, seconds: float) -> None:
        self.now += seconds


class _Probe(BaseCommand):
    name = "probe"
    keywords = ["probe"]
    cooldown_seconds = 30

    def __init__(self, bot, *, reply: bool = True, rerecord_after: float | None = None, clock=None):
        super().__init__(bot)
        self.reply = reply
        self.rerecord_after = rerecord_after
        self.clock = clock

    async def execute(self, message):
        if self.rerecord_after is not None:
            self.clock.advance(self.rerecord_after)
            self.record_execution(message.sender_id)
        if self.reply:
            await self.send_response(message, "probe reply")
        return True


class _LegacyPluginCommand(BaseCommand):
    """A local plugin written against the old zero-argument cooldown API."""

    name = "legacy"
    keywords = ["legacy"]
    cooldown_seconds = 30

    def __init__(self, bot):
        super().__init__(bot)
        self.recorded = 0

    def _record_execution(self):  # type: ignore[override]
        self.recorded += 1
        self._last_execution_time = __import__("time").time()

    def get_remaining_cooldown(self):  # type: ignore[override]
        return 7

    async def execute(self, message):
        return True


@pytest.fixture
def harness():
    clock = _Clock()
    cfg = configparser.ConfigParser()
    cfg.add_section("Bot")
    cfg.set("Bot", "bot_name", "AcctBot")
    cfg.add_section("Channels")
    cfg.set("Channels", "monitor_channels", "general")
    cfg.set("Channels", "respond_to_dms", "true")
    bot = Mock()
    bot.config = cfg
    bot.logger = Mock()
    bot.bot_root = Path("/tmp")
    bot._local_root = None
    bot.translator = Mock()
    bot.translator.translate = Mock(side_effect=lambda key, **kw: key)
    bot.meshcore = None
    bot.connected = True
    bot.is_radio_zombie = False
    bot.is_radio_offline = False
    bot.web_viewer_integration = None
    bot.transmission_tracker = None
    with patch("modules.command_manager.PluginLoader") as loader_cls:
        loader = Mock()
        loader.load_all_plugins = Mock(return_value={})
        loader_cls.return_value = loader
        cm = CommandManager(bot)
    bot.command_manager = cm
    sent: list[str] = []

    async def _send_response(message, content, *args, **kwargs):
        cm._last_response = content
        sent.append(content)
        return True

    cm.send_response = AsyncMock(side_effect=_send_response)
    stats = Mock()
    cm.commands = {"stats": stats}
    with patch("time.time", clock), patch("modules.command_manager.asyncio.sleep", AsyncMock()):
        yield cm, bot, clock, stats, sent


def _msg(text: str = "probe", sender: str = "Ann") -> MeshMessage:
    return MeshMessage(content=text, sender_id=sender, channel="general", is_dm=False)


def _stats_rows(stats):
    return [(c.args[1], c.args[2]) for c in stats.record_command.call_args_list]


async def test_cooldown_is_recorded_before_execute(harness):
    cm, bot, clock, stats, sent = harness
    probe = _Probe(bot, clock=clock)
    cm.commands["probe"] = probe
    start = clock.now
    await cm.execute_commands(_msg())
    assert probe._user_cooldowns["Ann"] == start
    assert sent == ["probe reply"]
    assert _stats_rows(stats) == [("probe", True)]


async def test_second_record_inside_execute_moves_the_boundary(harness):
    cm, bot, clock, stats, sent = harness
    probe = _Probe(bot, clock=clock, rerecord_after=5.0)
    cm.commands["probe"] = probe
    start = clock.now
    await cm.execute_commands(_msg())
    assert probe._user_cooldowns["Ann"] == start + 5.0
    clock.advance(26)  # 31 s after the manager's record, 26 s after the command's
    allowed, remaining = probe.check_cooldown("Ann")
    assert (allowed, round(remaining)) == (False, 4)


async def test_cooldown_rejection_replies_and_records_stats(harness):
    cm, bot, clock, stats, sent = harness
    probe = _Probe(bot, clock=clock)
    cm.commands["probe"] = probe
    await cm.execute_commands(_msg())
    clock.advance(10)
    await cm.execute_commands(_msg())
    assert sent[-1] == "errors.cooldown"
    assert _stats_rows(stats) == [("probe", True), ("probe", True)]


async def test_silent_command_after_any_reply_counts_as_responded(harness):
    """Defect pinned on purpose: _last_response is never reset."""
    cm, bot, clock, stats, sent = harness
    cm.commands["probe"] = _Probe(bot, clock=clock)
    cm.commands["quiet"] = quiet = _Probe(bot, reply=False, clock=clock)
    quiet.name = "quiet"
    quiet.keywords = ["quiet"]
    await cm.execute_commands(_msg("probe"))
    await cm.execute_commands(_msg("quiet", sender="Bob"))
    assert _stats_rows(stats) == [("probe", True), ("quiet", True)]


async def test_silent_command_with_no_prior_reply_counts_as_silent(harness):
    cm, bot, clock, stats, sent = harness
    cm.commands["quiet"] = quiet = _Probe(bot, reply=False, clock=clock)
    quiet.name = "quiet"
    quiet.keywords = ["quiet"]
    await cm.execute_commands(_msg("quiet"))
    assert _stats_rows(stats) == [("quiet", False)]


async def test_zero_argument_plugin_cooldown_methods_still_work(harness):
    cm, bot, clock, stats, sent = harness
    legacy = _LegacyPluginCommand(bot)
    cm.commands["legacy"] = legacy
    await cm.execute_commands(_msg("legacy"))
    assert legacy.recorded == 1
    clock.advance(1)
    with patch.object(legacy, "can_execute_now", return_value=False):
        await cm.execute_commands(_msg("legacy"))
    assert sent[-1] == "errors.cooldown"


async def test_keyword_format_commands_are_skipped_by_execute_commands(harness):
    cm, bot, clock, stats, sent = harness
    probe = _Probe(bot, clock=clock)
    probe.get_response_format = lambda: "fmt {sender}"
    cm.commands["probe"] = probe
    await cm.execute_commands(_msg())
    assert sent == []
    assert "Ann" not in probe._user_cooldowns
    assert _stats_rows(stats) == []