test: add energy mode control tests and CI workflow

Author: deKibi

.github/workflows/tests.yml

@@ -0,0 +1,30 @@
+name: Tests
+
+on:
+  push:
+  pull_request:
+
+permissions:
+  contents: read
+
+jobs:
+  tests:
+    runs-on: ubuntu-24.04
+    timeout-minutes: 10
+
+    steps:
+      - name: Check out repository
+        uses: actions/checkout@v6
+
+      - name: Set up Python
+        uses: actions/setup-python@v6
+        with:
+          python-version: "3.12"
+
+      - name: Install dependencies
+        run: |
+          python -m pip install --upgrade pip
+          python -m pip install -r requirements-dev.txt
+
+      - name: Run tests
+        run: python -m pytest

energy_mode_control/energy_mode_controller.py

@@ -2,7 +2,7 @@
 
 # Standard Libraries
 import logging
-from typing import Any, Final
+from typing import Any, Final, Protocol
 
 # Custom Modules
 from config import (
@@ -13,10 +13,8 @@
     GRID_OUTAGE_TARGET_MODE,
     EnergyMode
 )
+from energy_mode_control.mode_decision import decide_target_mode
 from energy_mode_control.time_utils import is_time_reached
-from energy_mode_control.energy_mode_switcher import (
-    switch_energy_mode,
-)
 
 
 LOGGER = logging.getLogger(__name__)
@@ -26,15 +24,26 @@
 GRID_OUTAGE_VOLTAGE_THRESHOLD: Final[float] = 10.0
 
 
-def handle_energy_mode_control(must_data: dict[str, Any] | None) -> bool:
+class InverterClient(Protocol):
+    def switch_energy_mode(self, target_mode: EnergyMode) -> None:
+        """Switch the inverter to the requested energy mode."""
+
+
+def handle_energy_mode_control(
+    must_data: dict[str, Any] | None,
+    inverter_client: InverterClient,
+) -> bool:
     try:
-        return _handle_energy_mode_control(must_data)
+        return _handle_energy_mode_control(must_data, inverter_client)
     except Exception:
         LOGGER.exception("Failed to handle energy mode control.")
         return False
 
 
-def _handle_energy_mode_control(must_data: dict[str, Any] | None) -> bool:
+def _handle_energy_mode_control(
+    must_data: dict[str, Any] | None,
+    inverter_client: InverterClient,
+) -> bool:
     if not ENABLE_AUTO_SWITCH and not ENABLE_GRID_OUTAGE_AUTO_SWITCH:
         LOGGER.debug("All automatic energy mode control features are disabled.")
         return False
@@ -76,46 +85,51 @@ def _handle_energy_mode_control(must_data: dict[str, Any] | None) -> bool:
         grid_voltage >= GRID_OUTAGE_VOLTAGE_THRESHOLD
     )
 
-    target_mode: EnergyMode | None = None
-    switch_reason: str | None = None
-
-    # Grid outage rule has the highest priority.
-    if (
-            ENABLE_GRID_OUTAGE_AUTO_SWITCH
-            and not is_grid_available
-    ):
-        target_mode = GRID_OUTAGE_TARGET_MODE
-        switch_reason = "electrical grid is unavailable"
-
-    # Time-based rule is checked when the grid is available.
-    elif (
+    is_auto_switch_time_reached = (
         ENABLE_AUTO_SWITCH
         and is_grid_available
         and is_time_reached(AUTO_SWITCH_TARGET_TIME)
-    ):
-        target_mode = AUTO_SWITCH_TARGET_MODE
-        switch_reason = "auto-switch target time has been reached"
+    )
+    target_mode = decide_target_mode(
+        is_grid_available=is_grid_available,
+        is_auto_switch_enabled=ENABLE_AUTO_SWITCH,
+        is_auto_switch_time_reached=is_auto_switch_time_reached,
+        auto_switch_target_mode=AUTO_SWITCH_TARGET_MODE,
+        is_grid_outage_auto_switch_enabled=(
+            ENABLE_GRID_OUTAGE_AUTO_SWITCH
+        ),
+        grid_outage_target_mode=GRID_OUTAGE_TARGET_MODE,
+    )
+
+    return execute_mode_decision(
+        current_mode=current_energy_mode,
+        target_mode=target_mode,
+        inverter_client=inverter_client,
+    )
+
+
+def execute_mode_decision(
+    current_mode: EnergyMode,
+    target_mode: EnergyMode | None,
+    inverter_client: InverterClient,
+) -> bool:
+    """Execute a mode decision when the target differs from current mode."""
 
     if target_mode is None:
         LOGGER.debug("No energy mode switch is currently required.")
         return False
 
