diff --git a/pyfibaro/common/rest_client.py b/pyfibaro/common/rest_client.py index 11bf3c3..724e39b 100644 --- a/pyfibaro/common/rest_client.py +++ b/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) diff --git a/pyfibaro/fibaro_scene.py b/pyfibaro/fibaro_scene.py index 39a3634..5df700c 100644 --- a/pyfibaro/fibaro_scene.py +++ b/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]: diff --git a/setup.cfg b/setup.cfg index 399cf5e..4618066 100644 --- a/setup.cfg +++ b/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 diff --git a/tests/test_fibaro_scene.py b/tests/test_fibaro_scene.py index 3397302..ef6a03d 100644 --- a/tests/test_fibaro_scene.py +++ b/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 diff --git a/tests/test_fibaro_scene_hc3.py b/tests/test_fibaro_scene_hc3.py index fe41513..545f5f8 100644 --- a/tests/test_fibaro_scene_hc3.py +++ b/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