Allow to enter user pin to start and stop a scene on hc3 (#25)

* Allow to send user pin for scene actions

* New version 0.7.3

pyfibaro/common/rest_client.py

@@ -38,23 +38,37 @@ def set_auth(self, username: str, password: str) -> None:
         self._session.auth = HTTPBasicAuth(username, password)
 
     def get(
-        self, endpoint: str, json: Any | None = None, timeout: int | None = None
+        self,
+        endpoint: str,
+        json: Any | None = None,
+        timeout: int | None = None,
+        http_headers: dict = None,
     ) -> Any:
         """Execute a get request."""
         current_timeout = timeout if timeout else DEFAULT_TIMEOUT
         response = self._session.get(
-            f"{self._base_url}{endpoint}", json=json, timeout=current_timeout
+            f"{self._base_url}{endpoint}",
+            json=json,
+            timeout=current_timeout,
+            headers=http_headers,
         )
 
         return self._process_json_result(response)
 
     def post(
-        self, endpoint: str, json: Any | None = None, timeout: int | None = None
+        self,
+        endpoint: str,
+        json: Any | None = None,
+        timeout: int | None = None,
+        http_headers: dict = None,
     ) -> Any:
         """Execute a post request."""
         current_timeout = timeout if timeout else DEFAULT_TIMEOUT
         response = self._session.post(
-            f"{self._base_url}{endpoint}", json=json, timeout=current_timeout
+            f"{self._base_url}{endpoint}",
+            json=json,
+            timeout=current_timeout,
+            headers=http_headers,
         )
 
         return self._process_json_result(response)

pyfibaro/fibaro_scene.py

@@ -12,7 +12,7 @@ class SceneModel:
     """Model of a scene."""
 
     def __init__(self, data: dict, rest_client: RestClient, api_version: int) -> None:
-        """One scne."""
+        """One scene."""
         self.raw_data = data
         self._rest_client = rest_client
         self._api_version = api_version
@@ -43,27 +43,34 @@ def visible(self) -> bool:
         else:
             return not self.raw_data.get("hidden", False)
 
-    def start(self) -> None:
+    def start(self, user_pin: str | None = None) -> None:
         """Start a scene."""
         if self._api_version == 4:
+            if user_pin:
+                raise NotImplementedError("Not supported on old fibaro hubs")
             self._send_action_v4("start")
         else:
-            self._send_action_v5("execute")
+            self._send_action_v5("execute", user_pin)
 
-    def stop(self) -> None:
+    def stop(self, user_pin: str | None = None) -> None:
         """Stop a scene."""
         if self._api_version == 4:
+            if user_pin:
+                raise NotImplementedError("Not supported on old fibaro hubs")
             self._send_action_v4("stop")
         else:
-            self._send_action_v5("kill")
+            self._send_action_v5("kill", user_pin)
 
     def _send_action_v4(self, action: str) -> None:
         url = f"scenes/{self.fibaro_id}/action/{action}"
         self._rest_client.post(url)
 
-    def _send_action_v5(self, action: str) -> None:
+    def _send_action_v5(self, action: str, user_pin: str | None) -> None:
         url = f"scenes/{self.fibaro_id}/{action}"
-        self._rest_client.post(url, {})
+        if user_pin:
+            self._rest_client.post(url, {}, http_headers={"Fibaro-User-PIN": user_pin})
+        else:
+            self._rest_client.post(url, {})
 
     @staticmethod
     def read_scenes(rest_client: RestClient, api_version: int) -> list[SceneModel]:

setup.cfg

@@ -1,6 +1,6 @@
 [metadata]
 name = pyfibaro
-version = 0.7.2
+version = 0.7.3
 description = Simple API to access fibaro home center from any Python 3 script. Designed for Home Assistant (but not only)
 long_description = file: README.md
 long_description_content_type = text/markdown

tests/test_fibaro_scene.py

@@ -1,5 +1,6 @@
 """Test SceneModel class."""
 
+import pytest
 import requests_mock
 
 from pyfibaro.fibaro_client import FibaroClient
@@ -55,6 +56,30 @@ def test_fibaro_scene_start() -> None:
         scenes[0].start()
         assert mock.call_count == 4
 
+
+def test_fibaro_scene_start_with_pin_raise_exception() -> None:
+    """Test get request"""
+    with requests_mock.Mocker() as mock:
+        assert isinstance(mock, requests_mock.Mocker)
+
+        mock.register_uri("GET", f"{TEST_BASE_URL}scenes", json=scene_payload)
+        mock.register_uri("POST", f"{TEST_BASE_URL}scenes/2/action/start")
+        mock.register_uri("GET", f"{TEST_BASE_URL}loginStatus", json=login_payload)
+        mock.register_uri("GET", f"{TEST_BASE_URL}settings/info", json=info_payload)
+
+        client = FibaroClient(TEST_BASE_URL)
+        client.set_authentication(TEST_USERNAME, TEST_PASSWORD)
+        client.connect()
+
+        scenes = client.read_scenes()
+
+        assert mock.call_count == 3
+
+        with pytest.raises(NotImplementedError):
+            scenes[0].start("1234")
+        assert mock.call_count == 3
+
+
 def test_fibaro_scene_stop() -> None:
     """Test get request"""
     with requests_mock.Mocker() as mock:
@@ -74,3 +99,25 @@ def test_fibaro_scene_stop() -> None:
         assert mock.call_count == 3
         scenes[0].stop()
         assert mock.call_count == 4
+
+
+def test_fibaro_scene_stop_with_pin_raises_exception() -> None:
+    """Test get request"""
+    with requests_mock.Mocker() as mock:
+        assert isinstance(mock, requests_mock.Mocker)
+
+        mock.register_uri("GET", f"{TEST_BASE_URL}scenes", json=scene_payload)
+        mock.register_uri("POST", f"{TEST_BASE_URL}scenes/2/action/stop")
+        mock.register_uri("GET", f"{TEST_BASE_URL}loginStatus", json=login_payload)
+        mock.register_uri("GET", f"{TEST_BASE_URL}settings/info", json=info_payload)
+
+        client = FibaroClient(TEST_BASE_URL)
+        client.set_authentication(TEST_USERNAME, TEST_PASSWORD)
+        client.connect()
+
+        scenes = client.read_scenes()
+
+        assert mock.call_count == 3
+        with pytest.raises(NotImplementedError):
+            scenes[0].stop("1234")
+        assert mock.call_count == 3

tests/test_fibaro_scene_hc3.py

@@ -35,6 +35,7 @@ def test_fibaro_scene() -> None:
         assert scenes[0].visible is True
         assert mock.call_count == 3
 
+
 def test_fibaro_scene_start() -> None:
     """Test get request"""
     with requests_mock.Mocker() as mock:
@@ -52,7 +53,7 @@ def test_fibaro_scene_start() -> None:
         scenes = client.read_scenes()
 
         assert mock.call_count == 3
-        scenes[0].start()
+        scenes[0].start("1234")
         assert mock.call_count == 4