Oleksandr Kozachuk
136 lines
6.0 KiB
Python
136 lines
6.0 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())
|