206 lines
7.4 KiB
Python
206 lines
7.4 KiB
Python
from __future__ import annotations
|
|
|
|
from unittest.mock import AsyncMock, patch
|
|
|
|
from asgiref.sync import async_to_sync
|
|
from django.test import TransactionTestCase, override_settings
|
|
|
|
from core.commands.base import CommandContext
|
|
from core.commands.handlers.bp import BPCommandHandler, parse_bp_subcommand
|
|
from core.models import (
|
|
BusinessPlanDocument,
|
|
ChatSession,
|
|
CommandAction,
|
|
CommandChannelBinding,
|
|
CommandProfile,
|
|
Message,
|
|
Person,
|
|
PersonIdentifier,
|
|
User,
|
|
)
|
|
|
|
|
|
@override_settings(BP_SUBCOMMANDS_V1=True)
|
|
class BPSubcommandTests(TransactionTestCase):
|
|
def setUp(self):
|
|
self.user = User.objects.create_user(
|
|
username="bp-sub-user",
|
|
email="bp-sub@example.com",
|
|
password="x",
|
|
)
|
|
self.person = Person.objects.create(user=self.user, name="Sub Person")
|
|
self.identifier = PersonIdentifier.objects.create(
|
|
user=self.user,
|
|
person=self.person,
|
|
service="whatsapp",
|
|
identifier="120363402761690215",
|
|
)
|
|
self.session = ChatSession.objects.create(user=self.user, identifier=self.identifier)
|
|
self.profile = CommandProfile.objects.create(
|
|
user=self.user,
|
|
slug="bp",
|
|
name="Business Plan",
|
|
enabled=True,
|
|
trigger_token="#bp#",
|
|
reply_required=True,
|
|
exact_match_only=True,
|
|
)
|
|
CommandChannelBinding.objects.create(
|
|
profile=self.profile,
|
|
direction="ingress",
|
|
service="whatsapp",
|
|
channel_identifier="120363402761690215",
|
|
enabled=True,
|
|
)
|
|
for action_type, position in (("extract_bp", 0), ("save_document", 1)):
|
|
CommandAction.objects.create(
|
|
profile=self.profile,
|
|
action_type=action_type,
|
|
enabled=True,
|
|
position=position,
|
|
)
|
|
|
|
def _ctx(self, trigger: Message, text: str) -> CommandContext:
|
|
return CommandContext(
|
|
service="whatsapp",
|
|
channel_identifier="120363402761690215",
|
|
message_id=str(trigger.id),
|
|
user_id=self.user.id,
|
|
message_text=text,
|
|
payload={},
|
|
)
|
|
|
|
def test_parser_detects_set_and_remainder(self):
|
|
parsed = parse_bp_subcommand(" #BP set# addendum text ")
|
|
self.assertEqual("set", parsed.command)
|
|
self.assertEqual("addendum text", parsed.remainder_text)
|
|
|
|
def test_parser_detects_set_range(self):
|
|
parsed = parse_bp_subcommand("#bp set range# now")
|
|
self.assertEqual("set_range", parsed.command)
|
|
|
|
def test_set_standalone_uses_remainder_only(self):
|
|
trigger = Message.objects.create(
|
|
user=self.user,
|
|
session=self.session,
|
|
sender_uuid="me",
|
|
text="#bp set# direct body",
|
|
ts=1000,
|
|
source_service="whatsapp",
|
|
source_chat_id="120363402761690215",
|
|
)
|
|
with patch("core.commands.handlers.bp.ai_runner.run_prompt", new=AsyncMock()) as mocked_ai:
|
|
result = async_to_sync(BPCommandHandler().execute)(self._ctx(trigger, trigger.text))
|
|
self.assertTrue(result.ok)
|
|
mocked_ai.assert_not_awaited()
|
|
doc = BusinessPlanDocument.objects.get(trigger_message=trigger)
|
|
self.assertEqual("direct body", doc.content_markdown)
|
|
self.assertEqual("Generated from 1 message.", doc.structured_payload.get("annotation"))
|
|
|
|
def test_set_reply_only_uses_anchor(self):
|
|
anchor = Message.objects.create(
|
|
user=self.user,
|
|
session=self.session,
|
|
sender_uuid="peer",
|
|
text="anchor body",
|
|
ts=1000,
|
|
source_service="whatsapp",
|
|
source_chat_id="120363402761690215",
|
|
)
|
|
trigger = Message.objects.create(
|
|
user=self.user,
|
|
session=self.session,
|
|
sender_uuid="me",
|
|
text="#bp set#",
|
|
ts=2000,
|
|
source_service="whatsapp",
|
|
source_chat_id="120363402761690215",
|
|
reply_to=anchor,
|
|
)
|
|
result = async_to_sync(BPCommandHandler().execute)(self._ctx(trigger, trigger.text))
|
|
self.assertTrue(result.ok)
|
|
doc = BusinessPlanDocument.objects.get(trigger_message=trigger)
|
|
self.assertEqual("anchor body", doc.content_markdown)
|
|
self.assertEqual("Generated from 1 message.", doc.structured_payload.get("annotation"))
|
|
|
|
def test_set_reply_plus_addendum_uses_divider(self):
|
|
anchor = Message.objects.create(
|
|
user=self.user,
|
|
session=self.session,
|
|
sender_uuid="peer",
|
|
text="base body",
|
|
ts=1000,
|
|
source_service="whatsapp",
|
|
source_chat_id="120363402761690215",
|
|
)
|
|
trigger = Message.objects.create(
|
|
user=self.user,
|
|
session=self.session,
|
|
sender_uuid="me",
|
|
text="#bp set# extra text",
|
|
ts=2000,
|
|
source_service="whatsapp",
|
|
source_chat_id="120363402761690215",
|
|
reply_to=anchor,
|
|
)
|
|
result = async_to_sync(BPCommandHandler().execute)(self._ctx(trigger, trigger.text))
|
|
self.assertTrue(result.ok)
|
|
doc = BusinessPlanDocument.objects.get(trigger_message=trigger)
|
|
self.assertIn("base body", doc.content_markdown)
|
|
self.assertIn("--- Addendum (newer message text) ---", doc.content_markdown)
|
|
self.assertIn("extra text", doc.content_markdown)
|
|
self.assertEqual(
|
|
"Generated from 1 message + 1 addendum.",
|
|
doc.structured_payload.get("annotation"),
|
|
)
|
|
|
|
def test_set_range_requires_reply(self):
|
|
trigger = Message.objects.create(
|
|
user=self.user,
|
|
session=self.session,
|
|
sender_uuid="me",
|
|
text="#bp set range#",
|
|
ts=3000,
|
|
source_service="whatsapp",
|
|
source_chat_id="120363402761690215",
|
|
)
|
|
result = async_to_sync(BPCommandHandler().execute)(self._ctx(trigger, trigger.text))
|
|
self.assertFalse(result.ok)
|
|
self.assertEqual("failed", result.status)
|
|
self.assertEqual("bp_set_range_requires_reply_target", result.error)
|
|
|
|
def test_set_range_exports_text_only_lines(self):
|
|
anchor = Message.objects.create(
|
|
user=self.user,
|
|
session=self.session,
|
|
sender_uuid="peer",
|
|
text="line 1",
|
|
ts=1000,
|
|
source_service="whatsapp",
|
|
source_chat_id="120363402761690215",
|
|
)
|
|
Message.objects.create(
|
|
user=self.user,
|
|
session=self.session,
|
|
sender_uuid="me",
|
|
text="",
|
|
ts=1500,
|
|
source_service="whatsapp",
|
|
source_chat_id="120363402761690215",
|
|
)
|
|
trigger = Message.objects.create(
|
|
user=self.user,
|
|
session=self.session,
|
|
sender_uuid="me",
|
|
text="#bp set range#",
|
|
ts=2000,
|
|
source_service="whatsapp",
|
|
source_chat_id="120363402761690215",
|
|
reply_to=anchor,
|
|
)
|
|
result = async_to_sync(BPCommandHandler().execute)(self._ctx(trigger, trigger.text))
|
|
self.assertTrue(result.ok)
|
|
doc = BusinessPlanDocument.objects.get(trigger_message=trigger)
|
|
self.assertEqual("line 1\n(no text)\n#bp set range#", doc.content_markdown)
|
|
self.assertEqual("Generated from 3 messages.", doc.structured_payload.get("annotation"))
|