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

create tests for config

Browse source
This commit is contained in:
HgO 2022-07-09 12:22:05 +02:00
parent 687afe32ef
commit bc3f6ac452
6 changed files with 471 additions and 96 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

86
matrix_alertbot/config.py
View file

@ -7,7 +7,11 @@ from typing import Any, List, Optional
import pytimeparse
import yaml
from matrix_alertbot.errors import ConfigError
from matrix_alertbot.errors import (
InvalidConfigError,
ParseConfigError,
RequiredConfigKeyError,
)
logger = logging.getLogger()
logging.getLogger("peewee").setLevel(
@ -21,7 +25,7 @@ class Config:
def __init__(self, filepath: str):
self.filepath = filepath
if not os.path.isfile(filepath):
raise ConfigError(f"Config file '{filepath}' does not exist")
raise ParseConfigError(f"Config file '{filepath}' does not exist")
# Load in the config file at the given filepath
with open(filepath) as file_stream:
@ -44,7 +48,7 @@ class Config:
["logging", "file_logging", "enabled"], default=False
)
file_logging_filepath = self._get_cfg(
["logging", "file_logging", "filepath"], default="bot.log"
["logging", "file_logging", "filepath"], default="matrix-alertbot.log"
)
if file_logging_enabled:
file_handler = logging.FileHandler(file_logging_filepath)
@ -60,67 +64,77 @@ class Config:
logger.addHandler(console_handler)
# Storage setup
self.store_path = self._get_cfg(["storage", "store_path"], required=True)
self.store_dir: str = self._get_cfg(["storage", "path"], required=True)
# Create the store folder if it doesn't exist
if not os.path.isdir(self.store_path):
if not os.path.exists(self.store_path):
os.mkdir(self.store_path)
if not os.path.isdir(self.store_dir):
if not os.path.exists(self.store_dir):
os.mkdir(self.store_dir)
else:
raise ConfigError(
f"storage.store_path '{self.store_path}' is not a directory"
raise InvalidConfigError(
f"storage.path '{self.store_dir}' is not a directory"
)
# Cache setup
self.cache_dir = self._get_cfg(["cache", "directory"], required=True)
expire_time = self._get_cfg(["cache", "expire_time"], default="1w")
self.cache_dir: str = self._get_cfg(["cache", "path"], required=True)
expire_time: str = self._get_cfg(["cache", "expire_time"], default="1w")
self.cache_expire_time = pytimeparse.parse(expire_time)
# Alertmanager client setup
self.alertmanager_url = self._get_cfg(["alertmanager", "url"], required=True)
self.alertmanager_url: str = self._get_cfg(
["alertmanager", "url"], required=True
)
# Matrix bot account setup
self.user_id = self._get_cfg(["matrix", "user_id"], required=True)
if not re.match("@.*:.*", self.user_id):
raise ConfigError("matrix.user_id must be in the form @name:domain")
self.user_id: str = self._get_cfg(["matrix", "user_id"], required=True)
if not re.match("@.+:.+", self.user_id):
raise InvalidConfigError("matrix.user_id must be in the form @name:domain")
self.user_password = self._get_cfg(["matrix", "user_password"], required=False)
self.user_token = self._get_cfg(["matrix", "user_token"], required=False)
if not self.user_token and not self.user_password:
raise ConfigError("Must supply either user token or password")
self.device_id = self._get_cfg(["matrix", "device_id"], required=True)
self.device_name = self._get_cfg(
["matrix", "device_name"], default="nio-template"
self.user_password: str = self._get_cfg(
["matrix", "user_password"], required=False
)
self.homeserver_url = self._get_cfg(["matrix", "url"], required=True)
self.room_id = self._get_cfg(["matrix", "room"], required=True)
self.user_token: str = self._get_cfg(["matrix", "user_token"], required=False)
if not self.user_token and not self.user_password:
raise RequiredConfigKeyError("Must supply either user token or password")
self.address = self._get_cfg(["webhook", "address"], required=False)
self.port = self._get_cfg(["webhook", "port"], required=False)
self.socket = self._get_cfg(["webhook", "socket"], required=False)
self.device_id: str = self._get_cfg(["matrix", "device_id"], required=True)
self.device_name: str = self._get_cfg(
["matrix", "device_name"], default="matrix-alertbot"
)
self.homeserver_url: str = self._get_cfg(["matrix", "url"], required=True)
self.room_id: str = self._get_cfg(["matrix", "room"], required=True)
self.address: str = self._get_cfg(["webhook", "address"], required=False)
self.port: int = self._get_cfg(["webhook", "port"], required=False)
self.socket: str = self._get_cfg(["webhook", "socket"], required=False)
if (
not (self.address or self.port or self.socket)
or (self.socket and self.address and self.port)
or (self.address and not self.port)
or (not self.address and self.port)
):
raise ConfigError(
raise RequiredConfigKeyError(
"Must supply either webhook.socket or both webhook.address and webhook.port"
)
self.command_prefix = self._get_cfg(["command_prefix"], default="!c") + " "
elif self.socket and self.address and self.port:
raise InvalidConfigError(
"Supplied both webhook.socket and both webhook.address"
)
self.command_prefix: str = (
self._get_cfg(["command_prefix"], default="!alert") + " "
)
def _get_cfg(
self,
path: List[str],
default: Optional[Any] = None,
required: Optional[bool] = True,
required: bool = True,
) -> Any:
"""Get a config option from a path and option name, specifying whether it is
required.
Raises:
ConfigError: If required is True and the object is not found (and there is
RequiredConfigKeyError: If required is True and the object is not found (and there is
no default value provided), a ConfigError will be raised.
"""
# Sift through the the config until we reach our option
@ -131,8 +145,10 @@ class Config:
# If at any point we don't get our expected option...
if config is None:
# Raise an error if it was required
if required and not default:
raise ConfigError(f"Config option {'.'.join(path)} is required")
if required and default is None:
raise RequiredConfigKeyError(
f"Config option {'.'.join(path)} is required"
)
# or return the default value
return default

18
matrix_alertbot/errors.py
View file

@ -7,6 +7,24 @@ class ConfigError(Exception):
pass
class ParseConfigError(ConfigError):
"""An error encountered when config file cannot be parsed."""
pass
class InvalidConfigError(ParseConfigError):
"""An error encountered when a config key is not valid."""
pass
class RequiredConfigKeyError(ConfigError):
"""An error encountered when a required config key is missing."""
pass
class AlertmanagerError(Exception):
"""An error encountered with Alertmanager."""

2
matrix_alertbot/main.py
View file

@ -40,7 +40,7 @@ def create_matrix_client(config: Config) -> AsyncClient:
config.homeserver_url,
config.user_id,
device_id=config.device_id,
store_path=config.store_path,
store_path=config.store_dir,
config=client_config,
)

54
tests/resources/config/config.full.yml Normal file
View file

@ -0,0 +1,54 @@
# Welcome to the sample config file
# Below you will find various config sections and options
# Default values are shown
# The string to prefix messages with to talk to the bot in group chats
command_prefix: "!alert"
# Options for connecting to the bot's Matrix account
matrix:
# The Matrix User ID of the bot account
user_id: "@fakes_user:matrix.example.com"
# Matrix account password (optional if access token used)
user_password: "password"
# Matrix account access token (optional if password used)
#user_token: ""
# The URL of the homeserver to connect to
url: https://matrix.example.com
# The device ID that is **non pre-existing** device
# If this device ID already exists, messages will be dropped silently in encrypted rooms
device_id: ABCDEFGHIJ
# What to name the logged in device
device_name: fake_device_name
room: "!abcdefgh:matrix.example.com"
webhook:
socket: matrix-alertbot.socket
alertmanager:
url: http://localhost:9093
cache:
# The path to a directory for caching alerts and silences
path: "data/cache"
storage:
# The path to a directory for internal bot storage
# containing encryption keys, sync tokens, etc.
path: "data/store"
# Logging setup
logging:
# Logging level
# Allowed levels are 'INFO', 'WARNING', 'ERROR', 'DEBUG' where DEBUG is most verbose
level: DEBUG
# Configure logging to a file
file_logging:
# Whether logging to a file is enabled
enabled: true
# The path to the file to log to. May be relative or absolute
filepath: fake.log
# Configure logging to the console output
console_logging:
# Whether logging to the console is enabled
enabled: false

34
tests/resources/config/config.minimal.yml Normal file
View file

@ -0,0 +1,34 @@
# Welcome to the sample config file
# Below you will find various config sections and options
# Default values are shown
# Options for connecting to the bot's Matrix account
matrix:
# The Matrix User ID of the bot account
user_id: "@fakes_user:matrix.example.com"
# Matrix account password (optional if access token used)
user_password: "password"
# Matrix account access token (optional if password used)
#user_token: ""
# The URL of the homeserver to connect to
url: https://matrix.example.com
# The device ID that is **non pre-existing** device
# If this device ID already exists, messages will be dropped silently in encrypted rooms
device_id: ABCDEFGHIJ
room: "!abcdefgh:matrix.example.com"
webhook:
address: 0.0.0.0
port: 8080
alertmanager:
url: http://localhost:9093
cache:
# The path to a directory for caching alerts and silences
path: "data/cache"
storage:
# The path to a directory for internal bot storage
# containing encryption keys, sync tokens, etc.
path: "data/store"

373
tests/test_config.py
View file

@ -1,80 +1,333 @@
import os
import unittest
from unittest.mock import Mock
from datetime import timedelta
from unittest.mock import Mock, patch
import yaml
from matrix_alertbot.config import Config
from matrix_alertbot.errors import ConfigError
from matrix_alertbot.errors import (
ConfigError,
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):
def test_get_cfg(self) -> None:
"""Test that Config._get_cfg works correctly"""
@patch("os.path.isdir")
@patch("os.path.exists")
@patch("os.mkdir")
def test_read_minimal_config(
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
# Here's our test dictionary. Pretend that this was parsed from a YAML config file.
test_config_dict = {"a_key": 5, "some_key": {"some_other_key": "some_value"}}
config_path = os.path.join(CONFIG_RESOURCES_DIR, "config.minimal.yml")
config = Config(config_path)
# We create a fake config using Mock. _get_cfg will attempt to pull from self.config_dict,
# so we use a Mock to quickly create a dummy class, and set the 'config_dict' attribute to
# our test dictionary.
fake_config = Mock()
fake_config.config_dict = test_config_dict
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")
# Now let's make some calls to Config._get_cfg. We provide 'fake_cfg' as the first argument
# as a substitute for 'self'. _get_cfg will then be pulling values from fake_cfg.config_dict.
self.assertEqual("@fakes_user:matrix.example.com", config.user_id)
self.assertEqual("password", config.user_password)
self.assertIsNone(config.user_token)
self.assertEqual("ABCDEFGHIJ", config.device_id)
self.assertEqual("matrix-alertbot", config.device_name)
self.assertEqual("https://matrix.example.com", config.homeserver_url)
self.assertEqual("!abcdefgh:matrix.example.com", config.room_id)
# Test that we can get the value of a top-level key
self.assertEqual(
Config._get_cfg(fake_config, ["a_key"]),
5,
)
self.assertEqual("0.0.0.0", config.address)
self.assertEqual(8080, config.port)
self.assertIsNone(config.socket)
# Test that we can get the value of a nested key
self.assertEqual(
Config._get_cfg(fake_config, ["some_key", "some_other_key"]),
"some_value",
)
self.assertEqual("http://localhost:9093", config.alertmanager_url)
# Test that the value provided by the default option is used when a key does not exist
self.assertEqual(
Config._get_cfg(
fake_config,
["a_made_up_key", "this_does_not_exist"],
default="The default",
),
"The default",
)
expected_expire_time = timedelta(days=7).total_seconds()
self.assertEqual(expected_expire_time, config.cache_expire_time)
self.assertEqual("data/cache", config.cache_dir)
# Test that the value provided by the default option is *not* used when a key *does* exist
self.assertEqual(
Config._get_cfg(fake_config, ["a_key"], default="The default"),
5,
)
self.assertEqual("data/store", config.store_dir)
# Test that keys that do not exist raise a ConfigError when the required argument is True
with self.assertRaises(ConfigError):
Config._get_cfg(
fake_config, ["a_made_up_key", "this_does_not_exist"], required=True
)
self.assertEqual("!alert ", config.command_prefix)
# Test that a ConfigError is not returned when a non-existent key is provided and required is False
self.assertIsNone(
Config._get_cfg(
fake_config, ["a_made_up_key", "this_does_not_exist"], required=False
)
)
@patch("os.path.isdir")
@patch("os.path.exists")
@patch("os.mkdir")
def test_read_full_config(
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
# Test that default is used for non-existent keys, even if required is True
# (Typically one shouldn't use a default with required=True anyways...)
self.assertEqual(
Config._get_cfg(
fake_config,
["a_made_up_key", "this_does_not_exist"],
default="something",
required=True,
),
"something",
)
config_path = os.path.join(CONFIG_RESOURCES_DIR, "config.full.yml")
config = Config(config_path)
# TODO: Test creating a test yaml file, passing the path to Config and _parse_config_values is called correctly
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")
self.assertEqual("@fakes_user:matrix.example.com", config.user_id)
self.assertEqual("password", config.user_password)
self.assertIsNone(config.user_token)
self.assertEqual("ABCDEFGHIJ", config.device_id)
self.assertEqual("fake_device_name", config.device_name)
self.assertEqual("https://matrix.example.com", config.homeserver_url)
self.assertEqual("!abcdefgh:matrix.example.com", config.room_id)
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("!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"]["user_id"]
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"]["user_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_device_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"]["device_id"]
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"]["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_room(
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"]["room"]
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"]["user_id"] = ""
with self.assertRaises(InvalidConfigError):
config._parse_config_values()
config.config_dict["matrix"]["user_id"] = "@fake_user"
with self.assertRaises(InvalidConfigError):
config._parse_config_values()
config.config_dict["matrix"]["user_id"] = "@fake_user:"
with self.assertRaises(InvalidConfigError):
config._parse_config_values()
config.config_dict["matrix"]["user_id"] = ":matrix.example.com"
with self.assertRaises(InvalidConfigError):
config._parse_config_values()
config.config_dict["matrix"]["user_id"] = "@:matrix.example.com"
with self.assertRaises(InvalidConfigError):
config._parse_config_values()
config.config_dict["matrix"]["user_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()
if __name__ == "__main__":