add spec system: sdd/bdd/tdd loops, trace enforcement, envelope+config specs

2026-07-13 12:53:38 +02:00
parent 1879992b22
commit f1578cbd99
11 changed files with 682 additions and 0 deletions
@@ -0,0 +1,15 @@
+import re
+from pathlib import Path
+
+FEATURES_DIR = Path(__file__).resolve().parent.parent / "features"
+TAG_RE = re.compile(r"@([A-Z]{2,8}-\d{2,3})\b")
+
+
+def pytest_configure(config):
+    # pytest-bdd turns @ENV-01 tags into markers; register them so
+    # --strict-markers stays enabled (D-005).
+    ids = set()
+    for feature in FEATURES_DIR.glob("**/*.feature"):
+        ids |= set(TAG_RE.findall(feature.read_text(encoding="utf-8")))
+    for req_id in sorted(ids):
+        config.addinivalue_line("markers", f"{req_id}: spec requirement tag")
@@ -0,0 +1,147 @@
+"""BDD steps for features/envelope.feature (SPEC-001, ENV-01..09).
+
+Scenarios drive AIResponder.send() through a FakeModelResponder with
+scripted model output — no live OpenAI, no live Discord (D-002).
+"""
+
+import asyncio
+import json
+from typing import Any, Dict, List, Optional, Tuple
+
+from pytest_bdd import given, parsers, scenarios, then, when
+
+from fjerkroa_bot.ai_responder import AIMessage, AIResponder
+
+scenarios("../features/envelope.feature")
+
+
+def envelope(answer=None, answer_needed=False, channel="chat", staff=None, picture=None, picture_edit=False, hack=False) -> str:
+    return json.dumps(
+        {
+            "answer": answer,
+            "answer_needed": answer_needed,
+            "channel": channel,
+            "staff": staff,
+            "picture": picture,
+            "picture_edit": picture_edit,
+            "hack": hack,
+        }
+    )
+
+
+class FakeModelResponder(AIResponder):
+    """AIResponder with the model calls scripted away (D-002)."""
+
+    def __init__(self, config: Dict[str, Any], channel: Optional[str] = None) -> None:
+        super().__init__(config, channel)
+        self.scripted: List[str] = []
+        self.chat_calls = 0
+
+    async def chat(self, messages: List[Dict[str, Any]], limit: int) -> Tuple[Optional[Dict[str, Any]], int]:
+        self.chat_calls += 1
+        if not self.scripted:
+            return None, limit
+        return {"role": "assistant", "content": self.scripted.pop(0)}, limit
+
+    async def memory_rewrite(self, memory: str, message_user: str, answer_user: str, question: str, answer: str) -> str:
+        return memory
+
+    async def translate(self, text: str, language: str = "english") -> str:
+        return text
+
+
+@given(parsers.parse("a responder with history limit {limit:d}"), target_fixture="responder")
+def responder(limit):
+    config = {"system": "You are a test bot. {memory}", "history-limit": limit}
+    return FakeModelResponder(config, "chat")
+
+
+@given(parsers.parse('the model answers with answer "{answer}" and answer_needed "{needed}"'))
+def script_answer(responder, answer, needed):
+    responder.scripted.append(envelope(answer=answer, answer_needed=needed == "true"))
+
+
+@given(parsers.parse('the model answers with answer "{answer}" and staff note "{staff}"'))
+def script_staff(responder, answer, staff):
+    responder.scripted.append(envelope(answer=answer, answer_needed=False, staff=staff))
+
+
+@given(parsers.parse('the model answers with answer "{answer}" and channel "{channel}"'))
+def script_channel(responder, answer, channel):
+    responder.scripted.append(envelope(answer=answer, answer_needed=True, channel=channel))
+
+
+@given(parsers.parse('the model answers with answer "{answer}" and no channel'))
+def script_channel_none(responder, answer):
+    responder.scripted.append(envelope(answer=answer, answer_needed=True, channel=None))
+
+
+@given(parsers.parse('a short-path rule for channels "{chan_re}" and users "{user_re}"'))
+def short_path_rule(responder, chan_re, user_re):
+    responder.config["short-path"] = [[chan_re, user_re]]
+
+
+@given(parsers.parse('{count:d} prior history entries in channel "{channel}"'))
+def prior_history(responder, count, channel):
+    for i in range(count):
+        responder.history.append({"role": "user", "content": json.dumps({"message": f"old {i}", "channel": channel})})
+
+
+@when(parsers.parse('user "{user}" sends "{text}" in channel "{channel}"'), target_fixture="response")
+def send_message(responder, user, text, channel):
+    return asyncio.run(responder.send(AIMessage(user, text, channel)))
+
+
+@when(parsers.parse('user "{user}" sends "{text}" directly to the bot'), target_fixture="response")
+def send_direct(responder, user, text):
+    return asyncio.run(responder.send(AIMessage(user, text, "chat", direct=True)))
+
+
+@then(parsers.parse('the response answer contains "{text}"'))
+def answer_contains(response, text):
+    assert response.answer is not None and text in response.answer
+
+
+@then(parsers.parse('the response answer does not contain "{text}"'))
+def answer_not_contains(response, text):
+    assert response.answer is not None and text not in response.answer
+
+
+@then("the response is marked as needed")
+def is_needed(response):
+    assert response.answer_needed is True
+
+
+@then("the response is not marked as needed")
+def not_needed(response):
+    assert response.answer_needed is False
+
+
+@then(parsers.parse('the response staff note is "{text}"'))
+def staff_note_is(response, text):
+    assert response.staff == text
+
+
+@then(parsers.parse('the response channel is "{channel}"'))
+def channel_is(response, channel):
+    assert response.channel == channel
+
+
+@then("the model was not called")
+def model_not_called(responder):
+    assert responder.chat_calls == 0
+
+
+@then("the response is empty")
+def response_empty(response):
+    assert response.answer is None and response.answer_needed is False
+
+
+@then(parsers.parse('the history contains the message from "{user}"'))
+def history_has_user(responder, user):
+    assert any(f'"user": "{user}"' in item["content"] for item in responder.history)
+
+
+@then(parsers.parse("the history length is at most {limit:d}"))
+def history_at_most(responder, limit):
+    assert len(responder.history) <= limit
@@ -0,0 +1,43 @@
+"""Unit coverage for SPEC-008 (CFG-01..03)."""
+
+import unittest
+from unittest.mock import mock_open, patch
+
+import toml
+
+from fjerkroa_bot import FjerkroaBot
+from fjerkroa_bot.ai_responder import AIMessage, AIResponder
+
+
+class TestConfigLoad(unittest.TestCase):
+    def test_load_config_parses_toml(self):
+        """CFG-01: load_config parses the TOML file into a plain dict."""
+        data = {"system": "prompt", "history-limit": 7, "additional-responders": []}
+        with patch("builtins.open", mock_open(read_data=toml.dumps(data))):
+            result = FjerkroaBot.load_config("config.toml")
+        self.assertEqual(result, data)
+
+
+class TestPerChannelPrompt(unittest.TestCase):
+    def test_channel_override_wins(self):
+        """CFG-02: responder bound to a channel uses config[channel] over config['system']."""
+        config = {"system": "Default prompt", "kitchen": "Kitchen prompt", "history-limit": 5}
+        responder = AIResponder(config, "kitchen")
+        system = responder.message(AIMessage("alice", "hei", "kitchen"))[0]["content"]
+        self.assertTrue(system.startswith("Kitchen prompt"))
+
+    def test_fallback_to_system(self):
+        """CFG-02: without a channel key the shared system prompt is used."""
+        config = {"system": "Default prompt", "history-limit": 5}
+        responder = AIResponder(config, "kitchen")
+        system = responder.message(AIMessage("alice", "hei", "kitchen"))[0]["content"]
+        self.assertTrue(system.startswith("Default prompt"))
+
+
+class TestNewsFileMissing(unittest.TestCase):
+    def test_missing_news_file_keeps_placeholder(self):
+        """CFG-03: nonexistent news file -> {news} placeholder stays literal, no crash."""
+        config = {"system": "N: {news}", "history-limit": 5, "news": "/nonexistent/news.txt"}
+        responder = AIResponder(config, "chat")
+        system = responder.message(AIMessage("alice", "hei"))[0]["content"]
+        self.assertIn("{news}", system)
@@ -0,0 +1,56 @@
+"""Unit coverage for SPEC-001 test-class requirements (ENV-10..12)."""
+
+import json
+import time
+import unittest
+
+from fjerkroa_bot.ai_responder import AIMessage, AIResponder
+
+from .test_bdd_envelope import FakeModelResponder
+
+
+def entry(channel: str, text: str = "x"):
+    return {"role": "user", "content": json.dumps({"message": text, "channel": channel})}
+
+
+class TestSystemPromptTemplate(unittest.TestCase):
+    def test_template_substitution(self):
+        """ENV-10: {date}/{memory} substituted; missing news file leaves {news} literal."""
+        config = {"system": "Date {date} memory {memory} news {news}", "history-limit": 5, "news": "/nonexistent/news.txt"}
+        responder = AIResponder(config, "chat")
+        responder.memory = "MEMSTR"
+        messages = responder.message(AIMessage("alice", "hei"))
+        system = messages[0]["content"]
+        self.assertIn(time.strftime("%Y-%m-%d"), system)
+        self.assertIn("MEMSTR", system)
+        self.assertIn("{news}", system)  # left literal — file does not exist
+
+
+class TestHistoryShrink(unittest.TestCase):
+    def test_shrink_prefers_busy_channel(self):
+        """ENV-11: entry from the channel exceeding history-per-channel is removed first."""
+        config = {"system": "s", "history-limit": 4}
+        responder = AIResponder(config, "chat")
+        responder.history = [entry("a", "a0"), entry("a", "a1"), entry("a", "a2"), entry("a", "a3"), entry("b", "b0")]
+        responder.shrink_history_by_one()
+        self.assertEqual(len(responder.history), 4)
+        self.assertNotIn("a0", responder.history[0]["content"])  # oldest busy-channel entry gone
+
+    def test_shrink_falls_back_to_oldest(self):
+        """ENV-11: when no channel exceeds the cap, the oldest entry overall is removed."""
+        config = {"system": "s", "history-limit": 4}
+        responder = AIResponder(config, "chat")
+        responder.history = [entry("a", "a0"), entry("b", "b0"), entry("c", "c0")]
+        responder.shrink_history_by_one()
+        self.assertEqual(len(responder.history), 2)
+        self.assertNotIn("a0", responder.history[0]["content"])
+
+
+class TestRetriesExhausted(unittest.IsolatedAsyncioTestCase):
+    async def test_send_raises_after_three_failures(self):
+        """ENV-12: three model failures -> RuntimeError (revision pending in FDB-004/D1)."""
+        responder = FakeModelResponder({"system": "s", "history-limit": 5}, "chat")
+        # scripted list empty -> chat() returns None every attempt
+        with self.assertRaises(RuntimeError):
+            await responder.send(AIMessage("alice", "hei", "chat"))
+        self.assertEqual(responder.chat_calls, 3)