Oleksandr Kozachuk
211 lines
10 KiB
Python
211 lines
10 KiB
Python
"""Unit coverage for SPEC-010 human behavior (BEH-01..08) + ENV-20, SAF-10, OPS-11."""
|
|
|
|
import hashlib
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
from unittest.mock import AsyncMock, MagicMock, Mock, patch
|
|
|
|
from discord import DMChannel, TextChannel
|
|
|
|
from fjerkroa_bot.ai_responder import AIMessage, AIResponse
|
|
from fjerkroa_bot.discord_bot import quiet_hours_active, split_answer
|
|
from fjerkroa_bot.openai_responder import OpenAIResponder
|
|
from fjerkroa_bot.persistence import PersistentStore
|
|
|
|
from .test_bdd_envelope import FakeModelResponder, envelope
|
|
from .test_spec_ops import OpsBase
|
|
|
|
|
|
class ClassifierGateBase(OpsBase):
|
|
def gate_setup(self, verdict):
|
|
self.bot.config["classifier-model"] = "gpt-5.6-luna"
|
|
self.bot.airesponder.classify = AsyncMock(return_value=verdict)
|
|
self.bot.respond = AsyncMock()
|
|
|
|
|
|
class TestClassifierGate(ClassifierGateBase):
|
|
async def test_no_reply_means_no_model_call(self):
|
|
"""BEH-01: classifier reply=false on a non-direct message -> respond() never runs."""
|
|
self.gate_setup({"reply": False, "factual": False, "emoji": None})
|
|
await self.bot.on_message(self.public_msg("just chatting with bob"))
|
|
self.bot.airesponder.classify.assert_awaited_once()
|
|
self.bot.respond.assert_not_awaited()
|
|
|
|
async def test_reply_true_passes_through_with_factual_flag(self):
|
|
"""BEH-01: reply=true proceeds; factual flag is forwarded."""
|
|
self.gate_setup({"reply": True, "factual": True, "emoji": None})
|
|
await self.bot.on_message(self.public_msg("når har dere åpent?"))
|
|
self.bot.respond.assert_awaited_once()
|
|
self.assertTrue(self.bot.respond.await_args.kwargs.get("factual"))
|
|
|
|
async def test_direct_message_bypasses_gate(self):
|
|
"""BEH-02: DMs never touch the classifier."""
|
|
self.gate_setup({"reply": False, "factual": False, "emoji": None})
|
|
message = self.public_msg("hei bot")
|
|
message.channel = MagicMock(spec=DMChannel)
|
|
message.channel.recipient = None
|
|
await self.bot.on_message(message)
|
|
self.bot.airesponder.classify.assert_not_awaited()
|
|
self.bot.respond.assert_awaited_once()
|
|
|
|
async def test_classifier_failure_fails_open(self):
|
|
"""BEH-03: classify() -> None falls through to the main model."""
|
|
self.gate_setup(None)
|
|
await self.bot.on_message(self.public_msg("hello?"))
|
|
self.bot.respond.assert_awaited_once()
|
|
|
|
async def test_reaction_instead_of_reply(self):
|
|
"""BEH-07: reply=false + emoji -> reaction on the message, no model call."""
|
|
self.gate_setup({"reply": False, "factual": False, "emoji": "👍"})
|
|
message = self.public_msg("gg everyone")
|
|
message.add_reaction = AsyncMock()
|
|
await self.bot.on_message(message)
|
|
message.add_reaction.assert_awaited_once_with("👍")
|
|
self.bot.respond.assert_not_awaited()
|
|
|
|
|
|
class TestTypingPacing(OpsBase):
|
|
async def send_with(self, answer, factual, cps=30):
|
|
if cps is not None:
|
|
self.bot.config["typing-chars-per-second"] = cps
|
|
response = AIResponse(answer, True, "chat", None, None, False, False)
|
|
channel = MagicMock(spec=TextChannel)
|
|
channel.send = AsyncMock()
|
|
with patch("fjerkroa_bot.discord_bot.asyncio.sleep", new_callable=AsyncMock) as sleep:
|
|
await self.bot.send_answer_with_typing(response, channel, self.bot.airesponder, factual=factual)
|
|
return sleep, channel
|
|
|
|
async def test_delay_proportional_and_capped(self):
|
|
"""BEH-04: delay = len/cps capped at typing-max-seconds."""
|
|
sleep, _ = await self.send_with("x" * 300, factual=False, cps=30)
|
|
sleep.assert_awaited_once_with(8.0) # 300/30=10 -> cap 8
|
|
|
|
async def test_factual_skips_delay(self):
|
|
"""BEH-05: factual answers go out instantly."""
|
|
sleep, _ = await self.send_with("x" * 300, factual=True, cps=30)
|
|
sleep.assert_not_awaited()
|
|
|
|
async def test_no_knob_no_delay(self):
|
|
"""BEH-04: without typing-chars-per-second there is no pacing."""
|
|
sleep, _ = await self.send_with("x" * 300, factual=False, cps=None)
|
|
sleep.assert_not_awaited()
|
|
|
|
|
|
class TestSplitting(unittest.TestCase):
|
|
def test_split_at_paragraphs_under_limit(self):
|
|
"""BEH-06: long answers split at paragraph boundaries, each under 2000."""
|
|
text = "\n\n".join(["Avsnitt " + str(i) + " " + "x" * 700 for i in range(4)])
|
|
parts = split_answer(text, threshold=1200, max_parts=3)
|
|
self.assertGreaterEqual(len(parts), 2)
|
|
self.assertLessEqual(len(parts), 3)
|
|
for part in parts:
|
|
self.assertLessEqual(len(part), 2000)
|
|
self.assertEqual("\n\n".join(parts).replace("\n\n", ""), text.replace("\n\n", ""))
|
|
|
|
def test_short_answers_untouched(self):
|
|
"""BEH-06: short answers stay a single message."""
|
|
self.assertEqual(split_answer("kort svar", 1200, 3), ["kort svar"])
|
|
|
|
def test_oversized_single_block_hard_split(self):
|
|
"""BEH-06: a single block over 2000 chars is hard-split under the Discord limit."""
|
|
parts = split_answer("y" * 4500, 1200, 3)
|
|
for part in parts:
|
|
self.assertLessEqual(len(part), 2000)
|
|
self.assertEqual(sum(len(p) for p in parts), 4500)
|
|
|
|
|
|
class TestSplitSends(OpsBase):
|
|
async def test_parts_sent_in_order_files_last(self):
|
|
"""BEH-06: parts sent sequentially; image files ride on the last part."""
|
|
self.bot.config["split-threshold"] = 50
|
|
answer = "Første del.\n\nAndre del som også er ganske lang her."
|
|
response = AIResponse(answer, True, "chat", None, "a cat", False, False)
|
|
self.bot.airesponder.draw = AsyncMock(return_value=[__import__("io").BytesIO(b"png")])
|
|
channel = MagicMock(spec=TextChannel)
|
|
channel.send = AsyncMock()
|
|
await self.bot.send_answer_with_typing(response, channel, self.bot.airesponder, factual=True)
|
|
self.assertEqual(channel.send.await_count, 2)
|
|
first_kwargs = channel.send.await_args_list[0].kwargs
|
|
last_kwargs = channel.send.await_args_list[1].kwargs
|
|
self.assertIsNone(first_kwargs.get("files"))
|
|
self.assertIsNotNone(last_kwargs.get("files"))
|
|
|
|
|
|
class TestQuietHours(OpsBase):
|
|
def test_quiet_hours_parsing(self):
|
|
"""BEH-08: window logic incl. midnight wrap."""
|
|
self.assertTrue(quiet_hours_active("23:00-08:00", "23:30"))
|
|
self.assertTrue(quiet_hours_active("23:00-08:00", "07:59"))
|
|
self.assertFalse(quiet_hours_active("23:00-08:00", "12:00"))
|
|
self.assertTrue(quiet_hours_active("13:00-15:00", "14:00"))
|
|
self.assertFalse(quiet_hours_active("13:00-15:00", "15:00"))
|
|
self.assertFalse(quiet_hours_active(None, "14:00"))
|
|
self.assertFalse(quiet_hours_active("garbage", "14:00"))
|
|
|
|
async def test_quiet_hours_block_bot_initiated(self):
|
|
"""BEH-08: inside the window bot_initiated_allowed is false, replies unaffected."""
|
|
self.bot.config["quiet-hours"] = "00:00-23:59"
|
|
self.assertFalse(self.bot.bot_initiated_allowed())
|
|
self.assertTrue(self.bot.replies_allowed())
|
|
|
|
|
|
class TestPromptPrefixStability(unittest.IsolatedAsyncioTestCase):
|
|
def test_persona_prefix_stable_and_context_suffix(self):
|
|
"""ENV-20 + ENV-10: byte-stable persona prefix; date/memory in the context suffix."""
|
|
config = {"system": "Du er Fjærkroa. I dag er {date} kl {time}. {news} {memory}", "history-limit": 5}
|
|
responder = FakeModelResponder(config, "chat")
|
|
responder.memory = "MEMSTR"
|
|
first = responder.message(AIMessage("alice", "hei"))[0]["content"]
|
|
second = responder.message(AIMessage("bob", "hallo"))[0]["content"]
|
|
self.assertIn("## Context", first)
|
|
prefix_one = first.split("## Context")[0]
|
|
prefix_two = second.split("## Context")[0]
|
|
self.assertEqual(prefix_one, prefix_two)
|
|
self.assertNotIn("{date}", first)
|
|
self.assertNotIn("{memory}", first)
|
|
suffix = first.split("## Context")[1]
|
|
self.assertIn("MEMSTR", suffix)
|
|
import time as _time
|
|
|
|
self.assertIn(_time.strftime("%Y-%m-%d"), suffix)
|
|
|
|
def test_missing_news_file_no_placeholder(self):
|
|
"""CFG-03 (revised): missing news file -> no literal placeholder, no crash."""
|
|
config = {"system": "N: {news}", "history-limit": 5, "news": "/nonexistent/news.txt"}
|
|
responder = FakeModelResponder(config, "chat")
|
|
system = responder.message(AIMessage("alice", "hei"))[0]["content"]
|
|
self.assertNotIn("{news}", system)
|
|
self.assertNotIn("news:", system.split("## Context")[1])
|
|
|
|
|
|
class TestSafetyIdentifier(unittest.IsolatedAsyncioTestCase):
|
|
async def test_chat_carries_hashed_user(self):
|
|
"""SAF-10: chat calls pass safety_identifier = sha256(user)[:16], never the raw name."""
|
|
responder = OpenAIResponder({"openai-token": "t", "model": "m", "system": "s", "history-limit": 5}, "chat")
|
|
message = Mock(content=envelope(answer="x", answer_needed=True), role="assistant", tool_calls=None, refusal=None)
|
|
with patch("fjerkroa_bot.openai_responder.openai_chat", new_callable=AsyncMock) as chat_mock:
|
|
chat_mock.return_value = Mock(choices=[Mock(message=message)], usage="usage")
|
|
payload = '{"user": "alice", "message": "hei", "channel": "chat", "direct": false, "historise_question": true}'
|
|
await responder.chat([{"role": "user", "content": payload}], 10)
|
|
identifier = chat_mock.await_args.kwargs.get("safety_identifier")
|
|
expected = "discord-" + hashlib.sha256(b"alice").hexdigest()[:16]
|
|
self.assertEqual(identifier, expected)
|
|
self.assertNotIn("alice", identifier)
|
|
|
|
|
|
class TestPinsListing(OpsBase):
|
|
async def test_pins_command_lists_ids(self):
|
|
"""OPS-11: !bot pins lists pinned facts with ids."""
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
store = PersistentStore(Path(tmp) / "bot.db")
|
|
self.bot.airesponder.store = store
|
|
store.add_pinned(None, "Åpningstider: tirsdag-fredag 12-17")
|
|
store.add_pinned("chat", "Kanalregel: norsk")
|
|
await self.bot.on_message(self.staff_msg("!bot pins"))
|
|
listing = self.bot.staff_channel.send.await_args.args[0]
|
|
self.assertIn("Åpningstider", listing)
|
|
self.assertIn("Kanalregel", listing)
|
|
self.assertIn("1", listing)
|
|
self.assertIn("2", listing)
|