feat: add claims functions (#320)

* feat: add claims functions

* add python 3.12 to tests

* update tox config

* formatting

* update tests

* formatting

* more tests

* fix error message

Author: firstof9

.github/workflows/test.yml

@@ -20,6 +20,7 @@ jobs:
         python-version:
           - "3.10"
           - "3.11"
+          - "3.12"
 
     steps:
       - uses: actions/[email protected]

openevsehttp/__main__.py

@@ -17,6 +17,7 @@
 from .const import (
     BAT_LVL,
     BAT_RANGE,
+    CLIENT,
     GRID,
     MAX_AMPS,
     MIN_AMPS,
@@ -757,6 +758,67 @@ async def get_limit(self) -> Any:
         )  # noqa: E501
         return response
 
+    async def make_claim(
+        self,
+        state: str | None = None,
+        charge_current: int | None = None,
+        max_current: int | None = None,
+        auto_release: bool = True,
+        client: int = CLIENT,
+    ) -> Any:
+        """Make a claim."""
+        if not self._version_check("4.1.0"):
+            _LOGGER.debug("Feature not supported for older firmware.")
+            raise UnsupportedFeature
+
+        if state not in ["active", "disabled", None]:
+            _LOGGER.error("Invalid claim state: %s", state)
+            raise ValueError
+
+        url = f"{self.url}claims/{client}"
+
+        data: dict[str, Any] = {}
+
+        data["auto_release"] = auto_release
+
+        if state is not None:
+            data["state"] = state
+        if charge_current is not None:
+            data["charge_current"] = charge_current
+        if max_current is not None:
+            data["max_current"] = max_current
+
+        _LOGGER.debug("Claim data: %s", data)
+        _LOGGER.debug("Setting up claim on %s", url)
+        response = await self.process_request(
+            url=url, method="post", data=data
+        )  # noqa: E501
+        return response
+
+    async def release_claim(self, client: int = CLIENT) -> Any:
+        """Delete a claim."""
+        if not self._version_check("4.1.0"):
+            _LOGGER.debug("Feature not supported for older firmware.")
+            raise UnsupportedFeature
+
+        url = f"{self.url}claims/{client}"
+
+        _LOGGER.debug("Releasing claim on %s", url)
+        response = await self.process_request(url=url, method="delete")  # noqa: E501
+        return response
+
+    async def list_claims(self) -> Any:
+        """List all claims."""
+        if not self._version_check("4.1.0"):
+            _LOGGER.debug("Feature not supported for older firmware.")
+            raise UnsupportedFeature
+
+        url = f"{self.url}claims"
+
+        _LOGGER.debug("Getting claims on %s", url)
+        response = await self.process_request(url=url, method="get")  # noqa: E501
+        return response
+
     @property
     def hostname(self) -> str:
         """Return charger hostname."""

openevsehttp/const.py

@@ -14,3 +14,4 @@
 TYPE = "type"
 VALUE = "value"
 RELEASE = "release"
+CLIENT = 4

setup.py

@@ -6,7 +6,7 @@
 
 PROJECT_DIR = Path(__file__).parent.resolve()
 README_FILE = PROJECT_DIR / "README.md"
-VERSION = "0.1.58"
+VERSION = "0.1.59"
 
 setup(
     name="python-openevse-http",
@@ -31,5 +31,6 @@
         "Programming Language :: Python :: 3",
         "Programming Language :: Python :: 3.10",
         "Programming Language :: Python :: 3.11",
+        "Programming Language :: Python :: 3.12",
     ],
 )

tests/test_main.py

@@ -30,6 +30,7 @@
 TEST_URL_RESTART = "http://openevse.test.tld/restart"
 TEST_URL_LIMIT = "http://openevse.test.tld/limit"
 TEST_URL_WS = "ws://openevse.test.tld/ws"
+TEST_URL_CLAIMS = "http://openevse.test.tld/claims"
 TEST_URL_GITHUB_v4 = (
     "https://api.github.com/repos/OpenEVSE/ESP32_WiFi_V4.x/releases/latest"
 )
@@ -1682,3 +1683,71 @@ async def test_voltage(test_charger, test_charger_v2, mock_aioclient, caplog):
         with caplog.at_level(logging.DEBUG):
             await test_charger_v2.grid_voltage(210)
             assert "Feature not supported for older firmware." in caplog.text
+
+
+async def test_list_claims(test_charger, test_charger_v2, mock_aioclient, caplog):
+    """Test list_claims function."""
+    await test_charger.update()
+    mock_aioclient.get(
+        TEST_URL_CLAIMS,
+        status=200,
+        body='[{"client":65540,"priority":10,"state":"disabled","auto_release":false}]',
+        repeat=True,
+    )
+    with caplog.at_level(logging.DEBUG):
+        await test_charger.list_claims()
+        assert f"Getting claims on {TEST_URL_CLAIMS}" in caplog.text
+
+    with pytest.raises(UnsupportedFeature):
+        with caplog.at_level(logging.DEBUG):
+            await test_charger_v2.list_claims()
+            assert "Feature not supported for older firmware." in caplog.text
+
+
+async def test_release_claim(test_charger, test_charger_v2, mock_aioclient, caplog):
+    """Test release_claim function."""
+    await test_charger.update()
+    mock_aioclient.delete(
+        f"{TEST_URL_CLAIMS}/4",
+        status=200,
+        body='[{"msg":"done"}]',
+        repeat=True,
+    )
+    with caplog.at_level(logging.DEBUG):
+        await test_charger.release_claim()
+        assert f"Releasing claim on {TEST_URL_CLAIMS}/4" in caplog.text
+
+    with pytest.raises(UnsupportedFeature):
+        with caplog.at_level(logging.DEBUG):
+            await test_charger_v2.release_claim()
+            assert "Feature not supported for older firmware." in caplog.text
+
+
+async def test_make_claim(test_charger, test_charger_v2, mock_aioclient, caplog):
+    """Test make_claim function."""
+    await test_charger.update()
+    mock_aioclient.post(
+        f"{TEST_URL_CLAIMS}/4",
+        status=200,
+        body='[{"msg":"done"}]',
+        repeat=True,
+    )
+    with caplog.at_level(logging.DEBUG):
+        await test_charger.make_claim(
+            state="disabled", charge_current=20, max_current=20
+        )
+        assert (
+            "Claim data: {'auto_release': True, 'state': 'disabled', 'charge_current': 20, 'max_current': 20}"
+            in caplog.text
+        )
+        assert f"Setting up claim on {TEST_URL_CLAIMS}/4" in caplog.text
+
+    with pytest.raises(ValueError):
+        with caplog.at_level(logging.DEBUG):
+            await test_charger.make_claim("invalid")
+            assert "Invalid claim state: invalid" in caplog.text
+
+    with pytest.raises(UnsupportedFeature):
+        with caplog.at_level(logging.DEBUG):
+            await test_charger_v2.make_claim()
+            assert "Feature not supported for older firmware." in caplog.text

tox.ini

@@ -1,11 +1,12 @@
 [tox]
-envlist = py310, py311, lint, mypy
+envlist = py310, py311, py312, lint, mypy
 skip_missing_interpreters = True
 
 [gh-actions]
 python =
-  3.10: py310, lint, mypy
+  3.10: py310
   3.11: py311
+  3.12: py312, lint, mypy
 
 [testenv]
 commands =