-    if current_energy_mode == target_mode:
+    if current_mode == target_mode:
         LOGGER.debug(
             "Inverter is already operating in %s mode.",
             target_mode.name,
         )
         return False
 
     LOGGER.info(
-        "Energy mode switch required: %s -> %s. Reason: %s.",
-        current_energy_mode.name,
+        "Energy mode switch required: %s -> %s.",
+        current_mode.name,
         target_mode.name,
-        switch_reason,
-    )
-
-    # The real inverter command will be added here:
-    switch_energy_mode(
-        target_mode=target_mode,
     )
+    inverter_client.switch_energy_mode(target_mode=target_mode)
     return True

energy_mode_control/mode_decision.py

@@ -0,0 +1,27 @@
+# energy_mode_control/mode_decision.py
+
+# Custom Modules
+from config import EnergyMode
+
+
+def decide_target_mode(
+    *,
+    is_grid_available: bool,
+    is_auto_switch_enabled: bool,
+    is_auto_switch_time_reached: bool,
+    auto_switch_target_mode: EnergyMode,
+    is_grid_outage_auto_switch_enabled: bool,
+    grid_outage_target_mode: EnergyMode,
+) -> EnergyMode | None:
+    """Return the required target mode without performing side effects."""
+    if is_grid_outage_auto_switch_enabled and not is_grid_available:
+        return grid_outage_target_mode
+
+    if (
+        is_auto_switch_enabled
+        and is_grid_available
+        and is_auto_switch_time_reached
+    ):
+        return auto_switch_target_mode
+
+    return None

energy_mode_control/time_utils.py

@@ -20,6 +20,15 @@ def get_current_kyiv_datetime() -> datetime:
     return datetime.now(tz=KYIV_TIMEZONE)
 
 
+def is_time_in_day_window(
+    current_time: time,
+    start_time: time,
+    end_time: time,
+) -> bool:
+    """Return whether the current time is in a daytime interval."""
+    return start_time <= current_time < end_time
+
+
 def is_time_reached(
     target_time: time,
     current_datetime: datetime | None = None,

main.py

@@ -24,6 +24,7 @@
     MYSQL_HOST,
     MYSQL_PASSWORD,
     MYSQL_USER,
+    EnergyMode,
     get_startup_configuration_summary,
     log_configuration_warnings,
 )
@@ -149,6 +150,14 @@ def get_data(self) -> Optional[dict]:
         except Exception:
             LOGGER.exception("Failed to get inverter data.")
 
+    def switch_energy_mode(self, target_mode: EnergyMode) -> None:
+        from energy_mode_control.energy_mode_switcher import switch_energy_mode
+
+        switch_energy_mode(
+            target_mode=target_mode,
+            port=self._serial_port,
+        )
+
 
 def main():
     configure_logging()
@@ -198,7 +207,10 @@ def main():
             LOGGER.info("Inverter data inserted into the database.")
 
             # 3. Handle optional energy mode control
-            handle_energy_mode_control(must_data=must_data)
+            handle_energy_mode_control(
+                must_data=must_data,
+                inverter_client=must_inverter_data_handler,
+            )
         else:
             LOGGER.warning("Unable to get data from the inverter.")
 

pytest.ini

@@ -0,0 +1,3 @@
+[pytest]
+testpaths = tests
+addopts = -ra

requirements-dev.txt

@@ -0,0 +1,2 @@
+-r requirements.txt
+pytest~=9.1

tests/__init__.py

@@ -0,0 +1 @@
+# tests/__init__.py

tests/conftest.py

@@ -0,0 +1,8 @@
+# tests/conftest.py
+
+# Third-party Libraries
+import dotenv
+
+
+# Keep business-logic tests independent from the developer's local .env file.
+dotenv.load_dotenv = lambda *args, **kwargs: False

tests/energy_mode_control/__init__.py

@@ -0,0 +1 @@
+# tests/energy_mode_control/__init__.py