Initial MVP commits #1
68
src/hasskiosk/config.py
Normal file
68
src/hasskiosk/config.py
Normal file
@ -0,0 +1,68 @@
|
|||||||
|
"""Configuration management from environment."""
|
||||||
|
|
||||||
|
import logging.config
|
||||||
|
from logging import getLogger
|
||||||
|
from typing import Any, Dict
|
||||||
|
|
||||||
|
from environs import Env
|
||||||
|
|
||||||
|
from ._version import version
|
||||||
|
|
||||||
|
|
||||||
|
def read_config() -> Dict[str, Any]:
|
||||||
|
"""Read the configuration from the environment."""
|
||||||
|
env = Env()
|
||||||
|
env.read_env()
|
||||||
|
with env.prefixed("HASSKIOSK_"):
|
||||||
|
config = {
|
||||||
|
"sysname": env("SYSTEM_NAME", "hasskiosk"),
|
||||||
|
"version": version,
|
||||||
|
"logging": {
|
||||||
|
"version": 1,
|
||||||
|
"disable_existing_loggers": False,
|
||||||
|
"formatters": {
|
||||||
|
"simple": {
|
||||||
|
"format": "%(asctime)s %(levelname)s (%(name)s) %(message)s",
|
||||||
|
"converter": "time.gmtime",
|
||||||
|
},
|
||||||
|
},
|
||||||
|
"handlers": {
|
||||||
|
"stdout": {
|
||||||
|
"class": "logging.StreamHandler",
|
||||||
|
"level": env("LOG_LEVEL", "INFO"),
|
||||||
|
"formatter": "simple",
|
||||||
|
"stream": "ext://sys.stdout",
|
||||||
|
},
|
||||||
|
},
|
||||||
|
"loggers": {
|
||||||
|
"hasskiosk": {
|
||||||
|
"level": "DEBUG",
|
||||||
|
"propagate": "yes",
|
||||||
|
},
|
||||||
|
},
|
||||||
|
"root": {
|
||||||
|
"level": "WARN",
|
||||||
|
"handlers": ["stdout"],
|
||||||
|
},
|
||||||
|
},
|
||||||
|
}
|
||||||
|
with env.prefixed("TOPIC_"):
|
||||||
|
config["topics"] = {
|
||||||
|
"presence": env("PRESENCE"),
|
||||||
|
}
|
||||||
|
with env.prefixed("MQTT_"):
|
||||||
|
config["mqtt"] = {
|
||||||
|
"host": env("HOST"),
|
||||||
|
"port": env.int("PORT", 1883),
|
||||||
|
"username": env("USERNAME"),
|
||||||
|
"password": env("PASSWORD"),
|
||||||
|
"keepalive": env.int("KEEPALIVE", 60),
|
||||||
|
}
|
||||||
|
return config
|
||||||
|
|
||||||
|
|
||||||
|
def configure_logging(config: Dict[str, Any]) -> None:
|
||||||
|
"""Configure logging using the logging key from the configuration."""
|
||||||
|
logging.config.dictConfig(config["logging"])
|
||||||
|
log = getLogger(__name__)
|
||||||
|
log.info("%s v%s starting up.", config["sysname"], config["version"])
|
||||||
81
tests/test_config.py
Normal file
81
tests/test_config.py
Normal file
@ -0,0 +1,81 @@
|
|||||||
|
"""Tests for the configuration management."""
|
||||||
|
|
||||||
|
import logging
|
||||||
|
from typing import Any, Dict
|
||||||
|
|
||||||
|
import pytest
|
||||||
|
|
||||||
|
from hasskiosk import __version__
|
||||||
|
from hasskiosk.config import configure_logging, read_config
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.fixture(autouse=True)
|
||||||
|
def mock_env(monkeypatch):
|
||||||
|
"""Environment mock to test values."""
|
||||||
|
monkeypatch.setenv("HASSKIOSK_TOPIC_PRESENCE", "home/test/presence")
|
||||||
|
monkeypatch.setenv("HASSKIOSK_MQTT_HOST", "ha.example.com")
|
||||||
|
monkeypatch.setenv("HASSKIOSK_MQTT_USERNAME", "testymctesterson")
|
||||||
|
monkeypatch.setenv("HASSKIOSK_MQTT_PASSWORD", "hunter2")
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.fixture
|
||||||
|
def logging_config() -> Dict[str, Any]:
|
||||||
|
"""logging configuration fixture."""
|
||||||
|
config = {
|
||||||
|
"version": 1,
|
||||||
|
"disable_existing_loggers": False,
|
||||||
|
"formatters": {
|
||||||
|
"simple": {
|
||||||
|
"format": "%(asctime)s %(levelname)s (%(name)s) %(message)s",
|
||||||
|
"converter": "time.gmtime",
|
||||||
|
},
|
||||||
|
},
|
||||||
|
"handlers": {
|
||||||
|
"stdout": {
|
||||||
|
"class": "logging.StreamHandler",
|
||||||
|
"level": "DEBUG",
|
||||||
|
"formatter": "simple",
|
||||||
|
"stream": "ext://sys.stdout",
|
||||||
|
},
|
||||||
|
},
|
||||||
|
"loggers": {
|
||||||
|
"hasskiosk": {
|
||||||
|
"level": "DEBUG",
|
||||||
|
"propagate": "yes",
|
||||||
|
},
|
||||||
|
},
|
||||||
|
"root": {
|
||||||
|
"level": "WARN",
|
||||||
|
"handlers": ["stdout"],
|
||||||
|
},
|
||||||
|
}
|
||||||
|
return config
|
||||||
|
|
||||||
|
|
||||||
|
def test_read_config():
|
||||||
|
"""Test the read_config function."""
|
||||||
|
config = read_config()
|
||||||
|
assert config["version"] == __version__
|
||||||
|
assert config["sysname"] == "hasskiosk"
|
||||||
|
assert config["mqtt"] == {
|
||||||
|
"host": "ha.example.com",
|
||||||
|
"username": "testymctesterson",
|
||||||
|
"password": "hunter2",
|
||||||
|
"port": 1883,
|
||||||
|
"keepalive": 60,
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
|
def test_configure_logging(logging_config):
|
||||||
|
"""Test the logging configuration."""
|
||||||
|
config = {
|
||||||
|
"version": "1.2.3.test",
|
||||||
|
"sysname": "testkiosk",
|
||||||
|
"logging": logging_config,
|
||||||
|
}
|
||||||
|
configure_logging(config)
|
||||||
|
log = logging.getLogger(__name__)
|
||||||
|
rootlog = log
|
||||||
|
while rootlog.parent:
|
||||||
|
rootlog = rootlog.parent
|
||||||
|
assert len(rootlog.handlers) == 1
|
||||||
2
tox.ini
2
tox.ini
@ -61,7 +61,7 @@ deps =
|
|||||||
{[testenv:security]deps}
|
{[testenv:security]deps}
|
||||||
{[testenv:security]deps}
|
{[testenv:security]deps}
|
||||||
{[testenv:lint]deps}
|
{[testenv:lint]deps}
|
||||||
envdir = {toxinidir}/.venv
|
#envdir = {toxinidir}/.venv
|
||||||
usedevelop = True
|
usedevelop = True
|
||||||
|
|
||||||
[pycodestyle]
|
[pycodestyle]
|
||||||
|
|||||||
Loading…
x
Reference in New Issue
Block a user