import os import re import sys import unittest from datetime import timedelta from unittest.mock import Mock, patch import yaml import matrix_alertbot.config from matrix_alertbot.config import DEFAULT_REACTIONS, BiDict, Config from matrix_alertbot.errors import ( InvalidConfigError, ParseConfigError, RequiredConfigKeyError, ) WORKING_DIR = os.path.dirname(__file__) CONFIG_RESOURCES_DIR = os.path.join(WORKING_DIR, "resources", "config") class DummyConfig(Config): def __init__(self, filepath: str): with open(filepath) as file_stream: self.config_dict = yaml.safe_load(file_stream.read()) def mock_path_isdir(path: str) -> bool: if path == "data/store": return False return True def mock_path_exists(path: str) -> bool: if path == "data/store": return False return True class BiDictTestCase(unittest.TestCase): def test_init_bidict(self) -> None: data = {"room1": "user1", "room2": "user1", "room3": "user2"} bidict = BiDict(data) self.assertDictEqual(data, bidict) self.assertDictEqual( {"user1": {"room1", "room2"}, "user2": {"room3"}}, bidict.inverse ) def test_del_item_bidict(self) -> None: data = {"room1": "user1", "room2": "user1", "room3": "user2"} bidict = BiDict(data) del bidict["room1"] self.assertDictEqual({"room2": "user1", "room3": "user2"}, bidict) self.assertDictEqual({"user1": {"room2"}, "user2": {"room3"}}, bidict.inverse) del bidict["room3"] self.assertDictEqual({"room2": "user1"}, bidict) self.assertDictEqual( { "user1": {"room2"}, }, bidict.inverse, ) del bidict["room2"] self.assertDictEqual({}, bidict) self.assertDictEqual({}, bidict.inverse) with self.assertRaises(KeyError): del bidict["room4"] def test_add_item_bidict(self) -> None: data = {"room1": "user1", "room2": "user1", "room3": "user2"} bidict = BiDict(data) bidict["room4"] = "user2" self.assertDictEqual( {"room1": "user1", "room2": "user1", "room3": "user2", "room4": "user2"}, bidict, ) self.assertDictEqual( {"user1": {"room1", "room2"}, "user2": {"room3", "room4"}}, bidict.inverse ) bidict["room4"] = "user1" self.assertDictEqual( {"room1": "user1", "room2": "user1", "room3": "user2", "room4": "user1"}, bidict, ) self.assertDictEqual( {"user1": {"room1", "room2", "room4"}, "user2": {"room3"}}, bidict.inverse ) bidict["room3"] = "user1" self.assertDictEqual( {"room1": "user1", "room2": "user1", "room3": "user1", "room4": "user1"}, bidict, ) self.assertDictEqual( {"user1": {"room1", "room2", "room3", "room4"}, "user2": set()}, bidict.inverse, ) class ConfigTestCase(unittest.TestCase): @patch("os.path.isdir") @patch("os.path.exists") @patch("os.mkdir") @patch.object(matrix_alertbot.config, "logger", autospec=True) @patch.object(matrix_alertbot.config, "logging", autospec=True) def test_read_minimal_config( self, fake_logging: Mock, fake_logger: Mock, fake_mkdir: Mock, fake_path_exists: Mock, fake_path_isdir: Mock, ) -> None: fake_path_isdir.return_value = False fake_path_exists.return_value = False config_path = os.path.join(CONFIG_RESOURCES_DIR, "config.minimal.yml") config = Config(config_path) fake_path_isdir.assert_called_once_with("data/store") fake_path_exists.assert_called_once_with("data/store") fake_mkdir.assert_called_once_with("data/store") fake_logger.setLevel.assert_called_once_with("DEBUG") fake_logger.addHandler.assert_called_once() fake_logging.StreamHandler.return_value.setLevel.assert_called_once_with("INFO") fake_logging.StreamHandler.assert_called_once_with(sys.stdout) self.assertEqual({"@fakes_user:matrix.example.com"}, config.user_ids) self.assertEqual(1, len(config.accounts)) self.assertEqual("password", config.accounts[0].password) self.assertIsNone(config.accounts[0].token) self.assertIsNone(config.accounts[0].device_id) self.assertEqual("matrix-alertbot", config.device_name) self.assertEqual( "https://matrix.example.com", config.accounts[0].homeserver_url ) self.assertEqual(["!abcdefgh:matrix.example.com"], config.allowed_rooms) self.assertEqual(DEFAULT_REACTIONS, config.allowed_reactions) self.assertEqual("0.0.0.0", config.address) self.assertEqual(8080, config.port) self.assertIsNone(config.socket) self.assertEqual("http://localhost:9093", config.alertmanager_url) expected_expire_time = timedelta(days=7).total_seconds() self.assertEqual(expected_expire_time, config.cache_expire_time) self.assertEqual("data/cache", config.cache_dir) self.assertEqual("data/store", config.store_dir) self.assertIsNone(config.template_dir) @patch("os.path.isdir") @patch("os.path.exists") @patch("os.mkdir") @patch.object(matrix_alertbot.config, "logger", autospec=True) @patch.object(matrix_alertbot.config, "logging", autospec=True) def test_read_full_config( self, fake_logging: Mock, fake_logger: Mock, fake_mkdir: Mock, fake_path_exists: Mock, fake_path_isdir: Mock, ) -> None: fake_path_isdir.return_value = False fake_path_exists.return_value = False config_path = os.path.join(CONFIG_RESOURCES_DIR, "config.full.yml") config = Config(config_path) fake_path_isdir.assert_called_once_with("data/store") fake_path_exists.assert_called_once_with("data/store") fake_mkdir.assert_called_once_with("data/store") fake_logger.setLevel.assert_called_once_with("DEBUG") fake_logger.addHandler.assert_called_once() fake_logging.FileHandler.return_value.setLevel.assert_called_once_with("INFO") fake_logging.FileHandler.assert_called_once_with("fake.log") self.assertEqual( {"@fakes_user:matrix.example.com", "@other_user:matrix.domain.tld"}, config.user_ids, ) self.assertEqual(2, len(config.accounts)) self.assertEqual("password", config.accounts[0].password) self.assertIsNone(config.accounts[0].token) self.assertEqual("fake_token.json", config.accounts[0].token_file) self.assertEqual("ABCDEFGHIJ", config.accounts[0].device_id) self.assertEqual( "https://matrix.example.com", config.accounts[0].homeserver_url ) self.assertIsNone(config.accounts[1].password) self.assertEqual("token", config.accounts[1].token) self.assertEqual("other_token.json", config.accounts[1].token_file) self.assertEqual("KLMNOPQRST", config.accounts[1].device_id) self.assertEqual("https://matrix.domain.tld", config.accounts[1].homeserver_url) self.assertEqual("fake_device_name", config.device_name) self.assertEqual(["!abcdefgh:matrix.example.com"], config.allowed_rooms) self.assertEqual({"🤫", "😶", "🤐", "🤗"}, config.allowed_reactions) self.assertEqual({"🤗"}, config.insult_reactions) self.assertDictEqual({"matrix": re.compile("dm")}, config.dm_filter_labels) self.assertEqual("uuid", config.dm_select_label) self.assertEqual( "Alerts for FakeUser", config.dm_room_title.format(user="FakeUser") ) self.assertDictEqual( { "a7b37c33-574c-45ac-bb07-a3b314c2da54": "@some_other_user1:example.com", "cfb32a1d-737a-4618-8ee9-09b254d98fee": "@some_other_user2:example.com", "27e73f9b-b40a-4d84-b5b5-225931f6c289": "@some_other_user2:example.com", }, config.dm_users, ) self.assertDictEqual( { "@some_other_user1:example.com": { "a7b37c33-574c-45ac-bb07-a3b314c2da54" }, "@some_other_user2:example.com": { "cfb32a1d-737a-4618-8ee9-09b254d98fee", "27e73f9b-b40a-4d84-b5b5-225931f6c289", }, }, config.dm_users.inverse, ) self.assertIsNone(config.address) self.assertIsNone(config.port) self.assertEqual("matrix-alertbot.socket", config.socket) self.assertEqual("http://localhost:9093", config.alertmanager_url) expected_expire_time = timedelta(days=7).total_seconds() self.assertEqual(expected_expire_time, config.cache_expire_time) self.assertEqual("data/cache", config.cache_dir) self.assertEqual("data/store", config.store_dir) self.assertEqual("data/templates", config.template_dir) def test_read_config_raise_config_error(self) -> None: with self.assertRaises(ParseConfigError): Config("") @patch("os.path.isdir") @patch("os.path.exists") @patch("os.mkdir") def test_parse_config_with_storage_path_error( self, fake_mkdir: Mock, fake_path_exists: Mock, fake_path_isdir: Mock ) -> None: fake_path_isdir.return_value = False fake_path_exists.return_value = True config_path = os.path.join(CONFIG_RESOURCES_DIR, "config.minimal.yml") with self.assertRaises(ParseConfigError): Config(config_path) fake_path_isdir.assert_called_once_with("data/store") fake_path_exists.assert_called_once_with("data/store") fake_mkdir.assert_not_called() @patch("os.path.isdir") @patch("os.path.exists") @patch("os.mkdir") def test_parse_config_with_missing_matrix_user_id( self, fake_mkdir: Mock, fake_path_exists: Mock, fake_path_isdir: Mock ) -> None: fake_path_isdir.return_value = False fake_path_exists.return_value = False config_path = os.path.join(CONFIG_RESOURCES_DIR, "config.minimal.yml") config = DummyConfig(config_path) del config.config_dict["matrix"]["accounts"] with self.assertRaises(RequiredConfigKeyError): config._parse_config_values() @patch("os.path.isdir") @patch("os.path.exists") @patch("os.mkdir") def test_parse_config_with_missing_matrix_user_password( self, fake_mkdir: Mock, fake_path_exists: Mock, fake_path_isdir: Mock ) -> None: fake_path_isdir.return_value = False fake_path_exists.return_value = False config_path = os.path.join(CONFIG_RESOURCES_DIR, "config.minimal.yml") config = DummyConfig(config_path) del config.config_dict["matrix"]["accounts"][0]["password"] with self.assertRaises(RequiredConfigKeyError): config._parse_config_values() @patch("os.path.isdir") @patch("os.path.exists") @patch("os.mkdir") def test_parse_config_with_missing_matrix_url( self, fake_mkdir: Mock, fake_path_exists: Mock, fake_path_isdir: Mock ) -> None: fake_path_isdir.return_value = False fake_path_exists.return_value = False config_path = os.path.join(CONFIG_RESOURCES_DIR, "config.minimal.yml") config = DummyConfig(config_path) del config.config_dict["matrix"]["accounts"][0]["url"] with self.assertRaises(RequiredConfigKeyError): config._parse_config_values() @patch("os.path.isdir") @patch("os.path.exists") @patch("os.mkdir") def test_parse_config_with_missing_matrix_allowed_rooms( self, fake_mkdir: Mock, fake_path_exists: Mock, fake_path_isdir: Mock ) -> None: fake_path_isdir.return_value = False fake_path_exists.return_value = False config_path = os.path.join(CONFIG_RESOURCES_DIR, "config.minimal.yml") config = DummyConfig(config_path) del config.config_dict["matrix"]["allowed_rooms"] with self.assertRaises(RequiredConfigKeyError): config._parse_config_values() @patch("os.path.isdir") @patch("os.path.exists") @patch("os.mkdir") def test_parse_config_with_missing_webhook_address( self, fake_mkdir: Mock, fake_path_exists: Mock, fake_path_isdir: Mock ) -> None: fake_path_isdir.return_value = False fake_path_exists.return_value = False config_path = os.path.join(CONFIG_RESOURCES_DIR, "config.minimal.yml") config = DummyConfig(config_path) del config.config_dict["webhook"]["address"] with self.assertRaises(RequiredConfigKeyError): config._parse_config_values() @patch("os.path.isdir") @patch("os.path.exists") @patch("os.mkdir") def test_parse_config_with_missing_alertmanager_url( self, fake_mkdir: Mock, fake_path_exists: Mock, fake_path_isdir: Mock ) -> None: fake_path_isdir.return_value = False fake_path_exists.return_value = False config_path = os.path.join(CONFIG_RESOURCES_DIR, "config.minimal.yml") config = DummyConfig(config_path) del config.config_dict["alertmanager"]["url"] with self.assertRaises(RequiredConfigKeyError): config._parse_config_values() @patch("os.path.isdir") @patch("os.path.exists") @patch("os.mkdir") def test_parse_config_with_missing_cache_path( self, fake_mkdir: Mock, fake_path_exists: Mock, fake_path_isdir: Mock ) -> None: fake_path_isdir.return_value = False fake_path_exists.return_value = False config_path = os.path.join(CONFIG_RESOURCES_DIR, "config.minimal.yml") config = DummyConfig(config_path) del config.config_dict["cache"]["path"] with self.assertRaises(RequiredConfigKeyError): config._parse_config_values() @patch("os.path.isdir") @patch("os.path.exists") @patch("os.mkdir") def test_parse_config_with_missing_storage_path( self, fake_mkdir: Mock, fake_path_exists: Mock, fake_path_isdir: Mock ) -> None: fake_path_isdir.return_value = False fake_path_exists.return_value = False config_path = os.path.join(CONFIG_RESOURCES_DIR, "config.minimal.yml") config = DummyConfig(config_path) del config.config_dict["storage"]["path"] with self.assertRaises(RequiredConfigKeyError): config._parse_config_values() @patch("os.path.isdir") @patch("os.path.exists") @patch("os.mkdir") def test_parse_config_with_invalid_matrix_user_id( self, fake_mkdir: Mock, fake_path_exists: Mock, fake_path_isdir: Mock ) -> None: fake_path_isdir.return_value = False fake_path_exists.return_value = False config_path = os.path.join(CONFIG_RESOURCES_DIR, "config.minimal.yml") config = DummyConfig(config_path) config.config_dict["matrix"]["accounts"][0]["id"] = "" with self.assertRaises(InvalidConfigError): config._parse_config_values() config.config_dict["matrix"]["accounts"][0]["id"] = "@fake_user" with self.assertRaises(InvalidConfigError): config._parse_config_values() config.config_dict["matrix"]["accounts"][0]["id"] = "@fake_user:" with self.assertRaises(InvalidConfigError): config._parse_config_values() config.config_dict["matrix"]["accounts"][0]["id"] = ":matrix.example.com" with self.assertRaises(InvalidConfigError): config._parse_config_values() config.config_dict["matrix"]["accounts"][0]["id"] = "@:matrix.example.com" with self.assertRaises(InvalidConfigError): config._parse_config_values() config.config_dict["matrix"]["accounts"][0]["id"] = "@:" with self.assertRaises(InvalidConfigError): config._parse_config_values() @patch("os.path.isdir") @patch("os.path.exists") @patch("os.mkdir") def test_parse_config_with_both_webhook_socket_and_address( self, fake_mkdir: Mock, fake_path_exists: Mock, fake_path_isdir: Mock ) -> None: fake_path_isdir.return_value = False fake_path_exists.return_value = False config_path = os.path.join(CONFIG_RESOURCES_DIR, "config.minimal.yml") config = DummyConfig(config_path) config.config_dict["webhook"]["socket"] = "matrix-alertbot.socket" with self.assertRaises(InvalidConfigError): config._parse_config_values() @patch("os.path.isdir") @patch("os.path.exists") @patch("os.mkdir") @patch.object(matrix_alertbot.config, "logger") def test_parse_config_with_both_logging_disabled( self, fake_logger: Mock, fake_mkdir: Mock, fake_path_exists: Mock, fake_path_isdir: Mock, ) -> None: fake_path_isdir.return_value = False fake_path_exists.return_value = False config_path = os.path.join(CONFIG_RESOURCES_DIR, "config.full.yml") config = DummyConfig(config_path) config.config_dict["logging"]["file_logging"]["enabled"] = False config.config_dict["logging"]["console_logging"]["enabled"] = False config._parse_config_values() fake_logger.addHandler.assert_not_called() fake_logger.setLevel.assert_called_once_with("DEBUG") @patch("os.path.isdir") @patch("os.path.exists") @patch("os.mkdir") @patch.object(matrix_alertbot.config, "logger", autospec=True) @patch.object(matrix_alertbot.config, "logging", autospec=True) def test_parse_config_with_level_logging_different( self, fake_logging: Mock, fake_logger: Mock, fake_mkdir: Mock, fake_path_exists: Mock, fake_path_isdir: Mock, ) -> None: fake_path_isdir.return_value = False fake_path_exists.return_value = False config_path = os.path.join(CONFIG_RESOURCES_DIR, "config.full.yml") config = DummyConfig(config_path) config.config_dict["logging"]["file_logging"]["enabled"] = True config.config_dict["logging"]["file_logging"]["level"] = "WARN" config.config_dict["logging"]["console_logging"]["enabled"] = True config.config_dict["logging"]["console_logging"]["level"] = "ERROR" config._parse_config_values() self.assertEqual(2, fake_logger.addHandler.call_count) fake_logger.setLevel.assert_called_once_with("DEBUG") fake_logging.FileHandler.return_value.setLevel.assert_called_once_with("WARN") fake_logging.StreamHandler.return_value.setLevel.assert_called_once_with( "ERROR" ) if __name__ == "__main__": unittest.main()