diff --git a/openevsehttp/commands.py b/openevsehttp/commands.py index c3e6f00..4ea7209 100644 --- a/openevsehttp/commands.py +++ b/openevsehttp/commands.py @@ -491,3 +491,35 @@ async def set_divert_mode(self, mode: str = "fast") -> None: if not success: _LOGGER.error("Problem issuing command: %s", response) raise UnknownError + + async def set_shaper(self, enable: bool = True) -> None: + """Set shaper mode.""" + if not self._version_check("4.0.0"): + _LOGGER.debug("Feature not supported for older firmware.") + raise UnsupportedFeature + + url = f"{self.url}shaper" + mode = 1 if enable else 0 + data = {"mode": mode} + + _LOGGER.debug("Setting shaper to %s", mode) + response = await self.process_request(url=url, method="post", data=data) + response = self._normalize_response(response) + msg = response.get("msg") if isinstance(response, Mapping) else None + if msg not in ["OK", "done", "no change"]: + _LOGGER.error("Problem issuing command: %s", response) + raise UnknownError + + async def toggle_shaper(self) -> None: + """Toggle shaper mode.""" + shaper_active = self._status.get("shaper") + if shaper_active is None: + await self.update() + shaper_active = self._status.get("shaper") + + if shaper_active is None: + _LOGGER.error("Cannot toggle shaper: unknown shaper state.") + raise RuntimeError("Cannot toggle shaper: unknown shaper state.") + + new_state = not bool(shaper_active) + await self.set_shaper(new_state) diff --git a/setup.py b/setup.py index 7a9aad8..3a790db 100644 --- a/setup.py +++ b/setup.py @@ -6,7 +6,7 @@ PROJECT_DIR = Path(__file__).parent.resolve() README_FILE = PROJECT_DIR / "README.md" -VERSION = "0.3.0" +VERSION = "0.3.1" setup( name="python_openevse_http", diff --git a/tests/test_shaper.py b/tests/test_shaper.py new file mode 100644 index 0000000..5591777 --- /dev/null +++ b/tests/test_shaper.py @@ -0,0 +1,128 @@ +"""Tests for shaper command methods.""" + +import logging + +import pytest + +from openevsehttp.exceptions import UnknownError, UnsupportedFeature + +pytestmark = pytest.mark.asyncio + +TEST_URL_SHAPER = "http://openevse.test.tld/shaper" + + +async def test_set_shaper(test_charger, test_charger_v2, mock_aioclient, caplog): + """Test set_shaper command.""" + await test_charger.update() + mock_aioclient.post( + TEST_URL_SHAPER, + status=200, + body='{"msg": "OK"}', + repeat=True, + ) + with caplog.at_level(logging.DEBUG): + await test_charger.set_shaper(True) + assert "Setting shaper to 1" in caplog.text + + await test_charger.set_shaper(False) + assert "Setting shaper to 0" in caplog.text + + await test_charger_v2.update() + # Force version lower than 4.0.0 + test_charger_v2._config["version"] = "3.3.1" + with pytest.raises(UnsupportedFeature): + await test_charger_v2.set_shaper(True) + + +async def test_set_shaper_fail(test_charger, mock_aioclient, caplog): + """Test set_shaper failure.""" + await test_charger.update() + mock_aioclient.post( + TEST_URL_SHAPER, + status=200, + body='{"msg": "failure!"}', + ) + with pytest.raises(UnknownError): + await test_charger.set_shaper(True) + + +async def test_toggle_shaper(test_charger, mock_aioclient, caplog): + """Test toggle_shaper command.""" + await test_charger.update() + # Initial state from status fixture is likely True or False + # Let's force it to 0 (off) + test_charger._status["shaper"] = 0 + + mock_aioclient.post( + TEST_URL_SHAPER, + status=200, + body='{"msg": "OK"}', + ) + + with caplog.at_level(logging.DEBUG): + await test_charger.toggle_shaper() + assert "Setting shaper to 1" in caplog.text + + # Now it's on (1) + test_charger._status["shaper"] = 1 + mock_aioclient.post( + TEST_URL_SHAPER, + status=200, + body='{"msg": "OK"}', + ) + with caplog.at_level(logging.DEBUG): + await test_charger.toggle_shaper() + assert "Setting shaper to 0" in caplog.text + + +async def test_toggle_shaper_missing_state(test_charger, mock_aioclient, caplog): + """Test toggle_shaper when state is missing.""" + # Clear status to force update() + test_charger._status = {} + + # Mock the /status call that update() will make + from tests.common import load_fixture + + TEST_URL_STATUS = "http://openevse.test.tld/status" + mock_aioclient.get( + TEST_URL_STATUS, + status=200, + body=load_fixture("v4_json/status.json"), + ) + + mock_aioclient.post( + TEST_URL_SHAPER, + status=200, + body='{"msg": "OK"}', + ) + + with caplog.at_level(logging.DEBUG): + await test_charger.toggle_shaper() + # status.json has shaper: 1, so it should toggle to 0 + assert "Setting shaper to 0" in caplog.text + + +async def test_toggle_shaper_failed_update(mock_aioclient, caplog): + """Test toggle_shaper when state is still missing after update().""" + from openevsehttp import OpenEVSE + + charger = OpenEVSE("openevse.test.tld") + + # Mock the /status call but return status without shaper + mock_aioclient.get( + "http://openevse.test.tld/status", + status=200, + body='{"mode": 1}', # No shaper key + ) + mock_aioclient.get( + "http://openevse.test.tld/config", + status=200, + body='{"firmware": "4.1.2"}', + ) + + with pytest.raises( + RuntimeError, match="Cannot toggle shaper: unknown shaper state." + ): + await charger.toggle_shaper() + + assert "Cannot toggle shaper: unknown shaper state." in caplog.text