Oleksandr Kozachuk
148 lines
5.1 KiB
Python
148 lines
5.1 KiB
Python
"""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 consolidate(self, observations, known_facts):
|
|
return {"facts": [], "episode": None}
|
|
|
|
async def classify(self, message, history_tail):
|
|
return getattr(self, "scripted_classification", None)
|
|
|
|
|
|
@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
|