import os 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, 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 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) self.assertEqual("!alert ", config.command_prefix) @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.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) self.assertEqual("!alert ", config.command_prefix) 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()