Repository: freifunkMUC/wgkex (language: python)

Issue:
Create index.html



Discussion / hints:

Fix patch:
diff --git a/wgkex/config/BUILD b/wgkex/config/BUILD
index be5873a..1239a74 100644
--- a/wgkex/config/BUILD
+++ b/wgkex/config/BUILD
@@ -7,8 +7,16 @@ py_library(
     srcs = ["config.py"],
     visibility = ["//visibility:public"],
     deps = [
-       requirement("voluptuous"),
        requirement("PyYAML"),
        "//wgkex/common:utils"
     ],
-)
\ No newline at end of file
+)
+
+py_test(
+    name = "config_test",
+    srcs = ["config_test.py"],
+    deps = [
+       ":config",
+       requirement("mock"),
+    ],
+)
diff --git a/wgkex/config/config.py b/wgkex/config/config.py
index f3ae15a..780e574 100644
--- a/wgkex/config/config.py
+++ b/wgkex/config/config.py
@@ -1,35 +1,115 @@
+"""Configuration handling class."""
 import os
 import sys
-
 import yaml
-from voluptuous import All, MultipleInvalid, Required, Schema
-
-MQTT_SCHEMA = Schema(
-    {
-        Required("broker_url"): str,
-        Required("broker_port", default=1883): int,
-        Required("username", default=""): str,
-        Required("password", default=""): str,
-        Required("keepalive", default=5): int,
-        Required("tls", default=False): bool,
-    }
-)
-
-CONFIG_SCHEMA = Schema(
-    {
-        Required("domains"): All([str], min=1),
-        Required("pubkeys_file", default=""): str,
-        Required("mqtt"): MQTT_SCHEMA,
-    }
-)
-
-
-def load_config():
-    config_file = os.environ.get("WGKEX_CONFIG_FILE", "/etc/wgkex.yaml")
-    with open(config_file, "r") as stream:
-        try:
-            config = CONFIG_SCHEMA(yaml.safe_load(stream))
-        except MultipleInvalid as ex:
-            print(f"Config file failed to validate: {ex}", file=sys.stderr)
-            sys.exit(1)
-    return config
+from functools import lru_cache
+from typing import Dict, Union, Any, List, Optional
+import dataclasses
+
+
+class Error(Exception):
+    """Base Exception handling class."""
+
+
+class ConfigFileNotFoundError(Error):
+    """File could not be found on disk."""
+
+
+WG_CONFIG_OS_ENV = "WGKEX_CONFIG_FILE"
+WG_CONFIG_DEFAULT_LOCATION = "/etc/wgkex.yaml"
+
+
+@dataclasses.dataclass
+class MQTT:
+    """A representation of MQTT key in Configuration file.
+
+    Attributes:
+        broker_url: The broker URL for MQTT to connect to.
+        username: The username to use for MQTT.
+        password: The password to use for MQTT.
+        tls: If TLS is used or not.
+        broker_port: The port for MQTT to connect on.
+        keepalive: The keepalive in seconds to use.
+    """
+
+    broker_url: str
+    username: str
+    password: str
+    tls: bool = False
+    broker_port: int = 1883
+    keepalive: int = 5
+
+
+@dataclasses.dataclass
+class Config:
+    """A representation of the configuration file.
+
+    Attributes:
+        domains: The list of domains to listen for.
+        pubkeys_file: The public keys file to use.
+        mqtt: The MQTT configuration.
+    """
+
+    domains: List[str]
+    pubkeys_file: str
+    mqtt: MQTT
+
+    @classmethod
+    def from_dict(cls, cfg: Dict[str, str]) -> "Config":
+        """Creates a Config object from a configuration file.
+        Arguments:
+            cfg: The configuration file as a dict.
+        Returns:
+            A Config object.
+        """
+        return cls(**cfg)
+
+
+@lru_cache(maxsize=10)
+def fetch_from_config(key: str) -> Optional[Union[Dict[str, Any], List[str]]]:
+    """Fetches a specific key from configuration.
+
+    Arguments:
+        key: The named key to fetch.
+    Returns:
+        The config value associated with the key
+    """
+    return load_config().get(key)
+
+
+def load_config() -> Dict[str, str]:
+    """Fetches and validates configuration file from disk.
+
+    Returns:
+        Linted configuration file.
+    """
+    cfg_contents = fetch_config_from_disk()
+    try:
+        config = yaml.safe_load(cfg_contents)
+    except yaml.YAMLError as e:
+        print(f"Failed to load YAML file: {e}")
+        sys.exit(1)
+    try:
+        _ = Config.from_dict(config)
+        return config
+    except (ValueError, TypeError) as e:
+        print(f"Failed to lint file: {e}", file=sys.stderr)
+        sys.exit(1)
+
+
+def fetch_config_from_disk() -> str:
+    """Fetches config file from disk and returns as string.
+
+    Raises:
+        ConfigFileNotFoundError: If we could not find the configuration file on disk.
+    Returns:
+        The file contents as string.
+    """
+    config_file = os.environ.get(WG_CONFIG_OS_ENV, WG_CONFIG_DEFAULT_LOCATION)
+    try:
+        with open(config_file, "r") as stream:
+            return stream.read()
+    except FileNotFoundError as e:
+        raise ConfigFileNotFoundError(
+            f"Could not locate configuration file in {config_file}"
+        ) from e


