109 lines
3.5 KiB
Python
109 lines
3.5 KiB
Python
from __future__ import annotations
|
||
|
||
from types import SimpleNamespace
|
||
|
||
from adapter.matrix.converter import from_command, from_reaction, from_room_event
|
||
from core.protocol import IncomingCallback, IncomingCommand, IncomingMessage
|
||
|
||
|
||
def text_event(body: str, sender: str = "@a:m.org", event_id: str = "$e1"):
|
||
return SimpleNamespace(
|
||
sender=sender, body=body, event_id=event_id, msgtype="m.text", replyto_event_id=None
|
||
)
|
||
|
||
|
||
def file_event(url: str = "mxc://x/y", filename: str = "doc.pdf", mime: str = "application/pdf"):
|
||
return SimpleNamespace(
|
||
sender="@a:m.org",
|
||
body=filename,
|
||
event_id="$e2",
|
||
msgtype="m.file",
|
||
replyto_event_id=None,
|
||
url=url,
|
||
mimetype=mime,
|
||
)
|
||
|
||
|
||
def image_event(url: str = "mxc://x/img", mime: str = "image/jpeg"):
|
||
return SimpleNamespace(
|
||
sender="@a:m.org",
|
||
body="img.jpg",
|
||
event_id="$e3",
|
||
msgtype="m.image",
|
||
replyto_event_id=None,
|
||
url=url,
|
||
mimetype=mime,
|
||
)
|
||
|
||
|
||
def reaction_event(key: str, relates_to: str = "$orig"):
|
||
return SimpleNamespace(
|
||
sender="@a:m.org",
|
||
event_id="$r1",
|
||
key=key,
|
||
content={"m.relates_to": {"key": key, "event_id": relates_to}},
|
||
)
|
||
|
||
|
||
async def test_plain_text_to_incoming_message():
|
||
result = from_room_event(text_event("Hello"), room_id="!r:m.org", chat_id="C1")
|
||
assert isinstance(result, IncomingMessage)
|
||
assert result.text == "Hello"
|
||
assert result.platform == "matrix"
|
||
assert result.chat_id == "C1"
|
||
assert result.attachments == []
|
||
|
||
|
||
async def test_bang_command_to_incoming_command():
|
||
result = from_room_event(text_event("!new Analysis"), room_id="!r:m.org", chat_id="C1")
|
||
assert isinstance(result, IncomingCommand)
|
||
assert result.command == "new"
|
||
assert result.args == ["Analysis"]
|
||
|
||
|
||
async def test_skills_alias_to_settings_command():
|
||
result = from_command("!skills", sender="@a:m.org", chat_id="C1")
|
||
assert isinstance(result, IncomingCommand)
|
||
assert result.command == "settings_skills"
|
||
|
||
|
||
async def test_yes_to_callback():
|
||
result = from_room_event(text_event("!yes"), room_id="!r:m.org", chat_id="C1")
|
||
assert isinstance(result, IncomingCallback)
|
||
assert result.action == "confirm"
|
||
|
||
|
||
async def test_no_to_callback():
|
||
result = from_room_event(text_event("!no"), room_id="!r:m.org", chat_id="C1")
|
||
assert isinstance(result, IncomingCallback)
|
||
assert result.action == "cancel"
|
||
|
||
|
||
async def test_file_attachment():
|
||
result = from_room_event(file_event(), room_id="!r:m.org", chat_id="C1")
|
||
assert isinstance(result, IncomingMessage)
|
||
assert len(result.attachments) == 1
|
||
a = result.attachments[0]
|
||
assert a.type == "document"
|
||
assert a.url == "mxc://x/y"
|
||
assert a.filename == "doc.pdf"
|
||
assert a.mime_type == "application/pdf"
|
||
|
||
|
||
async def test_image_attachment():
|
||
result = from_room_event(image_event(), room_id="!r:m.org", chat_id="C1")
|
||
assert result.attachments[0].type == "image"
|
||
assert result.attachments[0].mime_type == "image/jpeg"
|
||
|
||
|
||
async def test_reaction_confirm():
|
||
result = from_reaction(reaction_event("👍"), sender="@a:m.org", chat_id="C1")
|
||
assert isinstance(result, IncomingCallback)
|
||
assert result.action == "confirm"
|
||
|
||
|
||
async def test_reaction_toggle_skill():
|
||
result = from_reaction(reaction_event("2️⃣"), sender="@a:m.org", chat_id="C1")
|
||
assert isinstance(result, IncomingCallback)
|
||
assert result.action == "toggle_skill"
|
||
assert result.payload["skill_index"] == 2
|