feat: update override functions for better usability (#125)

* feat: update override functions for better usability

* update tests

Author: firstof9

openevsehttp/__main__.py

@@ -19,6 +19,7 @@
     MissingSerial,
     ParseJSONError,
     UnknownError,
+    UnsupportedFeature,
 )
 from .websocket import (
     SIGNAL_CONNECTION_STATE,
@@ -300,6 +301,9 @@ async def divert_mode(self, mode: str = "normal") -> None:
 
     async def get_override(self) -> None:
         """Get the manual override status."""
+        if not self._version_check("4.0.0"):
+            _LOGGER.debug("Feature not supported for older firmware.")
+            raise UnsupportedFeature
         url = f"{self.url}override"
 
         _LOGGER.debug("Getting data from %s", url)
@@ -309,27 +313,37 @@ async def get_override(self) -> None:
     async def set_override(
         self,
         state: str,
-        charge_current: int,
-        max_current: int,
-        energy_limit: int,
-        time_limit: int,
+        charge_current: int | None = None,
+        max_current: int | None = None,
+        energy_limit: int | None = None,
+        time_limit: int | None = None,
         auto_release: bool = True,
-    ) -> str:
+    ) -> Any:
         """Set the manual override status."""
+        if not self._version_check("4.0.0"):
+            _LOGGER.debug("Feature not supported for older firmware.")
+            raise UnsupportedFeature
         url = f"{self.url}override"
 
         if state not in ["active", "disabled"]:
+            _LOGGER.error("Invalid override state: %s", state)
             raise ValueError
 
         data = {
             "state": state,
-            "charge_current": charge_current,
-            "max_current": max_current,
-            "energy_limit": energy_limit,
-            "time_limit": time_limit,
             "auto_release": auto_release,
         }
 
