tobast/matrix-alertbot
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

add tests for chat functions

Browse source
This commit is contained in:
HgO 2022-08-08 12:38:09 +02:00
parent 6fd687cc90
commit 264687a62d
6 changed files with 176 additions and 18 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

11
matrix_alertbot/callback.py
View file

@ -5,11 +5,13 @@ from nio import (
AsyncClient,
InviteMemberEvent,
JoinError,
LocalProtocolError,
MatrixRoom,
MegolmEvent,
RedactionEvent,
RoomGetEventError,
RoomMessageText,
SendRetryError,
UnknownEvent,
)
@ -104,7 +106,10 @@ class Callbacks:
logging.error(f"Cannot process command '{cmd}': {e}")
return
try:
await command.process()
except (SendRetryError, LocalProtocolError) as e:
logger.exception(f"Unable to send message to {room.room_id}", exc_info=e)
async def invite(self, room: MatrixRoom, event: InviteMemberEvent) -> None:
"""Callback for when an invite is received. Join the room specified in the invite.
@ -204,7 +209,10 @@ class Callbacks:
alert_event_id,
)
try:
await command.process()
except (SendRetryError, LocalProtocolError) as e:
logger.exception(f"Unable to send message to {room.room_id}", exc_info=e)
async def redaction(self, room: MatrixRoom, event: RedactionEvent) -> None:
# Ignore events from unauthorized room
@ -227,7 +235,10 @@ class Callbacks:
event.event_id,
event.redacts,
)
try:
await command.process()
except (SendRetryError, LocalProtocolError) as e:
logger.exception(f"Unable to send message to {room.room_id}", exc_info=e)
async def decryption_failure(self, room: MatrixRoom, event: MegolmEvent) -> None:
"""Callback for when an event fails to decrypt. Inform the user.

9
matrix_alertbot/chat_functions.py
View file

