diff --git a/.github/workflows/linting.yml b/.github/workflows/linting.yml new file mode 100644 index 0000000..e13ff9f --- /dev/null +++ b/.github/workflows/linting.yml @@ -0,0 +1,28 @@ +name: Linting + +on: + push: + branches: [main] + pull_request: + +jobs: + ruff: + runs-on: ubuntu-latest + + steps: + - name: Checkout code + uses: actions/checkout@v4 + + - name: Set up Python + uses: actions/setup-python@v5 + with: + python-version: "3.11" + + - name: Install dependencies + run: | + python -m pip install --upgrade pip + pip install -e . + pip install ruff + + - name: Run ruff + run: ruff check src diff --git a/.github/workflows/tests.yml b/.github/workflows/tests.yml new file mode 100644 index 0000000..955ca72 --- /dev/null +++ b/.github/workflows/tests.yml @@ -0,0 +1,39 @@ +name: Tests + +on: + push: + branches: [main] + pull_request: + +jobs: + pytest: + runs-on: ubuntu-latest + + strategy: + matrix: + python-version: ["3.10", "3.11", "3.12", "3.13", "3.14"] + + steps: + - name: Checkout code + uses: actions/checkout@v4 + + - name: Set up Python + uses: actions/setup-python@v5 + with: + python-version: ${{ matrix.python-version }} + + - name: Install dependencies + run: | + python -m pip install --upgrade pip + pip install -e . + pip install pytest pytest-asyncio pytest-cov aioresponses + + - name: Run tests with coverage + run: | + pytest + + - name: Upload coverage artifact + uses: actions/upload-artifact@v4 + with: + name: coverage-${{ matrix.python-version }} + path: coverage.xml diff --git a/.github/workflows/typing.yml b/.github/workflows/typing.yml new file mode 100644 index 0000000..22176a7 --- /dev/null +++ b/.github/workflows/typing.yml @@ -0,0 +1,28 @@ +name: Typing + +on: + push: + branches: [main] + pull_request: + +jobs: + mypy: + runs-on: ubuntu-latest + + steps: + - name: Checkout code + uses: actions/checkout@v4 + + - name: Set up Python + uses: actions/setup-python@v5 + with: + python-version: "3.11" + + - name: Install dependencies + run: | + python -m pip install --upgrade pip + pip install -e . + pip install mypy + + - name: Run mypy + run: mypy src/hdfury diff --git a/.gitignore b/.gitignore index d40fcc1..701b503 100644 --- a/.gitignore +++ b/.gitignore @@ -4,3 +4,7 @@ # General files *~ *.DS_STORE + +# pytest +.coverage +coverage.xml diff --git a/README.md b/README.md index 7b4851f..b11c1da 100644 --- a/README.md +++ b/README.md @@ -143,7 +143,7 @@ print(TX0_INPUT_PORTS["1"]) # Output: Input 1 | `"1"` | Input 1 | | `"2"` | Input 2 | | `"3"` | Input 3 | -| `"4"` | Copy TX0 | +| `"4"` | Copy TX1 | ### `TX1_INPUT_PORTS` @@ -161,7 +161,7 @@ print(TX1_INPUT_PORTS["3"]) # Output: Input 3 | `"1"` | Input 1 | | `"2"` | Input 2 | | `"3"` | Input 3 | -| `"4"` | Copy TX1 | +| `"4"` | Copy TX0 | ## License diff --git a/pyproject.toml b/pyproject.toml index b27e108..3620bd5 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -7,7 +7,7 @@ authors = [ ] license = { text = "MIT" } readme = "README.md" -requires-python = ">=3.11" +requires-python = ">=3.10" dependencies = [ "aiohttp>=3.0.0", ] @@ -19,3 +19,40 @@ Issues = "https://github.com/glenndehaan/python-hdfury/issues" [build-system] requires = ["hatchling"] build-backend = "hatchling.build" + +[tool.hatch.build.targets.wheel] +packages = ["src/hdfury"] + +[tool.pytest.ini_options] +asyncio_mode = "auto" +testpaths = ["tests"] +addopts = [ + "--strict-markers", + "--cov=hdfury", + "--cov-report=term-missing", + "--cov-report=xml", +] + +[tool.ruff] +line-length = 120 +exclude = ["build", ".venv"] + +[tool.ruff.lint] +select = ["E", "F", "W", "I"] +extend-select = ["B", "D"] +ignore = [ + "D203", # incompatible with D211 + "D213", # incompatible with D212 +] +extend-ignore = [] + +[tool.mypy] +python_version = 3.11 +files = "src,hdfury" +ignore_missing_imports = true +strict = true +disallow_untyped_defs = true +warn_unused_ignores = true +warn_return_any = true +pretty = true +show_error_codes = true diff --git a/src/hdfury/__init__.py b/src/hdfury/__init__.py index 0354a62..43501cd 100644 --- a/src/hdfury/__init__.py +++ b/src/hdfury/__init__.py @@ -2,6 +2,14 @@ from .api import HDFuryAPI from .const import OPERATION_MODES, TX0_INPUT_PORTS, TX1_INPUT_PORTS -from .exceptions import HDFuryError +from .exceptions import HDFuryConnectionError, HDFuryError, HDFuryParseError -__all__ = ["OPERATION_MODES", "TX0_INPUT_PORTS", "TX1_INPUT_PORTS", "HDFuryAPI", "HDFuryError"] +__all__ = [ + "OPERATION_MODES", + "TX0_INPUT_PORTS", + "TX1_INPUT_PORTS", + "HDFuryAPI", + "HDFuryConnectionError", + "HDFuryError", + "HDFuryParseError", +] diff --git a/src/hdfury/api.py b/src/hdfury/api.py index 6d78ac7..7b7670f 100644 --- a/src/hdfury/api.py +++ b/src/hdfury/api.py @@ -1,31 +1,30 @@ """HDFury Client API.""" import asyncio -from asyncio import TimeoutError import json import time -from typing import Literal +from asyncio import TimeoutError +from typing import Literal, cast import aiohttp -from aiohttp import ClientError, ClientResponseError +from aiohttp import ClientError, ClientResponseError, ClientTimeout from .exceptions import HDFuryConnectionError, HDFuryParseError -StateStr = Literal["0", "1", "off", "on"] - class HDFuryAPI: """Asynchronous API client for HDFury devices.""" def __init__(self, host: str, session: aiohttp.ClientSession | None = None) -> None: """HDFury API Client.""" - - self.host = host - self._session = session or aiohttp.ClientSession() - self._last_command_time = 0 - self._debounce_delay = 2 # seconds - - def _normalize_state(self, state: StateStr, *, output: Literal["text", "number"] = "text") -> str: + self.host: str = host + self._session: aiohttp.ClientSession = session or aiohttp.ClientSession() + self._last_command_time: float = 0 + self._debounce_delay: int = 2 # seconds + + @staticmethod + def _normalize_state(state: str, output: Literal["text", "number"] = "text") -> str: + """Normalize state and optionally convert return type.""" state = state.lower() if state in ("1", "on"): @@ -36,7 +35,7 @@ def _normalize_state(self, state: StateStr, *, output: Literal["text", "number"] raise HDFuryParseError(f"Invalid state: {state}") async def _wait_for_debounce(self) -> None: - """Helper to ensure at least `_debounce_delay` seconds have passed since last command.""" + """Wait until at least `_debounce_delay` seconds have passed since the last command.""" elapsed = time.time() - self._last_command_time if elapsed < self._debounce_delay: wait_time = self._debounce_delay - elapsed @@ -47,41 +46,41 @@ async def _request(self, endpoint: str) -> str: url = f"http://{self.host}{endpoint}" try: - async with self._session.get(url, timeout=10) as response: + async with self._session.get(url, timeout=ClientTimeout(total=10)) as response: if response.status != 200: raise HDFuryConnectionError( f"Unexpected response from: {url} (Status: {response.status})" ) return await response.text() - except TimeoutError: - raise HDFuryConnectionError(f"Timeout while fetching: {url}") + except TimeoutError as err: + raise HDFuryConnectionError(f"Timeout while fetching: {url}") from err except (ClientError, ClientResponseError) as err: raise HDFuryConnectionError(f"Request failed ({url}): {err}") from err except Exception as err: raise HDFuryConnectionError(f"Unexpected error ({url}): {err}") from err - async def _request_json(self, path: str) -> dict: + async def _request_json(self, path: str) -> dict[str, str]: """Handle a request to the HDFury device and parse JSON.""" response = await self._request(path) try: - return json.loads(response) + return cast(dict[str, str], json.loads(response)) except json.JSONDecodeError as err: raise HDFuryParseError(f"Unable to decode JSON: {err}") from err - async def get_board(self) -> dict: + async def get_board(self) -> dict[str, str]: """Fetch board info.""" await self._wait_for_debounce() response = await self._request_json("/ssi/brdinfo.ssi") return response - async def get_info(self) -> dict: + async def get_info(self) -> dict[str, str]: """Fetch device info.""" await self._wait_for_debounce() response = await self._request_json("/ssi/infopage.ssi") return response - async def get_config(self) -> dict: + async def get_config(self) -> dict[str, str]: """Fetch device configuration.""" await self._wait_for_debounce() config_response = await self._request_json("/ssi/confpage.ssi") diff --git a/tests/__init__.py b/tests/__init__.py new file mode 100644 index 0000000..e523156 --- /dev/null +++ b/tests/__init__.py @@ -0,0 +1 @@ +"""Tests for the HDFury library.""" diff --git a/tests/test_api.py b/tests/test_api.py new file mode 100644 index 0000000..153b990 --- /dev/null +++ b/tests/test_api.py @@ -0,0 +1,364 @@ +"""Tests for the HDFury api.""" + +import time +from asyncio import TimeoutError +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest +from aiohttp import ClientError +from aioresponses import aioresponses + +from hdfury.api import HDFuryAPI +from hdfury.exceptions import HDFuryConnectionError, HDFuryParseError + +# --------------------------------------------------------------------------- +# Fixtures +# --------------------------------------------------------------------------- + +@pytest.fixture +async def client(): + """HDFury client fixture.""" + api = HDFuryAPI("192.168.1.123") + # Disable debounce delay to keep tests fast and deterministic + api._debounce_delay = 0 + yield api + await api.close() + +# --------------------------------------------------------------------------- +# _normalize_state tests +# --------------------------------------------------------------------------- + +def test_normalize_state_text_on(): + """Verify that text-based 'on' states are normalized correctly.""" + api = HDFuryAPI("x", session=MagicMock()) + assert api._normalize_state("1") == "on" + assert api._normalize_state("on") == "on" + assert api._normalize_state("ON") == "on" + +def test_normalize_state_text_off(): + """Verify that text-based 'off' states are normalized correctly.""" + api = HDFuryAPI("x", session=MagicMock()) + assert api._normalize_state("0") == "off" + assert api._normalize_state("off") == "off" + assert api._normalize_state("OFF") == "off" + +def test_normalize_state_number_output(): + """Verify that states are correctly converted to numeric output.""" + api = HDFuryAPI("x", session=MagicMock()) + assert api._normalize_state("on", output="number") == "1" + assert api._normalize_state("1", output="number") == "1" + assert api._normalize_state("off", output="number") == "0" + assert api._normalize_state("0", output="number") == "0" + +@pytest.mark.parametrize("bad", ["yes", "no", "", "2"]) +def test_normalize_state_invalid_raises(bad): + """Verify that invalid state values raise a HDFuryParseError.""" + api = HDFuryAPI("x", session=MagicMock()) + with pytest.raises(HDFuryParseError): + api._normalize_state(bad) # type: ignore[arg-type] + +# --------------------------------------------------------------------------- +# _request and _request_json tests +# --------------------------------------------------------------------------- + +@pytest.mark.asyncio +async def test_request_json_success(client: HDFuryAPI): + """Verify _request_json returns parsed JSON on success.""" + with aioresponses() as mock: + mock.get("http://192.168.1.123/test", body='{"a": 1}') + + result = await client._request_json("/test") + assert result == {"a": 1} + +@pytest.mark.asyncio +async def test_request_json_invalid_json_raises(client: HDFuryAPI): + """Verify _request_json raises HDFuryParseError for invalid JSON.""" + with aioresponses() as mock: + mock.get("http://192.168.1.123/test", body="not-json") + + with pytest.raises(HDFuryParseError): + await client._request_json("/test") + +@pytest.mark.asyncio +async def test_request_non_200_raises_connection_error(client: HDFuryAPI): + """Verify _request raises HDFuryConnectionError on non-200 responses.""" + with aioresponses() as mock: + mock.get("http://192.168.1.123/test", status=500) + + with pytest.raises(HDFuryConnectionError) as exc: + await client._request("/test") + + assert "Unexpected response from" in str(exc.value) + +@pytest.mark.asyncio +async def test_request_timeout(client): + """Verify that _request raises HDFuryConnectionError on a timeout.""" + with aioresponses() as mock: + mock.get("http://192.168.1.123/test", exception=TimeoutError()) + + with pytest.raises(HDFuryConnectionError) as exc: + await client._request("/test") + + assert "Timeout while fetching" in str(exc.value) + +@pytest.mark.asyncio +async def test_request_client_error(client): + """Verify that _request raises HDFuryConnectionError on aiohttp ClientError.""" + with aioresponses() as mock: + mock.get("http://192.168.1.123/test", exception=ClientError("Error")) + + with pytest.raises(HDFuryConnectionError) as exc: + await client._request("/test") + + assert "Request failed" in str(exc.value) + +# --------------------------------------------------------------------------- +# get_info / get_board basic behavior +# --------------------------------------------------------------------------- + +@pytest.mark.asyncio +async def test_get_info_success(client: HDFuryAPI): + """Verify get_info returns the correct info dictionary.""" + with aioresponses() as mock: + mock.get( + "http://192.168.1.123/ssi/infopage.ssi", + body='{"opmode": "0"}', + ) + + result = await client.get_info() + assert result == {"opmode": "0"} + +@pytest.mark.asyncio +async def test_get_board_success(client: HDFuryAPI): + """Verify get_board returns the correct board info dictionary.""" + with aioresponses() as mock: + mock.get( + "http://192.168.1.123/ssi/brdinfo.ssi", + body='{"hostname": "VRROOM-02"}', + ) + + result = await client.get_board() + assert result == {"hostname": "VRROOM-02"} + +# --------------------------------------------------------------------------- +# get_config merge behavior +# --------------------------------------------------------------------------- + +@pytest.mark.asyncio +async def test_get_config_merges_cec_and_config(client: HDFuryAPI): + """Verify get_config merges standard config and CEC config correctly.""" + with aioresponses() as mock: + mock.get( + "http://192.168.1.123/ssi/confpage.ssi", + body='{"autosw": "1", "oled": "1"}', + ) + mock.get( + "http://192.168.1.123/ssi/cecpage.ssi", + body='{"cec0en": "1"}', + ) + + result = await client.get_config() + assert result == { + "cec0en": "1", + "autosw": "1", + "oled": "1", + } + +@pytest.mark.asyncio +async def test_get_config_cec_failure_is_ignored(client: HDFuryAPI): + """Verify get_config ignores CEC page errors and returns available config.""" + with aioresponses() as mock: + mock.get( + "http://192.168.1.123/ssi/confpage.ssi", + body='{"autosw": "1"}', + ) + # cecpage returns 500 -> ignored + mock.get( + "http://192.168.1.123/ssi/cecpage.ssi", + status=500, + ) + + result = await client.get_config() + assert result == {"autosw": "1"} + +@pytest.mark.asyncio +async def test_get_config_cec_invalid_json_is_ignored(client: HDFuryAPI): + """Verify get_config ignores invalid JSON from CEC page.""" + with aioresponses() as mock: + mock.get( + "http://192.168.1.123/ssi/confpage.ssi", + body='{"autosw": "1"}', + ) + mock.get( + "http://192.168.1.123/ssi/cecpage.ssi", + body="not-json", + ) + + result = await client.get_config() + assert result == {"autosw": "1"} + +# --------------------------------------------------------------------------- +# Command and setter tests +# --------------------------------------------------------------------------- + +@pytest.mark.asyncio +async def test_issue_reboot(client: HDFuryAPI): + """Verify issue_reboot sends the correct command to the device.""" + with aioresponses() as mock: + mock.get("http://192.168.1.123/cmd?reboot=", status=200) + + await client.issue_reboot() + +@pytest.mark.asyncio +async def test_issue_hotplug(client: HDFuryAPI): + """Verify issue_hotplug sends the correct command to the device.""" + with aioresponses() as mock: + mock.get("http://192.168.1.123/cmd?hotplug=", status=200) + + await client.issue_hotplug() + +@pytest.mark.asyncio +async def test_set_operation_mode(client: HDFuryAPI): + """Verify set_operation_mode sends the correct operation mode command.""" + with aioresponses() as mock: + mock.get("http://192.168.1.123/cmd?opmode=3", status=200) + + await client.set_operation_mode("3") + +@pytest.mark.asyncio +async def test_set_port_selection(client: HDFuryAPI): + """Verify set_port_selection sends the correct input selection command.""" + with aioresponses() as mock: + # Note: space encoded as %20 in implementation + mock.get("http://192.168.1.123/cmd?insel=0%204", status=200) + + await client.set_port_selection("0", "4") + +@pytest.mark.asyncio +@pytest.mark.parametrize( + ("endpoint", "method", "value"), + [ + ("autosw", "set_auto_switch_inputs", "on"), + ("autosw", "set_auto_switch_inputs", "off"), + ], +) +async def test_set_auto_switch_inputs(client: HDFuryAPI, endpoint: str, method: str, value: str): + """Verify set_auto_switch_inputs sends the correct command for each state.""" + with aioresponses() as mock: + mock.get(f"http://192.168.1.123/cmd?{endpoint}={value}", status=200) + + await getattr(client, method)(value) + +@pytest.mark.asyncio +@pytest.mark.parametrize( + ("endpoint", "method", "value"), + [ + ("htpcmode0", "set_htpc_mode_rx0", "on"), + ("htpcmode1", "set_htpc_mode_rx1", "on"), + ("htpcmode2", "set_htpc_mode_rx2", "on"), + ("htpcmode3", "set_htpc_mode_rx3", "on"), + ("htpcmode0", "set_htpc_mode_rx0", "off"), + ("htpcmode1", "set_htpc_mode_rx1", "off"), + ("htpcmode2", "set_htpc_mode_rx2", "off"), + ("htpcmode3", "set_htpc_mode_rx3", "off"), + ], +) +async def test_set_htpc_mode(client: HDFuryAPI, endpoint: str, method: str, value: str): + """Verify set_htpc_mode commands are sent correctly for each input and state.""" + with aioresponses() as mock: + mock.get(f"http://192.168.1.123/cmd?{endpoint}={value}", status=200) + + await getattr(client, method)(value) + +@pytest.mark.asyncio +@pytest.mark.parametrize( + ("endpoint", "method", "value"), + [ + ("mutetx0audio", "set_mute_tx0_audio", "on"), + ("mutetx1audio", "set_mute_tx1_audio", "on"), + ("mutetx0audio", "set_mute_tx0_audio", "off"), + ("mutetx1audio", "set_mute_tx1_audio", "off"), + ], +) +async def test_set_mute_tx_audio(client: HDFuryAPI, endpoint: str, method: str, value: str): + """Verify set_mute_tx_audio commands are sent correctly for each transmitter and state.""" + with aioresponses() as mock: + mock.get(f"http://192.168.1.123/cmd?{endpoint}={value}", status=200) + + await getattr(client, method)(value) + +@pytest.mark.asyncio +@pytest.mark.parametrize( + ("endpoint", "method", "value"), + [ + ("oled", "set_oled", "on"), + ("oled", "set_oled", "off"), + ], +) +async def test_set_oled(client: HDFuryAPI, endpoint: str, method: str, value: str): + """Verify set_oled sends the correct command for each OLED state.""" + with aioresponses() as mock: + mock.get(f"http://192.168.1.123/cmd?{endpoint}={value}", status=200) + + await getattr(client, method)(value) + +@pytest.mark.asyncio +@pytest.mark.parametrize( + ("endpoint", "method", "value"), + [ + ("iractive", "set_ir_active", "on"), + ("iractive", "set_ir_active", "off"), + ], +) +async def test_set_ir_active(client: HDFuryAPI, endpoint: str, method: str, value: str): + """Verify set_ir_active sends the correct command for IR activation states.""" + with aioresponses() as mock: + mock.get(f"http://192.168.1.123/cmd?{endpoint}={value}", status=200) + + await getattr(client, method)(value) + +@pytest.mark.asyncio +@pytest.mark.parametrize( + ("endpoint", "method", "value"), + [ + ("relay", "set_relay", "on"), + ("relay", "set_relay", "off"), + ], +) +async def test_set_relay(client: HDFuryAPI, endpoint: str, method: str, value: str): + """Verify set_relay sends the correct command for relay states.""" + with aioresponses() as mock: + mock.get(f"http://192.168.1.123/cmd?{endpoint}={value}", status=200) + + await getattr(client, method)(value) + +@pytest.mark.asyncio +@pytest.mark.parametrize( + ("endpoint", "method", "value"), + [ + ("cec0en", "set_cec_rx0", "1"), + ("cec1en", "set_cec_rx1", "1"), + ("cec2en", "set_cec_rx2", "0"), + ("cec3en", "set_cec_rx3", "0"), + ], +) +async def test_set_cec_rx(client: HDFuryAPI, endpoint: str, method: str, value: str): + """Verify set_cec_rx commands are sent correctly for each CEC input.""" + with aioresponses() as mock: + mock.get(f"http://192.168.1.123/cmd?{endpoint}={value}", status=200) + + await getattr(client, method)(value) + +# --------------------------------------------------------------------------- +# Debounce behavior +# --------------------------------------------------------------------------- + +@pytest.mark.asyncio +async def test_wait_for_debounce_sleeps_when_called_too_fast(client: HDFuryAPI): + """Verify that _wait_for_debounce sleeps if commands are called too quickly.""" + client._debounce_delay = 2 + client._last_command_time = time.time() + + with patch("asyncio.sleep", new=AsyncMock()) as sleep_mock: + await client._wait_for_debounce() + sleep_mock.assert_called_once() diff --git a/tests/test_const.py b/tests/test_const.py new file mode 100644 index 0000000..3dbefb1 --- /dev/null +++ b/tests/test_const.py @@ -0,0 +1,50 @@ +"""Tests for the HDFury constants.""" + +from hdfury import OPERATION_MODES, TX0_INPUT_PORTS, TX1_INPUT_PORTS + + +def test_operation_modes_structure(): + """Ensure correct operation modes structure.""" + assert isinstance(OPERATION_MODES, dict) + assert len(OPERATION_MODES) > 0 + +def test_operation_modes_keys_are_numeric_strings(): + """Ensure correct operation mode keys.""" + for key in OPERATION_MODES.keys(): + assert isinstance(key, str) + assert key.isdigit() + +def test_operation_modes_values_are_descriptions(): + """Ensure correct operation mode values.""" + for value in OPERATION_MODES.values(): + assert isinstance(value, str) + assert "Mode" in value + +def test_tx0_input_ports_structure(): + """Ensure correct TX0 structure.""" + assert isinstance(TX0_INPUT_PORTS, dict) + assert len(TX0_INPUT_PORTS) > 0 + +def test_tx1_input_ports_structure(): + """Ensure correct TX1 structure.""" + assert isinstance(TX1_INPUT_PORTS, dict) + assert len(TX1_INPUT_PORTS) > 0 + +def test_tx_input_ports_keys_are_numeric_strings(): + """Ensure correct TX keys.""" + for ports in (TX0_INPUT_PORTS, TX1_INPUT_PORTS): + for key in ports.keys(): + assert isinstance(key, str) + assert key.isdigit() + +def test_tx_input_ports_values_are_descriptions(): + """Ensure correct TX values.""" + for ports in (TX0_INPUT_PORTS, TX1_INPUT_PORTS): + for value in ports.values(): + assert isinstance(value, str) + assert "Input" in value or "Copy" in value + +def test_tx_copy_ports_are_symmetric(): + """Ensure Copy TX0/TX1 are correctly mirrored.""" + assert TX0_INPUT_PORTS["4"] == "Copy TX1" + assert TX1_INPUT_PORTS["4"] == "Copy TX0"