matrix-alertbot/tests/test_config.py

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()