Test patch:
diff --git a/wgkex/config/config_test.py b/wgkex/config/config_test.py
new file mode 100644
index 0000000..044483e
--- /dev/null
+++ b/wgkex/config/config_test.py
@@ -0,0 +1,62 @@
+import unittest
+import mock
+import config
+import yaml
+
+_VALID_CFG = "domains:\n- a\n- b\nmqtt:\n  broker_port: 1883\n  broker_url: mqtt://broker\n  keepalive: 5\n  password: pass\n  tls: true\n  username: user\npubkeys_file: some/file\n"
+_INVALID_LINT = "derpmains:\n- a\n- b\nmqtt:\n  broker_port: 1883\n  broker_url: mqtt://broker\n  keepalive: 5\n  password: pass\n  tls: true\n  username: user\npubkeys_file: some/file\n"
+_INVALID_CFG = "asdasdasdasd"
+
+
+class TestConfig(unittest.TestCase):
+    def test_load_config_success(self):
+        """Test loads and lint config successfully."""
+        mock_open = mock.mock_open(read_data=_VALID_CFG)
+        with mock.patch("builtins.open", mock_open):
+            self.assertDictEqual(yaml.safe_load(_VALID_CFG), config.load_config())
+
+    @mock.patch.object(config.sys, "exit", autospec=True)
+    def test_load_config_fails_good_yaml_bad_format(self, exit_mock):
+        """Test loads yaml successfully and fails lint."""
+        mock_open = mock.mock_open(read_data=_INVALID_LINT)
+        with mock.patch("builtins.open", mock_open):
+            config.load_config()
+            exit_mock.assert_called_with(1)
+
+    @mock.patch.object(config.sys, "exit", autospec=True)
+    def test_load_config_fails_bad_yaml(self, exit_mock):
+        """Test loads bad YAML."""
+        mock_open = mock.mock_open(read_data=_INVALID_CFG)
+        with mock.patch("builtins.open", mock_open):
+            config.load_config()
+            exit_mock.assert_called_with(1)
+
+    def test_fetch_config_from_disk_success(self):
+        """Test fetch file from disk."""
+        mock_open = mock.mock_open(read_data=_VALID_CFG)
+        with mock.patch("builtins.open", mock_open):
+            self.assertEqual(config.fetch_config_from_disk(), _VALID_CFG)
+
+    def test_fetch_config_from_disk_fails_file_not_found(self):
+        """Test fails on file not found on disk."""
+        mock_open = mock.mock_open()
+        mock_open.side_effect = FileNotFoundError
+        with mock.patch("builtins.open", mock_open):
+            with self.assertRaises(config.ConfigFileNotFoundError):
+                config.fetch_config_from_disk()
+
+    def test_fetch_from_config_success(self):
+        """Test fetch key from configuration."""
+        mock_open = mock.mock_open(read_data=_VALID_CFG)
+        with mock.patch("builtins.open", mock_open):
+            self.assertListEqual(["a", "b"], config.fetch_from_config("domains"))
+
+    def test_fetch_from_config_no_key_in_config(self):
+        """Test fetch non existent key from configuration."""
+        mock_open = mock.mock_open(read_data=_VALID_CFG)
+        with mock.patch("builtins.open", mock_open):
+            self.assertIsNone(config.fetch_from_config("key_does_not_exist"))
+
+
+if __name__ == "__main__":
+    unittest.main()