@ -1,7 +1,7 @@
import logging
from typing import Dict, Optional, TypedDict, Union
from nio import AsyncClient, ErrorResponse, Response, RoomSendResponse, SendRetryError
from nio import AsyncClient, ErrorResponse, Response, RoomSendResponse
from typing_extensions import NotRequired
logger = logging.getLogger(__name__)
@ -20,7 +20,7 @@ ContentEventDict = TypedDict(
async def send_text_to_room(
client: AsyncClient,
matrix_client: AsyncClient,
room_id: str,
plaintext: str,
html: str = None,
@ -62,15 +62,12 @@ async def send_text_to_room(
if reply_to_event_id:
content["m.relates_to"] = {"m.in_reply_to": {"event_id": reply_to_event_id}}
try:
return await client.room_send(
return await matrix_client.room_send(
room_id,
"m.room.message",
content,
ignore_unverified_devices=True,
)
except SendRetryError:
logger.exception(f"Unable to send message response to {room_id}")
def make_pill(user_id: str, displayname: str = None) -> str:

2
matrix_alertbot/command.py
View file

@ -135,7 +135,7 @@ class AckAlertCommand(BaseAlertCommand):
alert_fingerprint,
self.room.user_name(self.sender),
duration_seconds,
force=True
force=True,
)
except AlertNotFoundError as e:
logger.warning(f"Unable to create silence: {e}")

4
matrix_alertbot/webhook.py
View file

@ -7,7 +7,7 @@ from aiohttp import ClientError, web, web_request
from aiohttp_prometheus_exporter.handler import metrics
from aiohttp_prometheus_exporter.middleware import prometheus_middleware_factory
from diskcache import Cache
from nio import AsyncClient, LocalProtocolError
from nio import AsyncClient, LocalProtocolError, SendRetryError
from matrix_alertbot.alert import Alert, AlertRenderer
from matrix_alertbot.alertmanager import AlertmanagerClient
@ -80,7 +80,7 @@ async def create_alerts(request: web_request.Request) -> web.Response:
status=500,
body=f"An error occured with Alertmanager when handling alert with fingerprint {alert.fingerprint}.",
)
except (LocalProtocolError, ClientError) as e:
except (SendRetryError, LocalProtocolError, ClientError) as e:
logger.error(
f"Unable to send alert {alert.fingerprint} to Matrix room: {e}"
)

150
tests/test_chat_functions.py Normal file
View file

@ -0,0 +1,150 @@
import unittest
from typing import Any, Dict, Optional
from unittest.mock import Mock
import nio
from matrix_alertbot.chat_functions import send_text_to_room, strip_fallback
from tests.utils import make_awaitable
async def send_room_raise_send_retry_error(
room_id: str,
message_type: str,
content: Dict[Any, Any],
tx_id: Optional[str] = None,
ignore_unverified_devices: bool = False,
) -> nio.RoomSendResponse:
raise nio.SendRetryError
class ChatFunctionsTestCase(unittest.IsolatedAsyncioTestCase):
def setUp(self) -> None:
pass
def test_strip_fallback(self) -> None:
fake_body = "> some message\n\nsome reply"
message = strip_fallback(fake_body)
self.assertEqual("some reply", message)
fake_body = "some message"
message = strip_fallback(fake_body)
self.assertEqual(fake_body, message)
async def test_send_text_to_room_as_notice(self) -> None:
fake_response = Mock(spec=nio.RoomSendResponse)
fake_matrix_client = Mock(spec=nio.AsyncClient)
fake_matrix_client.room_send = Mock(return_value=make_awaitable(fake_response))
fake_room_id = "!abcdefgh:example.com"
fake_plaintext_body = "some plaintext message"
fake_html_body = "some html message"
response = await send_text_to_room(
fake_matrix_client, fake_room_id, fake_plaintext_body, fake_html_body
)
fake_matrix_client.room_send.assert_called_once_with(
fake_room_id,
"m.room.message",
{
"msgtype": "m.notice",
"format": "org.matrix.custom.html",
"body": fake_plaintext_body,
"formatted_body": fake_html_body,
},
ignore_unverified_devices=True,
)
self.assertEqual(fake_response, response)
async def test_send_text_to_room_as_message(self) -> None:
fake_response = Mock(spec=nio.RoomSendResponse)
fake_matrix_client = Mock(spec=nio.AsyncClient)
fake_matrix_client.room_send = Mock(return_value=make_awaitable(fake_response))
fake_room_id = "!abcdefgh:example.com"
fake_plaintext_body = "some plaintext message"
fake_html_body = "some html message"
response = await send_text_to_room(
fake_matrix_client,
fake_room_id,
fake_plaintext_body,
fake_html_body,
notice=False,
)
fake_matrix_client.room_send.assert_called_once_with(
fake_room_id,
"m.room.message",
{
"msgtype": "m.text",
"format": "org.matrix.custom.html",
"body": fake_plaintext_body,
"formatted_body": fake_html_body,
},
ignore_unverified_devices=True,
)
self.assertEqual(fake_response, response)
async def test_send_text_to_room_in_reply_to_event(self) -> None:
fake_response = Mock(spec=nio.RoomSendResponse)
fake_matrix_client = Mock(spec=nio.AsyncClient)
fake_matrix_client.room_send = Mock(return_value=make_awaitable(fake_response))
fake_room_id = "!abcdefgh:example.com"
fake_plaintext_body = "some plaintext message"
fake_html_body = "some html message"
fake_event_id = "some event id"
response = await send_text_to_room(
fake_matrix_client,
fake_room_id,
fake_plaintext_body,
fake_html_body,
reply_to_event_id=fake_event_id,
)
fake_matrix_client.room_send.assert_called_once_with(
fake_room_id,
"m.room.message",
{
"msgtype": "m.notice",
"format": "org.matrix.custom.html",
"body": fake_plaintext_body,
"formatted_body": fake_html_body,
"m.relates_to": {"m.in_reply_to": {"event_id": fake_event_id}},
},
ignore_unverified_devices=True,
)
self.assertEqual(fake_response, response)
async def test_send_text_to_room_raise_send_retry_error(self) -> None:
fake_matrix_client = Mock(spec=nio.AsyncClient)
fake_matrix_client.room_send = Mock(
side_effect=send_room_raise_send_retry_error
)
fake_room_id = "!abcdefgh:example.com"
fake_plaintext_body = "some plaintext message"
fake_html_body = "some html message"
with self.assertRaises(nio.SendRetryError):
await send_text_to_room(
fake_matrix_client,
fake_room_id,
fake_plaintext_body,
fake_html_body,
)
fake_matrix_client.room_send.assert_called_once_with(
fake_room_id,
"m.room.message",
{
"msgtype": "m.notice",
"format": "org.matrix.custom.html",
"body": fake_plaintext_body,
"formatted_body": fake_html_body,
},
ignore_unverified_devices=True,
)
if __name__ == "__main__":
unittest.main()

2
tests/test_webhook.py
View file

@ -22,7 +22,7 @@ from matrix_alertbot.webhook import Webhook
def send_text_to_room_raise_error(
client: nio.AsyncClient, room_id: str, plaintext: str, html: str, notice: bool
) -> RoomSendResponse:
raise LocalProtocolError()
raise LocalProtocolError
def update_silence_raise_silence_not_found(fingerprint: str) -> str: