add tests for key verification callbacks

matrix_alertbot/callback.py

@@ -303,7 +303,7 @@ class Callbacks:
             f"Verification has been cancelled by {event.sender} for reason: {event.reason}."
         )
 
-    async def key_verification_accept(self, event: KeyVerificationKey):
+    async def key_verification_confirm(self, event: KeyVerificationKey):
         sas = self.matrix_client.key_verifications[event.transaction_id]
         emoji_list = sas.get_emoji()
         emoji_str = " ".join(emoji for emoji, alt_text in emoji_list)
@@ -343,7 +343,7 @@ class Callbacks:
             todevice_msg = sas.get_mac()
         except LocalProtocolError as e:
             # e.g. it might have been cancelled by ourselves
-            logger.warn(f"Unable to conclude verification with {event.sender}: {e}.")
+            logger.warning(f"Unable to conclude verification with {event.sender}: {e}.")
             return
 
         event_response = await self.matrix_client.to_device(todevice_msg)

matrix_alertbot/main.py

@@ -168,7 +168,7 @@ def main() -> None:
         callbacks.key_verification_cancel, (KeyVerificationCancel,)
     )
     matrix_client.add_to_device_callback(
-        callbacks.key_verification_accept, (KeyVerificationKey,)
+        callbacks.key_verification_confirm, (KeyVerificationKey,)
     )
     matrix_client.add_to_device_callback(
         callbacks.key_verification_end, (KeyVerificationMac,)

tests/test_callback.py

@@ -3,6 +3,7 @@ from typing import Dict
 from unittest.mock import MagicMock, Mock, patch
 
 import nio
+import nio.crypto
 from diskcache import Cache
 
 import matrix_alertbot.callback
@@ -14,6 +15,10 @@ from matrix_alertbot.command import BaseCommand
 from tests.utils import make_awaitable
 
 
+def key_verification_get_mac_raise_protocol_error():
+    raise nio.LocalProtocolError
+
+
 class CallbacksTestCase(unittest.IsolatedAsyncioTestCase):
     def setUp(self) -> None:
         # Create a Callbacks object and give it some Mock'd objects to use
@@ -48,7 +53,7 @@ class CallbacksTestCase(unittest.IsolatedAsyncioTestCase):
         fake_invite_event.sender = "@some_other_fake_user:example.com"
 
         # Pretend that attempting to join a room is always successful
-        self.fake_matrix_client.join.return_value = make_awaitable(None)
+        self.fake_matrix_client.join.return_value = make_awaitable()
 
         # Pretend that we received an invite event
         await self.callbacks.invite(self.fake_room, fake_invite_event)
@@ -540,6 +545,260 @@ class CallbacksTestCase(unittest.IsolatedAsyncioTestCase):
         fake_command.assert_not_called()
         self.fake_cache.__getitem__.assert_not_called()
 
+    async def test_key_verification_start(self) -> None:
+        """Tests the callback for RoomMessageText with the command prefix"""
+        # Tests that the bot process messages in the room that contain a command
+        fake_transaction_id = "fake transaction id"
+
+        fake_key_verification_event = Mock(spec=nio.KeyVerificationStart)
+        fake_key_verification_event.from_device = "ABCDEFGH"
+        fake_key_verification_event.sender = "@some_other_fake_user:example.com"
+        fake_key_verification_event.short_authentication_string = ["emoji"]
+        fake_key_verification_event.transaction_id = fake_transaction_id
+
+        self.fake_matrix_client.accept_key_verification.return_value = make_awaitable()
+        self.fake_matrix_client.to_device.return_value = make_awaitable()
+
+        fake_sas = Mock()
+        fake_transactions_dict = {fake_transaction_id: fake_sas}
+        self.fake_matrix_client.key_verifications = fake_transactions_dict
+
+        # Pretend that we received a text message event
+        await self.callbacks.key_verification_start(fake_key_verification_event)
+
+        # Check that we attempted to execute the command
+        self.fake_matrix_client.accept_key_verification.assert_called_once_with(
+            fake_transaction_id
+        )
+        self.fake_matrix_client.to_device.assert_called_once_with(fake_sas.share_key())
+
+    async def test_key_verification_start_with_emoji_not_supported(self) -> None:
+        """Tests the callback for RoomMessageText with the command prefix"""
+        # Tests that the bot process messages in the room that contain a command
+        fake_transaction_id = "fake transaction id"
+
+        fake_key_verification_event = Mock(spec=nio.KeyVerificationStart)
+        fake_key_verification_event.from_device = "ABCDEFGH"
+        fake_key_verification_event.sender = "@some_other_fake_user:example.com"
+        fake_key_verification_event.short_authentication_string = []
+        fake_key_verification_event.transaction_id = fake_transaction_id
+
+        self.fake_matrix_client.accept_key_verification.return_value = make_awaitable()
+        self.fake_matrix_client.to_device.return_value = make_awaitable()
+
+        fake_sas = Mock()
+        fake_transactions_dict = {fake_transaction_id: fake_sas}
+        self.fake_matrix_client.key_verifications = fake_transactions_dict
+
+        # Pretend that we received a text message event
+        await self.callbacks.key_verification_start(fake_key_verification_event)
+
+        # Check that we attempted to execute the command
+        self.fake_matrix_client.accept_key_verification.assert_not_called()
+        self.fake_matrix_client.to_device.assert_not_called()
+
+    async def test_key_verification_start_with_accept_key_verification_error(
+        self,
+    ) -> None:
+        """Tests the callback for RoomMessageText with the command prefix"""
+        # Tests that the bot process messages in the room that contain a command
+        fake_transaction_id = "fake transaction id"
+
+        fake_key_verification_event = Mock(spec=nio.KeyVerificationStart)
+        fake_key_verification_event.from_device = "ABCDEFGH"
+        fake_key_verification_event.sender = "@some_other_fake_user:example.com"
+        fake_key_verification_event.short_authentication_string = ["emoji"]
+        fake_key_verification_event.transaction_id = fake_transaction_id
+
+        self.fake_matrix_client.accept_key_verification.return_value = make_awaitable(
+            Mock(spec=nio.ToDeviceError)
+        )
+        self.fake_matrix_client.to_device.return_value = make_awaitable()
+
+        fake_sas = Mock()
+        fake_transactions_dict = {fake_transaction_id: fake_sas}
+        self.fake_matrix_client.key_verifications = fake_transactions_dict
+
+        # Pretend that we received a text message event
+        await self.callbacks.key_verification_start(fake_key_verification_event)
+
+        # Check that we attempted to execute the command
+        self.fake_matrix_client.accept_key_verification.assert_called_once_with(
+            fake_transaction_id
+        )
+        self.fake_matrix_client.to_device.assert_not_called()
+
+    async def test_key_verification_start_with_to_device_error(
+        self,
+    ) -> None:
+        """Tests the callback for RoomMessageText with the command prefix"""
+        # Tests that the bot process messages in the room that contain a command
+        fake_transaction_id = "fake transaction id"
+
+        fake_key_verification_event = Mock(spec=nio.KeyVerificationStart)
+        fake_key_verification_event.from_device = "ABCDEFGH"
+        fake_key_verification_event.sender = "@some_other_fake_user:example.com"
+        fake_key_verification_event.short_authentication_string = ["emoji"]
+        fake_key_verification_event.transaction_id = fake_transaction_id
+
+        self.fake_matrix_client.accept_key_verification.return_value = make_awaitable()
+        self.fake_matrix_client.to_device.return_value = make_awaitable(
+            Mock(spec=nio.ToDeviceError)
+        )
+
+        fake_sas = Mock()
+        fake_transactions_dict = {fake_transaction_id: fake_sas}
+        self.fake_matrix_client.key_verifications = fake_transactions_dict
+
+        # Pretend that we received a text message event
+        await self.callbacks.key_verification_start(fake_key_verification_event)
+
+        # Check that we attempted to execute the command
+        self.fake_matrix_client.accept_key_verification.assert_called_once_with(
+            fake_transaction_id
+        )
+        self.fake_matrix_client.to_device.assert_called_once_with(fake_sas.share_key())
+
+    async def test_key_verification_cancel(self) -> None:
+        """Tests the callback for RoomMessageText with the command prefix"""
+        # Tests that the bot process messages in the room that contain a command
+        fake_key_verification_event = Mock(spec=nio.KeyVerificationCancel)
+        fake_key_verification_event.sender = "@some_other_fake_user:example.com"
+        fake_key_verification_event.reason = "fake reason"
+
+        # Pretend that we received a text message event
+        await self.callbacks.key_verification_cancel(fake_key_verification_event)
+
+        # Check that we attempted to execute the command
+
+    async def test_key_verification_confirm(self) -> None:
+        """Tests the callback for RoomMessageText with the command prefix"""
+        # Tests that the bot process messages in the room that contain a command
+        fake_transaction_id = "fake transaction id"
+
+        fake_key_verification_event = Mock(spec=nio.KeyVerificationStart)
+        fake_key_verification_event.sender = "@some_other_fake_user:example.com"
+        fake_key_verification_event.transaction_id = fake_transaction_id
+
+        self.fake_matrix_client.confirm_short_auth_string.return_value = (
+            make_awaitable()
+        )
+
+        fake_sas = Mock()
+        fake_sas.get_emoji.return_value = [
+            ("emoji1", "alt text1"),
+            ("emoji2", "alt text2"),
+        ]
+        fake_transactions_dict = {fake_transaction_id: fake_sas}
+        self.fake_matrix_client.key_verifications = fake_transactions_dict
+
+        # Pretend that we received a text message event
+        await self.callbacks.key_verification_confirm(fake_key_verification_event)
+
+        # Check that we attempted to execute the command
+        self.fake_matrix_client.confirm_short_auth_string.assert_called_once_with(
+            fake_transaction_id
+        )
+
+    async def test_key_verification_confirm_with_error(self) -> None:
+        """Tests the callback for RoomMessageText with the command prefix"""
+        # Tests that the bot process messages in the room that contain a command
+        fake_transaction_id = "fake transaction id"
+
+        fake_key_verification_event = Mock(spec=nio.KeyVerificationStart)
+        fake_key_verification_event.sender = "@some_other_fake_user:example.com"
+        fake_key_verification_event.transaction_id = fake_transaction_id
+
+        self.fake_matrix_client.confirm_short_auth_string.return_value = make_awaitable(
+            Mock(spec=nio.ToDeviceError)
+        )
+
+        fake_sas = Mock()
+        fake_sas.get_emoji.return_value = [
+            ("emoji1", "alt text1"),
+            ("emoji2", "alt text2"),
+        ]
+        fake_transactions_dict = {fake_transaction_id: fake_sas}
+        self.fake_matrix_client.key_verifications = fake_transactions_dict
+
+        # Pretend that we received a text message event
+        await self.callbacks.key_verification_confirm(fake_key_verification_event)
+
+        # Check that we attempted to execute the command
+        self.fake_matrix_client.confirm_short_auth_string.assert_called_once_with(
+            fake_transaction_id
+        )
+
+    async def test_key_verification_end(self) -> None:
+        """Tests the callback for RoomMessageText with the command prefix"""
+        # Tests that the bot process messages in the room that contain a command
+        fake_transaction_id = "fake transaction id"
+
+        fake_key_verification_event = Mock(spec=nio.KeyVerificationStart)
+        fake_key_verification_event.sender = "@some_other_fake_user:example.com"
+        fake_key_verification_event.transaction_id = fake_transaction_id
+
+        self.fake_matrix_client.to_device.return_value = make_awaitable()
+
+        fake_sas = Mock()
+        fake_sas.verified_devices = ["HGFEDCBA"]
+        fake_transactions_dict = {fake_transaction_id: fake_sas}
+        self.fake_matrix_client.key_verifications = fake_transactions_dict
+
+        # Pretend that we received a text message event
+        await self.callbacks.key_verification_end(fake_key_verification_event)
+
+        # Check that we attempted to execute the command
+        fake_sas.get_mac.assert_called_once_with()
+        self.fake_matrix_client.to_device.assert_called_once_with(fake_sas.get_mac())
+
+    async def test_key_verification_end_with_mac_error(self) -> None:
+        """Tests the callback for RoomMessageText with the command prefix"""
+        # Tests that the bot process messages in the room that contain a command
+        fake_transaction_id = "fake transaction id"
+
+        fake_key_verification_event = Mock(spec=nio.KeyVerificationStart)
+        fake_key_verification_event.sender = "@some_other_fake_user:example.com"
+        fake_key_verification_event.transaction_id = fake_transaction_id
+
+        self.fake_matrix_client.to_device.return_value = make_awaitable()
+
+        fake_sas = Mock()
+        fake_sas.get_mac.side_effect = key_verification_get_mac_raise_protocol_error
+        fake_transactions_dict = {fake_transaction_id: fake_sas}
+        self.fake_matrix_client.key_verifications = fake_transactions_dict
+
+        # Pretend that we received a text message event
+        await self.callbacks.key_verification_end(fake_key_verification_event)
+
+        # Check that we attempted to execute the command
+        fake_sas.get_mac.assert_called_once_with()
+        self.fake_matrix_client.to_device.assert_not_called()
+
+    async def test_key_verification_end_with_to_device_error(self) -> None:
+        """Tests the callback for RoomMessageText with the command prefix"""
+        # Tests that the bot process messages in the room that contain a command
+        fake_transaction_id = "fake transaction id"
+
+        fake_key_verification_event = Mock(spec=nio.KeyVerificationStart)
+        fake_key_verification_event.sender = "@some_other_fake_user:example.com"
+        fake_key_verification_event.transaction_id = fake_transaction_id
+
+        self.fake_matrix_client.to_device.return_value = make_awaitable(
+            Mock(spec=nio.ToDeviceError)
+        )
+
+        fake_sas = Mock()
+        fake_transactions_dict = {fake_transaction_id: fake_sas}
+        self.fake_matrix_client.key_verifications = fake_transactions_dict
+
+        # Pretend that we received a text message event
+        await self.callbacks.key_verification_end(fake_key_verification_event)
+
+        # Check that we attempted to execute the command
+        fake_sas.get_mac.assert_called_once_with()
+        self.fake_matrix_client.to_device.assert_called_once_with(fake_sas.get_mac())
+
     @patch.object(matrix_alertbot.callback.CommandFactory, "create", autospec=True)
     async def test_unknown(self, fake_command_create: Mock) -> None:
         """Tests the callback for RoomMessageText with the command prefix"""

tests/test_command.py

@@ -85,7 +85,7 @@ class CommandTestCase(unittest.IsolatedAsyncioTestCase):
         self.fake_matrix_client = Mock(spec=nio.AsyncClient)
         self.fake_matrix_client.user = "@fake_user:example.com"
         # Pretend that attempting to send a message is always successful
-        self.fake_matrix_client.room_send.return_value = make_awaitable(None)
+        self.fake_matrix_client.room_send.return_value = make_awaitable()
 
         self.fake_cache = MagicMock(spec=Cache)
         self.fake_cache.__getitem__.side_effect = cache_get_item

tests/utils.py

@@ -11,7 +11,7 @@ def run_coroutine(result: Awaitable[Any]) -> Any:
     return result
 
 
-def make_awaitable(result: Any) -> Awaitable[Any]:
+def make_awaitable(result: Any = None) -> Awaitable[Any]:
     """
     Makes an awaitable, suitable for mocking an `async` function.
     This uses Futures as they can be awaited multiple times so can be returned