Oleksandr Kozachuk
202 lines
8.6 KiB
Python
202 lines
8.6 KiB
Python
"""Unit coverage for SPEC-006 operator controls (OPS-01..09)."""
|
|
|
|
import time
|
|
from unittest.mock import AsyncMock, MagicMock
|
|
|
|
from discord import Message, TextChannel, User
|
|
|
|
from fjerkroa_bot.ai_responder import AIMessage, AIResponse
|
|
|
|
from .test_main import TestBotBase
|
|
|
|
|
|
class OpsBase(TestBotBase):
|
|
def staff_msg(self, content: str) -> Message:
|
|
message = MagicMock(spec=Message)
|
|
message.content = content
|
|
message.author = MagicMock(spec=User)
|
|
message.author.bot = False
|
|
message.author.id = 999
|
|
message.channel = self.bot.staff_channel
|
|
return message
|
|
|
|
def public_msg(self, content: str) -> Message:
|
|
message = self.create_message(content)
|
|
message.content = content
|
|
return message
|
|
|
|
|
|
class TestStaffCommandAuth(OpsBase):
|
|
async def test_commands_ignored_outside_staff_channel(self):
|
|
"""OPS-01: !bot commands in a public channel do not flip flags."""
|
|
self.bot.handle_message_through_responder = AsyncMock()
|
|
await self.bot.on_message(self.public_msg("!bot pause"))
|
|
self.assertTrue(self.bot.replies_enabled)
|
|
self.bot.handle_message_through_responder.assert_awaited_once()
|
|
|
|
async def test_commands_honored_in_staff_channel(self):
|
|
"""OPS-01: !bot commands in the staff channel are executed."""
|
|
await self.bot.on_message(self.staff_msg("!bot pause"))
|
|
self.assertFalse(self.bot.replies_enabled)
|
|
|
|
|
|
class TestPauseResume(OpsBase):
|
|
async def test_pause_blocks_public_replies(self):
|
|
"""OPS-02: paused bot ignores public messages; resume restores replying."""
|
|
self.bot.handle_message_through_responder = AsyncMock()
|
|
await self.bot.on_message(self.staff_msg("!bot pause"))
|
|
await self.bot.on_message(self.public_msg("hello?"))
|
|
self.bot.handle_message_through_responder.assert_not_awaited()
|
|
await self.bot.on_message(self.staff_msg("!bot resume"))
|
|
self.assertTrue(self.bot.replies_enabled)
|
|
await self.bot.on_message(self.public_msg("hello again"))
|
|
self.bot.handle_message_through_responder.assert_awaited_once()
|
|
|
|
|
|
class TestImageKillSwitch(OpsBase):
|
|
async def test_images_off_strips_picture(self):
|
|
"""OPS-03: with images off the picture request is dropped, answer still sent."""
|
|
await self.bot.on_message(self.staff_msg("!bot images off"))
|
|
self.assertFalse(self.bot.images_enabled)
|
|
response = AIResponse("here is your cat", True, "chat", None, "a cat", False, False)
|
|
self.bot.send_message_with_typing = AsyncMock(return_value=response)
|
|
self.bot.send_answer_with_typing = AsyncMock()
|
|
origin = MagicMock(spec=TextChannel)
|
|
await self.bot.respond(AIMessage("alice", "draw a cat", "chat"), origin)
|
|
sent = self.bot.send_answer_with_typing.await_args.args[0]
|
|
self.assertIsNone(sent.picture)
|
|
self.assertEqual(sent.answer, "here is your cat")
|
|
|
|
|
|
class TestQuietMode(OpsBase):
|
|
async def test_quiet_pauses_then_auto_resumes(self):
|
|
"""OPS-04: !bot quiet N silences replies for N minutes, then auto-resumes."""
|
|
await self.bot.on_message(self.staff_msg("!bot quiet 10"))
|
|
self.assertFalse(self.bot.replies_allowed())
|
|
self.bot.quiet_until = time.monotonic() - 1
|
|
self.assertTrue(self.bot.replies_allowed())
|
|
|
|
|
|
class TestStatus(OpsBase):
|
|
async def test_status_reports_flags(self):
|
|
"""OPS-05: !bot status answers in the staff channel with the flag state."""
|
|
await self.bot.on_message(self.staff_msg("!bot status"))
|
|
self.bot.staff_channel.send.assert_awaited()
|
|
text = self.bot.staff_channel.send.await_args.args[0]
|
|
self.assertIn("replies", text)
|
|
self.assertIn("images", text)
|
|
self.assertIn("tasks", text)
|
|
|
|
|
|
class TestKeywordAlerts(OpsBase):
|
|
async def test_keyword_forces_staff_alert(self):
|
|
"""OPS-06: staff-alert-keywords match forces an alert when the model set none."""
|
|
self.bot.config["staff-alert-keywords"] = ["(?i)hjelp|help"]
|
|
response = AIResponse("ok", True, "chat", None, None, False, False)
|
|
self.bot.send_message_with_typing = AsyncMock(return_value=response)
|
|
self.bot.send_answer_with_typing = AsyncMock()
|
|
await self.bot.respond(AIMessage("guest", "HELP at table 4", "chat"), MagicMock(spec=TextChannel))
|
|
self.bot.staff_channel.send.assert_awaited_once()
|
|
self.assertIn("guest", self.bot.staff_channel.send.await_args.args[0])
|
|
|
|
|
|
class TestAlertRateLimit(OpsBase):
|
|
async def test_alerts_rate_limited(self):
|
|
"""OPS-07: staff alerts above the hourly cap are logged, not sent."""
|
|
self.bot.config["staff-alert-max-per-hour"] = 2
|
|
await self.bot.send_staff_alert("one")
|
|
await self.bot.send_staff_alert("two")
|
|
await self.bot.send_staff_alert("three")
|
|
self.assertEqual(self.bot.staff_channel.send.await_count, 2)
|
|
|
|
|
|
class TestAlertFallback(OpsBase):
|
|
async def test_lost_alert_is_logged_not_raised(self):
|
|
"""OPS-08: no staff channel -> alert goes to the error log, no crash."""
|
|
self.bot.staff_channel = None
|
|
with self.assertLogs(level="ERROR") as logs:
|
|
await self.bot.send_staff_alert("nobody hears this")
|
|
self.assertTrue(any("nobody hears this" in line for line in logs.output))
|
|
|
|
|
|
class TestTasksKillSwitch(OpsBase):
|
|
async def test_tasks_off_blocks_bot_initiated(self):
|
|
"""OPS-09: !bot tasks off disables bot-initiated posting; on restores."""
|
|
self.assertTrue(self.bot.bot_initiated_allowed())
|
|
await self.bot.on_message(self.staff_msg("!bot tasks off"))
|
|
self.assertFalse(self.bot.tasks_enabled)
|
|
self.assertFalse(self.bot.bot_initiated_allowed())
|
|
await self.bot.on_message(self.staff_msg("!bot tasks on"))
|
|
self.assertTrue(self.bot.bot_initiated_allowed())
|
|
|
|
async def test_pause_also_blocks_bot_initiated(self):
|
|
"""OPS-09: bot-initiated posts respect pause/quiet."""
|
|
await self.bot.on_message(self.staff_msg("!bot pause"))
|
|
self.assertFalse(self.bot.bot_initiated_allowed())
|
|
|
|
|
|
class TestHelp(OpsBase):
|
|
STAFF_CMDS = (
|
|
"pause",
|
|
"resume",
|
|
"quiet <minutes>",
|
|
"status",
|
|
"spend",
|
|
"images on|off",
|
|
"memory <user>",
|
|
"forget-fact <id>",
|
|
"pin <channel|global>",
|
|
"unpin <id>",
|
|
"pins",
|
|
"task-approve <id>",
|
|
"task-cancel <id>",
|
|
"(list)",
|
|
)
|
|
|
|
def test_staff_help_is_complete_and_grouped(self):
|
|
"""OPS-17: staff help lists every operator command, grouped by purpose."""
|
|
text = self.bot._help_text(staff=True)
|
|
for cmd in self.STAFF_CMDS:
|
|
self.assertIn(cmd, text, f"missing {cmd!r} in staff help")
|
|
for group in ("Control:", "Cost:", "Memory:", "Tasks:"):
|
|
self.assertIn(group, text)
|
|
for cmd in ("!help", "!forgetme", "!privacy", "!wichtel"):
|
|
self.assertIn(cmd, text) # everywhere-commands shown too
|
|
|
|
def test_user_help_hides_operator_commands(self):
|
|
"""OPS-17: non-staff help shows only the everyone-commands."""
|
|
text = self.bot._help_text(staff=False)
|
|
for cmd in ("!help", "!forgetme", "!privacy", "!wichtel"):
|
|
self.assertIn(cmd, text)
|
|
for op in ("task-approve", "images on|off", "spend", "Staff commands", "Control:"):
|
|
self.assertNotIn(op, text)
|
|
|
|
async def test_bot_help_in_staff_channel_returns_full_help(self):
|
|
"""OPS-17: `!bot help` answers with the complete staff help."""
|
|
await self.bot.on_message(self.staff_msg("!bot help"))
|
|
text = self.bot.staff_channel.send.await_args.args[0]
|
|
self.assertIn("Staff commands", text)
|
|
self.assertIn("task-cancel <id>", text)
|
|
|
|
async def test_unknown_bot_command_falls_back_to_help(self):
|
|
"""OPS-17: an unrecognised `!bot` command shows the full help, not a partial line."""
|
|
await self.bot.on_message(self.staff_msg("!bot wat"))
|
|
text = self.bot.staff_channel.send.await_args.args[0]
|
|
self.assertIn("Control:", text)
|
|
|
|
async def test_help_in_public_channel_is_user_scoped(self):
|
|
"""OPS-17: `!help` in a normal channel lists only everyone-commands."""
|
|
msg = self.public_msg("!help")
|
|
await self.bot.on_message(msg)
|
|
text = msg.channel.send.await_args.args[0]
|
|
self.assertIn("!forgetme", text)
|
|
self.assertNotIn("Staff commands", text)
|
|
self.assertNotIn("task-approve", text)
|
|
|
|
async def test_help_works_while_paused(self):
|
|
"""OPS-17: help answers even when replies are paused."""
|
|
await self.bot.on_message(self.staff_msg("!bot pause"))
|
|
msg = self.public_msg("!help")
|
|
await self.bot.on_message(msg)
|
|
msg.channel.send.assert_awaited()
|