diff --git a/.github/workflows/build.yml b/.github/workflows/build.yml index 3fb6a13..dbbac4d 100644 --- a/.github/workflows/build.yml +++ b/.github/workflows/build.yml @@ -11,7 +11,7 @@ jobs: runs-on: ubuntu-latest strategy: matrix: - python-version: ["3.9", "3.11", "3.x"] + python-version: ["3.11", "3.13", "3.x"] steps: - name: 🛒 Checkout repository @@ -26,7 +26,7 @@ jobs: - name: 👷🏻 Install dependencies run: | python -m pip install --upgrade pip - pip install -r requirements.txt pytest + pip install . build - name: 🏗️ Build application wheel run: python -m build @@ -35,7 +35,7 @@ jobs: runs-on: ubuntu-latest strategy: matrix: - python-version: ["3.9", "3.11", "3.x"] + python-version: ["3.11", "3.13", "3.x"] steps: - name: 🛒 Checkout repository @@ -50,7 +50,7 @@ jobs: - name: 👷🏻 Install dependencies run: | python -m pip install --upgrade pip - pip install -r requirements.txt pytest + pip install . pytest - name: 🕵🏻 Test with pytest run: pytest --doctest-modules --junitxml=junit/test-results-${{ matrix.python-version }}.xml @@ -63,7 +63,7 @@ jobs: # Use always() to always run this step to publish test results when there are test failures if: ${{ always() }} - build-pylint: + build-ruff: runs-on: ubuntu-latest steps: - name: 🛒 Checkout repository @@ -74,14 +74,15 @@ jobs: with: python-version: "3.x" - - name: 👷🏻 Install dependencies + - name: 👷🏻 Install ruff run: | python -m pip install --upgrade pip - pip install -r requirements.txt pylint pylint-exit + pip install ruff - - name: 🕵🏻 Analyse code with pylint + - name: 🕵🏻 Check code with ruff run: | - pylint --init-hook="import sys; import os; sys.path.append(os.path.abspath('.'));" $(git ls-files '*.py') || pylint-exit $? + ruff check . + ruff format --check . build-docker: runs-on: ubuntu-latest diff --git a/Dockerfile b/Dockerfile index ec06124..ebe5544 100644 --- a/Dockerfile +++ b/Dockerfile @@ -1,18 +1,26 @@ -FROM python:3.11-slim AS builder +FROM python:3.13-slim AS builder -RUN apt-get update && apt-get install -y --no-install-recommends binutils +RUN apt-get update && apt-get install -y --no-install-recommends binutils \ + && rm -rf /var/lib/apt/lists/* -COPY requirements.txt . -RUN pip install --user -r requirements.txt pyinstaller --no-warn-script-location - -COPY src/aps2mqtt/ /app/aps2mqtt -WORKDIR /app -RUN /root/.local/bin/pyinstaller --collect-all tzdata --onefile /app/aps2mqtt/__main__.py -n aps2mqtt +COPY . /build +WORKDIR /build +RUN pip install --no-cache-dir --user . pyinstaller --no-warn-script-location +RUN SITE_PACKAGES=$(python -c "import sysconfig; print(sysconfig.get_path('purelib', 'posix_user'))") \ + && /root/.local/bin/pyinstaller --collect-all tzdata --onefile "$SITE_PACKAGES/aps2mqtt/__main__.py" -n aps2mqtt FROM ubuntu:noble AS runner -RUN apt update && apt install tzdata -y && apt clean && rm -rf /var/lib/apt/lists/* -COPY --from=builder /app/dist/aps2mqtt /app/ +LABEL org.opencontainers.image.source="https://github.com/fligneul/aps2mqtt" \ + org.opencontainers.image.description="Access APSystems ECU data over MQTT" \ + org.opencontainers.image.licenses="GPL-3.0-or-later" + +RUN apt-get update && apt-get install -y --no-install-recommends tzdata \ + && rm -rf /var/lib/apt/lists/* \ + && groupadd -r aps2mqtt && useradd -r -g aps2mqtt aps2mqtt + +COPY --from=builder /build/dist/aps2mqtt /app/ +USER aps2mqtt WORKDIR /app CMD ["./aps2mqtt"] \ No newline at end of file diff --git a/MANIFEST.in b/MANIFEST.in deleted file mode 100644 index f364b3c..0000000 --- a/MANIFEST.in +++ /dev/null @@ -1,9 +0,0 @@ -# Exclude the entire test directory -prune test - -# Exclude macOS-specific files -global-exclude .DS_Store - -# Exclude Python bytecode and cache -global-exclude *.pyc -global-exclude __pycache__ \ No newline at end of file diff --git a/pyproject.toml b/pyproject.toml index 5ec55ee..58d6822 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -7,7 +7,7 @@ name = "aps2mqtt" dynamic = ["version"] description = "Access APSystems ECU data over MQTT" readme = "README.md" -requires-python = ">=3.9, <3.14" +requires-python = ">=3.11, <3.14" license = "GPL-3.0-or-later" license-files = ["LICENSE"] authors = [ @@ -18,8 +18,6 @@ urls = { Homepage = "https://github.com/fligneul/aps2mqtt" } classifiers = [ "Operating System :: OS Independent", "Programming Language :: Python :: 3", - "Programming Language :: Python :: 3.9", - "Programming Language :: Python :: 3.10", "Programming Language :: Python :: 3.11", "Programming Language :: Python :: 3.12", "Programming Language :: Python :: 3.13", @@ -29,7 +27,6 @@ dependencies = [ "requests~=2.32", "paho.mqtt~=2.1", "pyyaml~=6.0", - "numpy~=2.3", "suntime~=1.2.5", "str2bool~=1.1", "tzdata>=2025.2", @@ -38,8 +35,13 @@ dependencies = [ [tool.setuptools.dynamic] version = { attr = "aps2mqtt.__version__" } -[tool.black] +[tool.ruff] line-length = 100 +target-version = "py311" +exclude = ["src/aps2mqtt/apsystems/", "test/aps2mqtt/apsystems/"] + +[tool.ruff.lint] +select = ["E", "F", "W", "I", "UP", "B", "SIM", "N"] [tool.semantic_release] version_variables = ["src/aps2mqtt/__init__.py:__version__"] @@ -47,7 +49,7 @@ commit_message = "[release] {version}\n\nAutomatically generated by python-seman commit_parser = "conventional" major_on_zero = true tag_format = "v{version}" -build_command = "pip install -r requirements.txt && python -m build" +build_command = "pip install . && python -m build" [tool.semantic_release.branches.main] match = "(main|master)" diff --git a/requirements.txt b/requirements.txt deleted file mode 100644 index 149f434..0000000 --- a/requirements.txt +++ /dev/null @@ -1,10 +0,0 @@ -requests==2.32.4 -paho.mqtt==2.1.0 -pyyaml==6.0.2 -suntime==1.2.5 -build==1.2.2 -setuptools>=80.9.0 -twine==6.1.0 -python-semantic-release==10.2.0 -str2bool==1.1 -tzdata==2025.2 \ No newline at end of file diff --git a/src/aps2mqtt/config.py b/src/aps2mqtt/config.py index 8e97178..d61082d 100644 --- a/src/aps2mqtt/config.py +++ b/src/aps2mqtt/config.py @@ -1,23 +1,49 @@ """Application config classes, can be set by file or env variable""" import os +from dataclasses import dataclass, field from zoneinfo import ZoneInfo -from dateutil import tz + import yaml +from dateutil import tz from str2bool import str2bool_exc +@dataclass class MQTTDiscoveryConfig: """MQTT Discovery config""" - def __init__(self, cfg): + prefix: str + + def __init__(self, cfg: dict) -> None: self.prefix = cfg.get("MQTT_DISCOVERY_PREFIX", "homeassistant") +@dataclass +class WifiConfig: + """Wifi config of the ECU""" + + ssid: str + passwd: str + + +@dataclass class MQTTConfig: """MQTT config""" - def __init__(self, cfg): + broker_addr: str = "127.0.0.1" + broker_port: int = 1883 + broker_user: str = "" + broker_passwd: str = "" + client_id: str = "APS2MQTT" + topic_prefix: str = "" + retain: bool = False + discovery_enabled: bool = False + discovery: MQTTDiscoveryConfig | None = None + secured_connection: bool = False + cacerts_path: str | None = None + + def __init__(self, cfg: dict) -> None: self.broker_addr = cfg.get("MQTT_BROKER_HOST", "127.0.0.1") self.broker_port = int(cfg.get("MQTT_BROKER_PORT", 1883)) self.broker_user = cfg.get("MQTT_BROKER_USER", "") @@ -32,13 +58,23 @@ def __init__(self, cfg): str(cfg.get("MQTT_BROKER_SECURED_CONNECTION", False)) ) if self.secured_connection: - self.cacerts_path = cfg.get("MQTT_BROKER_CACERTS_PATH", None) + self.cacerts_path = cfg.get("MQTT_BROKER_CACERTS_PATH") +@dataclass class ECUConfig: """ECU config""" - def __init__(self, cfg): + ipaddr: str = "" + port: int = 8899 + timezone: ZoneInfo | tz.tzlocal = field(default_factory=tz.tzlocal) + auto_restart: bool = False + wifi_config: WifiConfig | None = None + stop_at_night: bool = False + ecu_position_latitude: float = 48.864716 + ecu_position_longitude: float = 2.349014 + + def __init__(self, cfg: dict) -> None: self.ipaddr = cfg["APS_ECU_IP"] self.port = int(cfg.get("APS_ECU_PORT", 8899)) ecu_timezone = cfg.get("APS_ECU_TIMEZONE", os.getenv("TZ", None)) @@ -54,29 +90,24 @@ def __init__(self, cfg): self.ecu_position_longitude = float(cfg.get("APS_ECU_POSITION_LNG", 2.349014)) -class WifiConfig: - """Wifi config of the ECU""" - - def __init__(self, ssid, passwd): - self.ssid = ssid - self.passwd = passwd - - class Config: """Application config""" - def __init__(self, config_path=None): + mqtt_config: MQTTConfig + ecu_config: ECUConfig + + def __init__(self, config_path: str | None = None) -> None: if config_path is not None: - self.__load_yaml_config_file(config_path) + self._load_yaml_config_file(config_path) elif os.getenv("CONFIG_FILE") is not None: - self.__load_yaml_config_file(os.getenv("CONFIG_FILE")) + self._load_yaml_config_file(os.getenv("CONFIG_FILE")) else: cfg = os.environ self.mqtt_config = MQTTConfig(cfg) self.ecu_config = ECUConfig(cfg) - def __load_yaml_config_file(self, config_path): - with open(config_path, "r", encoding="UTF-8") as yml_cfg: + def _load_yaml_config_file(self, config_path: str) -> None: + with open(config_path, encoding="UTF-8") as yml_cfg: cfg = yaml.safe_load(yml_cfg) self.mqtt_config = MQTTConfig(cfg["mqtt"]) self.ecu_config = ECUConfig(cfg["ecu"]) diff --git a/src/aps2mqtt/main.py b/src/aps2mqtt/main.py index d031a5b..ee963b8 100644 --- a/src/aps2mqtt/main.py +++ b/src/aps2mqtt/main.py @@ -3,18 +3,19 @@ import logging import os import time +from argparse import ArgumentParser, Namespace +from datetime import UTC, datetime, timedelta -from argparse import ArgumentParser -from datetime import datetime, timedelta, timezone from str2bool import str2bool_exc -from aps2mqtt.mqtthandler import MQTTHandler -from aps2mqtt.config import Config + from aps2mqtt.apsystems.ECU import ECU +from aps2mqtt.config import Config +from aps2mqtt.mqtthandler import MQTTHandler _LOGGER = logging.getLogger(__name__) -def cli_args(): +def cli_args() -> Namespace: """Create CLI arguments and parse them""" parser = ArgumentParser(prog="aps2mqtt") parser.add_argument( @@ -27,7 +28,7 @@ def cli_args(): return parser.parse_args() -def main(): +def main() -> None: """Application main""" args = cli_args() conf = Config(args.config_path) @@ -37,14 +38,14 @@ def main(): else: logging.basicConfig(level=logging.INFO) - update_time = datetime(1970, 1, 1, tzinfo=timezone.utc) + update_time = datetime(1970, 1, 1, tzinfo=UTC) ecu = ECU(conf.ecu_config) mqtt_handler = MQTTHandler(conf.mqtt_config) mqtt_handler.connect_mqtt() - while 1: - if datetime.now(timezone.utc) > update_time: + while True: + if datetime.now(UTC) > update_time: if ecu.should_sleep(): update_time = ecu.wake_up_time() _LOGGER.info( @@ -54,15 +55,15 @@ def main(): else: try: data = ecu.update() - if data is None or len(data) == 0: + if not data: raise ValueError("Retrieved data are empty") update_time = datetime.strptime(data["timestamp"], "%Y-%m-%d %H:%M:%S").replace( tzinfo=conf.ecu_config.timezone ) + timedelta(seconds=360) mqtt_handler.publish_values(data) except Exception as e: - update_time = datetime.now(timezone.utc) + timedelta(seconds=60) - _LOGGER.error("An exception occured: %s -> %s", e.__class__.__name__, str(e)) + update_time = datetime.now(UTC) + timedelta(seconds=60) + _LOGGER.error("An exception occurred: %s -> %s", e.__class__.__name__, str(e)) _LOGGER.debug("Exception trace:", exc_info=True) _LOGGER.info( "Update finished, next update at: %s", diff --git a/src/aps2mqtt/mqtthandler.py b/src/aps2mqtt/mqtthandler.py index 78f96fd..70b7848 100644 --- a/src/aps2mqtt/mqtthandler.py +++ b/src/aps2mqtt/mqtthandler.py @@ -1,13 +1,18 @@ """Handle MQTT connection and data publishing""" -import logging -import time import atexit import json +import logging +import time from statistics import mean +from typing import TYPE_CHECKING + import certifi from paho.mqtt import client as mqtt_client +if TYPE_CHECKING: + from aps2mqtt.config import MQTTConfig + _LOGGER = logging.getLogger(__name__) _MAX_RETRY = 10 @@ -16,7 +21,7 @@ class MQTTHandler: """Handle MQTT connection to broker and publish message""" - def __init__(self, mqtt_config): + def __init__(self, mqtt_config: "MQTTConfig") -> None: self.mqtt_config = mqtt_config self.topic_prefix = ( mqtt_config.topic_prefix + "/" if len(mqtt_config.topic_prefix.strip()) > 0 else "" @@ -27,7 +32,7 @@ def __init__(self, mqtt_config): self.status_topic = self.topic_prefix + "aps/status" self.discovery_messages_sent = False - def on_connect(self, client, userdata, flags, reason_code, properties): + def on_connect(self, client, userdata, flags, reason_code, properties) -> None: """Callback function on broker connection""" del userdata, flags, properties if reason_code == 0: @@ -36,12 +41,12 @@ def on_connect(self, client, userdata, flags, reason_code, properties): else: _LOGGER.error("Failed to connect: %s", reason_code) - def on_disconnect(self, client, userdata, flags, reason_code, properties): + def on_disconnect(self, client, userdata, flags, reason_code, properties) -> None: """Callback function on broker disconnection""" del client, userdata, flags, properties _LOGGER.info("Disconnected from MQTT Broker: %s", reason_code) - def _publish(self, client, topic, msg, retain=False): + def _publish(self, client, topic: str, msg: str, retain: bool = False) -> None: # If mqtt_retain is True in config, all messages are retained. # Otherwise, only LWT uses retain. actual_retain = retain or self.mqtt_config.retain @@ -51,7 +56,7 @@ def _publish(self, client, topic, msg, retain=False): else: _LOGGER.error("Failed to send message to topic %s: %s", topic, result.rc) - def connect_mqtt(self): + def connect_mqtt(self) -> None: """Create connection to MQTT broker""" _LOGGER.debug("Create MQTT client") self.client = mqtt_client.Client( @@ -96,13 +101,13 @@ def connect_mqtt(self): self.client.loop_start() atexit.register(self.disconnect) - def disconnect(self): + def disconnect(self) -> None: if self.client.is_connected(): _LOGGER.info("Publishing 'offline' status on graceful exit.") self._publish(self.client, self.status_topic, "offline", retain=True) self.client.loop_stop() - def publish_discovery_messages(self, data): + def publish_discovery_messages(self, data: dict) -> None: """Publish discovery messages for all sensors""" if not self.mqtt_config.discovery_enabled or self.discovery_messages_sent: return @@ -229,7 +234,7 @@ def publish_discovery_messages(self, data): ) # Panel sensors - for i, panel_power in enumerate(inverter.get("power", [])): + for i, _panel_power in enumerate(inverter.get("power", [])): panel_num = i + 1 self._publish_discovery_payload( "sensor", @@ -244,7 +249,7 @@ def publish_discovery_messages(self, data): "mdi:solar-panel", ) - for i, panel_voltage in enumerate(inverter.get("voltage", [])): + for i, _panel_voltage in enumerate(inverter.get("voltage", [])): panel_num = i + 1 self._publish_discovery_payload( "sensor", @@ -261,7 +266,9 @@ def publish_discovery_messages(self, data): self.discovery_messages_sent = True - def _get_device_payload(self, device_id, device_name, via_device=None): + def _get_device_payload( + self, device_id: str, device_name: str, via_device: str | None = None + ) -> dict: payload = { "identifiers": [str(device_id)], "name": f"APS {device_name} {device_id}", @@ -273,17 +280,17 @@ def _get_device_payload(self, device_id, device_name, via_device=None): def _publish_discovery_payload( self, - component, - device_id, - object_id, - device_payload, - state_topic_base, - value_key, - name, - device_class, - unit, - icon, - ): + component: str, + device_id: str, + object_id: str, + device_payload: dict, + state_topic_base: str, + value_key: str, + name: str, + device_class: str | None, + unit: str | None, + icon: str | None, + ) -> None: discovery_topic = f"{self.discovery_topic}{component}/aps_{device_id}_{object_id}/config" payload = { "name": name, @@ -307,7 +314,7 @@ def _publish_discovery_payload( self._publish(self.client, discovery_topic, json.dumps(payload), retain=True) - def publish_values(self, data): + def publish_values(self, data: dict) -> None: """Publish ECU data to MQTT""" _LOGGER.debug("Start MQTT publish") @@ -327,7 +334,7 @@ def publish_values(self, data): self._publish(self.client, topic, value) _LOGGER.debug("MQTT values published") - def _parse_data(self, data): + def _parse_data(self, data: dict) -> dict: output = {} ecu_id = data["ecu_id"] ecu_topic_base = self.topic_prefix + "aps/" + str(ecu_id) diff --git a/src/aps2mqtt/py.typed b/src/aps2mqtt/py.typed new file mode 100644 index 0000000..e69de29 diff --git a/test/aps2mqtt/apsystems/test_APSystemsSocket.py b/test/aps2mqtt/apsystems/test_APSystemsSocket.py index 7f816af..266b497 100644 --- a/test/aps2mqtt/apsystems/test_APSystemsSocket.py +++ b/test/aps2mqtt/apsystems/test_APSystemsSocket.py @@ -1,6 +1,8 @@ import unittest -from unittest.mock import patch, MagicMock, call -from aps2mqtt.apsystems.APSystemsSocket import APSystemsSocket, APSystemsInvalidData +from unittest.mock import call, patch + +from aps2mqtt.apsystems.APSystemsSocket import APSystemsInvalidData, APSystemsSocket + class TestAPSystemsSocket(unittest.TestCase): diff --git a/test/aps2mqtt/apsystems/test_ecu.py b/test/aps2mqtt/apsystems/test_ecu.py index fcfa5b6..0ef5f9a 100644 --- a/test/aps2mqtt/apsystems/test_ecu.py +++ b/test/aps2mqtt/apsystems/test_ecu.py @@ -1,11 +1,12 @@ import logging import unittest +from datetime import datetime, timedelta from unittest.mock import MagicMock, patch -from datetime import datetime, timedelta, timezone -from zoneinfo import ZoneInfo -from aps2mqtt.apsystems.ECU import ECU + from suntime import Sun +from aps2mqtt.apsystems.ECU import ECU + class TestECU(unittest.TestCase): diff --git a/test/aps2mqtt/test_config.py b/test/aps2mqtt/test_config.py index eb17af3..c55793e 100644 --- a/test/aps2mqtt/test_config.py +++ b/test/aps2mqtt/test_config.py @@ -1,25 +1,30 @@ -import unittest -from unittest.mock import patch, mock_open import os +import unittest +from unittest.mock import mock_open, patch + import yaml -from aps2mqtt.config import MQTTConfig, ECUConfig, WifiConfig, Config -class TestConfig(unittest.TestCase): +from aps2mqtt.config import Config, ECUConfig, MQTTConfig + +class TestConfig(unittest.TestCase): def test_mqtt_config_from_env(self): - with patch.dict(os.environ, { - "MQTT_BROKER_HOST": "mqtt.example.com", - "MQTT_BROKER_PORT": "1884", - "MQTT_BROKER_USER": "user", - "MQTT_BROKER_PASSWD": "password", - "MQTT_CLIENT_ID": "test_client", - "MQTT_TOPIC_PREFIX": "test/topic", - "MQTT_RETAIN": "True", - "MQTT_DISCOVERY_ENABLED": "True", - "MQTT_DISCOVERY_PREFIX": "test_discovery", - "MQTT_BROKER_SECURED_CONNECTION": "True", - "MQTT_BROKER_CACERTS_PATH": "/path/to/ca.crt" - }): + with patch.dict( + os.environ, + { + "MQTT_BROKER_HOST": "mqtt.example.com", + "MQTT_BROKER_PORT": "1884", + "MQTT_BROKER_USER": "user", + "MQTT_BROKER_PASSWD": "password", + "MQTT_CLIENT_ID": "test_client", + "MQTT_TOPIC_PREFIX": "test/topic", + "MQTT_RETAIN": "True", + "MQTT_DISCOVERY_ENABLED": "True", + "MQTT_DISCOVERY_PREFIX": "test_discovery", + "MQTT_BROKER_SECURED_CONNECTION": "True", + "MQTT_BROKER_CACERTS_PATH": "/path/to/ca.crt", + }, + ): cfg = MQTTConfig(os.environ) self.assertEqual(cfg.broker_addr, "mqtt.example.com") self.assertEqual(cfg.broker_port, 1884) @@ -34,17 +39,20 @@ def test_mqtt_config_from_env(self): self.assertEqual(cfg.cacerts_path, "/path/to/ca.crt") def test_ecu_config_from_env(self): - with patch.dict(os.environ, { - "APS_ECU_IP": "192.168.1.100", - "APS_ECU_PORT": "8888", - "APS_ECU_TIMEZONE": "America/New_York", - "APS_ECU_AUTO_RESTART": "True", - "APS_ECU_WIFI_SSID": "my_wifi", - "APS_ECU_WIFI_PASSWD": "wifi_password", - "APS_ECU_STOP_AT_NIGHT": "True", - "APS_ECU_POSITION_LAT": "40.7128", - "APS_ECU_POSITION_LNG": "-74.0060" - }): + with patch.dict( + os.environ, + { + "APS_ECU_IP": "192.168.1.100", + "APS_ECU_PORT": "8888", + "APS_ECU_TIMEZONE": "America/New_York", + "APS_ECU_AUTO_RESTART": "True", + "APS_ECU_WIFI_SSID": "my_wifi", + "APS_ECU_WIFI_PASSWD": "wifi_password", + "APS_ECU_STOP_AT_NIGHT": "True", + "APS_ECU_POSITION_LAT": "40.7128", + "APS_ECU_POSITION_LNG": "-74.0060", + }, + ): cfg = ECUConfig(os.environ) self.assertEqual(cfg.ipaddr, "192.168.1.100") self.assertEqual(cfg.port, 8888) @@ -64,12 +72,15 @@ def test_config_from_yaml(self): ecu: APS_ECU_IP: 192.168.1.200 """ - with patch("builtins.open", mock_open(read_data=yaml_content)): - with patch.object(yaml, 'safe_load', return_value=yaml.safe_load(yaml_content)): - cfg = Config(config_path="dummy_path.yaml") - self.assertEqual(cfg.mqtt_config.broker_addr, "yaml.broker") - self.assertEqual(cfg.mqtt_config.broker_port, 1885) - self.assertEqual(cfg.ecu_config.ipaddr, "192.168.1.200") + with ( + patch("builtins.open", mock_open(read_data=yaml_content)), + patch.object(yaml, "safe_load", return_value=yaml.safe_load(yaml_content)), + ): + cfg = Config(config_path="dummy_path.yaml") + self.assertEqual(cfg.mqtt_config.broker_addr, "yaml.broker") + self.assertEqual(cfg.mqtt_config.broker_port, 1885) + self.assertEqual(cfg.ecu_config.ipaddr, "192.168.1.200") + -if __name__ == '__main__': +if __name__ == "__main__": unittest.main() diff --git a/test/aps2mqtt/test_main.py b/test/aps2mqtt/test_main.py index a492379..77bac12 100644 --- a/test/aps2mqtt/test_main.py +++ b/test/aps2mqtt/test_main.py @@ -1,12 +1,13 @@ import unittest -from unittest.mock import patch, MagicMock -from datetime import datetime, timedelta, timezone +from datetime import UTC, datetime +from unittest.mock import MagicMock, patch from zoneinfo import ZoneInfo + from aps2mqtt.main import cli_args, main -class TestMain(unittest.TestCase): - @patch('aps2mqtt.main.ArgumentParser') +class TestMain(unittest.TestCase): + @patch("aps2mqtt.main.ArgumentParser") def test_cli_args(self, mock_parser): # Arrange mock_parser_instance = MagicMock() @@ -24,13 +25,15 @@ def test_cli_args(self, mock_parser): ) mock_parser_instance.parse_args.assert_called_once() - @patch('aps2mqtt.main.Config') - @patch('aps2mqtt.main.ECU') - @patch('aps2mqtt.main.MQTTHandler') - @patch('aps2mqtt.main.time.sleep', side_effect=InterruptedError) # To break the loop - @patch('aps2mqtt.main.datetime') - @patch('aps2mqtt.main.cli_args') - def test_main_loop(self, mock_cli_args, mock_datetime, mock_sleep, mock_mqtt, mock_ecu, mock_config): + @patch("aps2mqtt.main.Config") + @patch("aps2mqtt.main.ECU") + @patch("aps2mqtt.main.MQTTHandler") + @patch("aps2mqtt.main.time.sleep", side_effect=InterruptedError) # To break the loop + @patch("aps2mqtt.main.datetime") + @patch("aps2mqtt.main.cli_args") + def test_main_loop( + self, mock_cli_args, mock_datetime, mock_sleep, mock_mqtt, mock_ecu, mock_config + ): # Arrange mock_args = MagicMock() mock_args.config_path = None @@ -51,12 +54,14 @@ def test_main_loop(self, mock_cli_args, mock_datetime, mock_sleep, mock_mqtt, mo # Mock datetime.now to control the loop mock_datetime.now.side_effect = [ - datetime(2025, 7, 20, 11, 59, 0, tzinfo=timezone.utc), # Before update_time - datetime(2025, 7, 20, 12, 0, 1, tzinfo=timezone.utc), # After update_time, to trigger update - datetime(2025, 7, 20, 12, 10, 0, tzinfo=timezone.utc) # After next update_time + datetime(2025, 7, 20, 11, 59, 0, tzinfo=UTC), # Before update_time + datetime(2025, 7, 20, 12, 0, 1, tzinfo=UTC), # After update_time, to trigger update + datetime(2025, 7, 20, 12, 10, 0, tzinfo=UTC), # After next update_time ] mock_datetime.strptime.return_value = datetime(2025, 7, 20, 12, 0, 0) - mock_datetime.side_effect = lambda *args, **kw: datetime(*args, **kw) if args else mock_datetime.now() + mock_datetime.side_effect = lambda *args, **kw: ( + datetime(*args, **kw) if args else mock_datetime.now() + ) # Act with self.assertRaises(InterruptedError): @@ -66,5 +71,6 @@ def test_main_loop(self, mock_cli_args, mock_datetime, mock_sleep, mock_mqtt, mo mock_ecu_instance.update.assert_called_once() mock_mqtt_instance.publish_values.assert_called_once() -if __name__ == '__main__': + +if __name__ == "__main__": unittest.main() diff --git a/test/aps2mqtt/test_mqtthandler.py b/test/aps2mqtt/test_mqtthandler.py index bdebe34..e32d31b 100644 --- a/test/aps2mqtt/test_mqtthandler.py +++ b/test/aps2mqtt/test_mqtthandler.py @@ -1,11 +1,11 @@ -import unittest -from unittest.mock import MagicMock, patch, call import json +import unittest +from unittest.mock import MagicMock, patch + from aps2mqtt.mqtthandler import MQTTHandler class TestMQTTHandler(unittest.TestCase): - def setUp(self): self.mock_mqtt_config = MagicMock() self.mock_mqtt_config.discovery_enabled = True