+        if charge_current:
+            data["charge_current"] = charge_current
+        if max_current:
+            data["max_current"] = max_current
+        if energy_limit:
+            data["energy_limit"] = energy_limit
+        if time_limit:
+            data["time_limit"] = time_limit
+
+        _LOGGER.debug("Override data: %s", data)
         _LOGGER.debug("Setting override config on %s", url)
         response = await self.process_request(
             url=url, method="post", data=data
@@ -340,22 +354,7 @@ async def toggle_override(self) -> None:
         """Toggle the manual override status."""
         #   3.x: use RAPI commands $FE (enable) and $FS (sleep)
         #   4.x: use HTTP API call
-
-        cutoff = AwesomeVersion("4.0.0")
-        current = ""
-
-        _LOGGER.debug("Detected firmware: %s", self._config["version"])
-
-        if "dev" in self._config["version"]:
-            value = self._config["version"]
-            _LOGGER.debug("Stripping 'dev' from version.")
-            value = value.split(".")
-            value = ".".join(value[0:3])
-            current = AwesomeVersion(value)
-        else:
-            current = AwesomeVersion(self._config["version"])
-
-        if current >= cutoff:
+        if self._version_check("4.0.0"):
             url = f"{self.url}override"
 
             _LOGGER.debug("Toggling manual override %s", url)
@@ -370,6 +369,9 @@ async def toggle_override(self) -> None:
 
     async def clear_override(self) -> None:
         """Clear the manual override status."""
+        if not self._version_check("4.0.0"):
+            _LOGGER.debug("Feature not supported for older firmware.")
+            raise UnsupportedFeature
         url = f"{self.url}override"
 
         _LOGGER.debug("Clearing manual override %s", url)
@@ -381,21 +383,8 @@ async def set_current(self, amps: int = 6) -> None:
         #   3.x - 4.1.0: use RAPI commands $SC <amps>
         #   4.1.2: use HTTP API call
         amps = int(amps)
-        cutoff = AwesomeVersion("4.1.2")
-        current = ""
-
-        _LOGGER.debug("Detected firmware: %s", current)
-
-        if "dev" in self._config["version"]:
-            value = self._config["version"]
-            _LOGGER.debug("Stripping 'dev' from version.")
-            value = value.split(".")
-            value = ".".join(value[0:3])
-            current = AwesomeVersion(value)
-        else:
-            current = AwesomeVersion(self._config["version"])
 
-        if current >= cutoff:
+        if self._version_check("4.1.2"):
             url = f"{self.url}config"
 
             if (
@@ -488,6 +477,26 @@ async def firmware_check(self) -> dict | None:
                 response["release_url"] = message["html_url"]
                 return response
 
+    def _version_check(self, min_version: str) -> bool:
+        """Return bool if minimum version is met."""
+        cutoff = AwesomeVersion(min_version)
+        current = ""
+
+        _LOGGER.debug("Detected firmware: %s", self._config["version"])
+
+        if "dev" in self._config["version"]:
+            value = self._config["version"]
+            _LOGGER.debug("Stripping 'dev' from version.")
+            value = value.split(".")
+            value = ".".join(value[0:3])
+            current = AwesomeVersion(value)
+        else:
+            current = AwesomeVersion(self._config["version"])
+
+        if current >= cutoff:
+            return True
+        return False
+
     @property
     def hostname(self) -> str:
         """Return charger hostname."""

openevsehttp/exceptions.py

@@ -23,3 +23,7 @@ class AlreadyListening(Exception):
 
 class MissingSerial(Exception):
     """Exception for missing serial number."""
+
+
+class UnsupportedFeature(Exception):
+    """Exception for firmware that is too old."""

tests/test_main.py

@@ -10,7 +10,7 @@
 
 import openevsehttp.__main__ as main
 from tests.common import load_fixture
-from openevsehttp.exceptions import MissingSerial
+from openevsehttp.exceptions import MissingSerial, UnsupportedFeature
 
 pytestmark = pytest.mark.asyncio
 
@@ -971,3 +971,115 @@ async def test_max_current_soft(fixture, expected, request):
     await charger.update()
     status = charger.max_current_soft
     assert status == expected
+
+
+async def test_set_override(test_charger, test_charger_v2, mock_aioclient, caplog):
+    """Test set override function."""
+    await test_charger.update()
+    mock_aioclient.post(
+        TEST_URL_OVERRIDE,
+        status=200,
+        body='{"msg": "OK"}',
+    )
+    with caplog.at_level(logging.DEBUG):
+        status = await test_charger.set_override("active")
+        assert status == {"msg": "OK"}
+        assert "Override data: {'state': 'active', 'auto_release': True}" in caplog.text
+
+        mock_aioclient.post(
+            TEST_URL_OVERRIDE,
+            status=200,
+            body='{"msg": "OK"}',
+        )
+        status = await test_charger.set_override("active", 30)
+        assert (
+            "Override data: {'state': 'active', 'auto_release': True, 'charge_current': 30}"
+            in caplog.text
+        )
+        mock_aioclient.post(
+            TEST_URL_OVERRIDE,
+            status=200,
+            body='{"msg": "OK"}',
+        )
+        status = await test_charger.set_override("active", 30, 32)
+        assert (
+            "Override data: {'state': 'active', 'auto_release': True, 'charge_current': 30, 'max_current': 32}"
+            in caplog.text
+        )
+        mock_aioclient.post(
+            TEST_URL_OVERRIDE,
+            status=200,
+            body='{"msg": "OK"}',
+        )
+        status = await test_charger.set_override("active", 30, 32, 2000)
+        assert (
+            "Override data: {'state': 'active', 'auto_release': True, 'charge_current': 30, 'max_current': 32, 'energy_limit': 2000}"
+            in caplog.text
+        )
+        mock_aioclient.post(
+            TEST_URL_OVERRIDE,
+            status=200,
+            body='{"msg": "OK"}',
+        )
+        status = await test_charger.set_override("active", 30, 32, 2000, 5000)
+        assert (
+            "Override data: {'state': 'active', 'auto_release': True, 'charge_current': 30, 'max_current': 32, 'energy_limit': 2000, 'time_limit': 5000}"
+            in caplog.text
+        )
+
+    with pytest.raises(ValueError):
+        with caplog.at_level(logging.DEBUG):
+            await test_charger.set_override("invalid")
+            assert "Invalid override state: invalid" in caplog.text
+
+    with pytest.raises(UnsupportedFeature):
+        with caplog.at_level(logging.DEBUG):
+            await test_charger_v2.update()
+            status = await test_charger_v2.set_override("active")
+            assert "Feature not supported for older firmware." in caplog.text
+
+
+async def test_clear_override(test_charger, test_charger_v2, mock_aioclient, caplog):
+    """Test clear override function."""
+    await test_charger.update()
+    mock_aioclient.delete(
+        TEST_URL_OVERRIDE,
+        status=200,
+        body='{"msg": "OK"}',
+    )
+    with caplog.at_level(logging.DEBUG):
+        await test_charger.clear_override()
+        assert "Toggle response: OK" in caplog.text
+
+    with pytest.raises(UnsupportedFeature):
+        with caplog.at_level(logging.DEBUG):
+            await test_charger_v2.update()
+            await test_charger_v2.clear_override()
+            assert "Feature not supported for older firmware." in caplog.text
+
+
+async def test_get_override(test_charger, test_charger_v2, mock_aioclient, caplog):
+    """Test get override function."""
+    await test_charger.update()
+    value = {
+        "state": "active",
+        "charge_current": 0,
+        "max_current": 0,
+        "energy_limit": 0,
+        "time_limit": 0,
+        "auto_release": True,
+    }
+    mock_aioclient.get(
+        TEST_URL_OVERRIDE,
+        status=200,
+        body=json.dumps(value),
+    )
+    with caplog.at_level(logging.DEBUG):
+        status = await test_charger.get_override()
+        assert status == value
+
+    with pytest.raises(UnsupportedFeature):
+        with caplog.at_level(logging.DEBUG):
+            await test_charger_v2.update()
+            await test_charger_v2.get_override()
+            assert "Feature not supported for older firmware." in caplog.text