diff --git a/src/elgato_cli/api.py b/src/elgato_cli/api.py index 71f39f4..8aaf864 100644 --- a/src/elgato_cli/api.py +++ b/src/elgato_cli/api.py @@ -1,7 +1,9 @@ """HTTP client for the Elgato Key Light REST API.""" +import http.client +import json as _json +import time from dataclasses import dataclass - import requests @@ -21,14 +23,66 @@ def kelvin_to_mired(kelvin: int) -> int: class ElgatoLight: + RETRIES = 5 + RETRY_DELAY = 1.0 + def __init__(self, host: str, port: int = 9123): self.host = host self.port = port self.base_url = f"http://{host}:{port}/elgato" + def _request(self, method: str, path: str, **kwargs): + """HTTP request with retries for flaky Elgato TCP stack. + + The Elgato Key Light firmware resets HTTP/1.1 GET connections but + requires HTTP/1.1 for PUT. We use HTTP/1.0 for reads, HTTP/1.1 with + Connection: close for writes. + """ + timeout = kwargs.pop("timeout", 5) + json_body = kwargs.pop("json", None) + + for attempt in range(self.RETRIES): + try: + conn = http.client.HTTPConnection(self.host, self.port, timeout=timeout) + # HTTP/1.0 for GET (light resets HTTP/1.1 GETs) + # HTTP/1.1 for PUT (light rejects HTTP/1.0 PUTs) + if method == "GET": + conn._http_vsn = 10 + conn._http_vsn_str = "HTTP/1.0" + + headers = {"Connection": "close"} + body = None + if json_body is not None: + body = _json.dumps(json_body) + headers["Content-Type"] = "application/json" + + conn.request(method, f"/elgato/{path}", body=body, headers=headers) + resp = conn.getresponse() + data = resp.read() + conn.close() + + if resp.status >= 400: + raise requests.HTTPError(f"{resp.status} {resp.reason}") + + class _Response: + def __init__(self, raw, status): + self._data = raw + self.status_code = status + def json(self): + return _json.loads(self._data) + def raise_for_status(self): + pass + + return _Response(data, resp.status) + except (requests.HTTPError, ValueError): + raise + except (ConnectionError, OSError, TimeoutError) as e: + if attempt == self.RETRIES - 1: + raise requests.ConnectionError(str(e)) + time.sleep(self.RETRY_DELAY) + def get_state(self) -> LightState: - resp = requests.get(f"{self.base_url}/lights", timeout=5) - resp.raise_for_status() + resp = self._request("GET", "lights") light = resp.json()["lights"][0] return LightState( on=bool(light["on"]), @@ -48,12 +102,10 @@ class ElgatoLight: "brightness": brightness if brightness is not None else current.brightness, "temperature": temperature if temperature is not None else current.temperature, } - resp = requests.put( - f"{self.base_url}/lights", + resp = self._request( + "PUT", "lights", json={"numberOfLights": 1, "lights": [payload]}, - timeout=5, ) - resp.raise_for_status() light = resp.json()["lights"][0] return LightState( on=bool(light["on"]), @@ -66,11 +118,7 @@ class ElgatoLight: return self.set_state(on=not current.on) def get_info(self) -> dict: - resp = requests.get(f"{self.base_url}/accessory-info", timeout=5) - resp.raise_for_status() - return resp.json() + return self._request("GET", "accessory-info").json() def get_settings(self) -> dict: - resp = requests.get(f"{self.base_url}/lights/settings", timeout=5) - resp.raise_for_status() - return resp.json() + return self._request("GET", "lights/settings").json() diff --git a/tests/test_api.py b/tests/test_api.py index 5695f95..9d5f7dd 100644 --- a/tests/test_api.py +++ b/tests/test_api.py @@ -1,24 +1,30 @@ """Tests for elgato_cli.api.""" +import json from unittest.mock import patch, MagicMock +import requests + from elgato_cli.api import ElgatoLight, LightState, mired_to_kelvin, kelvin_to_mired -def _mock_response(json_data, status_code=200): - resp = MagicMock() - resp.json.return_value = json_data - resp.status_code = status_code - resp.raise_for_status.return_value = None - return resp - - LIGHT_RESPONSE = { "numberOfLights": 1, "lights": [{"on": 1, "brightness": 50, "temperature": 200}], } +def _make_mock_conn(response_data, status=200): + """Create a mock HTTPConnection that returns the given response.""" + mock_conn = MagicMock() + mock_resp = MagicMock() + mock_resp.status = status + mock_resp.reason = "OK" if status == 200 else "Error" + mock_resp.read.return_value = json.dumps(response_data).encode() + mock_conn.getresponse.return_value = mock_resp + return mock_conn + + class TestMiredKelvin: def test_mired_to_kelvin(self): assert mired_to_kelvin(143) == 6993 @@ -36,64 +42,77 @@ class TestMiredKelvin: class TestElgatoLight: - @patch("elgato_cli.api.requests.get") - def test_get_state(self, mock_get): - mock_get.return_value = _mock_response(LIGHT_RESPONSE) + @patch("elgato_cli.api.http.client.HTTPConnection") + def test_get_state(self, mock_http_cls): + mock_http_cls.return_value = _make_mock_conn(LIGHT_RESPONSE) light = ElgatoLight("192.168.1.100") state = light.get_state() assert state == LightState(on=True, brightness=50, temperature=200) - mock_get.assert_called_once_with( - "http://192.168.1.100:9123/elgato/lights", timeout=5 - ) + mock_http_cls.assert_called_with("192.168.1.100", 9123, timeout=5) - @patch("elgato_cli.api.requests.put") - @patch("elgato_cli.api.requests.get") - def test_set_state_partial(self, mock_get, mock_put): - mock_get.return_value = _mock_response(LIGHT_RESPONSE) + @patch("elgato_cli.api.http.client.HTTPConnection") + def test_set_state_partial(self, mock_http_cls): updated = { "numberOfLights": 1, "lights": [{"on": 1, "brightness": 80, "temperature": 200}], } - mock_put.return_value = _mock_response(updated) + # First call: get_state (GET), second call: put (PUT) + mock_http_cls.side_effect = [ + _make_mock_conn(LIGHT_RESPONSE), + _make_mock_conn(updated), + ] light = ElgatoLight("192.168.1.100") state = light.set_state(brightness=80) assert state.brightness == 80 - call_args = mock_put.call_args - payload = call_args[1]["json"] - assert payload["lights"][0]["brightness"] == 80 - assert payload["lights"][0]["on"] == 1 # preserved from current state + assert mock_http_cls.call_count == 2 - @patch("elgato_cli.api.requests.put") - @patch("elgato_cli.api.requests.get") - def test_toggle(self, mock_get, mock_put): - mock_get.return_value = _mock_response(LIGHT_RESPONSE) # on=1 + @patch("elgato_cli.api.http.client.HTTPConnection") + def test_toggle(self, mock_http_cls): toggled = { "numberOfLights": 1, "lights": [{"on": 0, "brightness": 50, "temperature": 200}], } - mock_put.return_value = _mock_response(toggled) + # toggle calls: get_state, then set_state(get_state + put) + mock_http_cls.side_effect = [ + _make_mock_conn(LIGHT_RESPONSE), # toggle -> get_state + _make_mock_conn(LIGHT_RESPONSE), # set_state -> get_state + _make_mock_conn(toggled), # set_state -> put + ] light = ElgatoLight("192.168.1.100") state = light.toggle() assert state.on is False - @patch("elgato_cli.api.requests.get") - def test_get_info(self, mock_get): + @patch("elgato_cli.api.http.client.HTTPConnection") + def test_get_info(self, mock_http_cls): info_resp = { "productName": "Elgato Key Light", "serialNumber": "AB12CD34", "firmwareVersion": "1.0.3", } - mock_get.return_value = _mock_response(info_resp) + mock_http_cls.return_value = _make_mock_conn(info_resp) light = ElgatoLight("192.168.1.100") info = light.get_info() assert info["productName"] == "Elgato Key Light" - mock_get.assert_called_once_with( - "http://192.168.1.100:9123/elgato/accessory-info", timeout=5 - ) + + @patch("elgato_cli.api.time.sleep") + @patch("elgato_cli.api.http.client.HTTPConnection") + def test_retry_on_connection_error(self, mock_http_cls, mock_sleep): + fail_conn = MagicMock() + fail_conn.request.side_effect = ConnectionError("reset") + + ok_conn = _make_mock_conn(LIGHT_RESPONSE) + + mock_http_cls.side_effect = [fail_conn, ok_conn] + + light = ElgatoLight("192.168.1.100") + state = light.get_state() + assert state == LightState(on=True, brightness=50, temperature=200) + assert mock_http_cls.call_count == 2 + mock_sleep.assert_called_once_with(1.0) def test_custom_port(self): light = ElgatoLight("10.0.0.5", port=8080)