From dd394346d01800a7f075b364a60755a3cacc5a90 Mon Sep 17 00:00:00 2001 From: Franck Nijhof Date: Sun, 4 Oct 2026 08:25:09 +0000 Subject: [PATCH] Add checking whether a device supports the button press Asking for a challenge shows it on the screen of the device right away, so it is no way to find out whether a device supports the flow. The new supported() only checks for the web interface of the device: an sa8 TIME serves it, an LM 37X8 TIME asks for credentials. Nothing shows on the screen. Verified on both. --- src/demetriek/local_auth.py | 44 ++++++++++++++++++++++++++++++ tests/test_local_auth.py | 54 +++++++++++++++++++++++++++++++++++++ 2 files changed, 98 insertions(+) diff --git a/src/demetriek/local_auth.py b/src/demetriek/local_auth.py index 184c6b69..693194d3 100644 --- a/src/demetriek/local_auth.py +++ b/src/demetriek/local_auth.py @@ -154,6 +154,50 @@ def _parse_challenge(self, data: Any) -> AuthChallenge: ) raise LaMetricError(msg) from exception + async def supported(self) -> bool: + """Tell whether the device can hand out its API key with a button press. + + This only checks for the web interface of the device, so unlike + `request_challenge()`, nothing shows on the screen of the device. + + Returns + ------- + True when the device has the web interface, like devices from 2022 + onward. An LM 37X8 TIME asks for credentials instead. + + Raises + ------ + LaMetricConnectionError: An error occurred while communicating with + the LaMetric device. + LaMetricConnectionTimeoutError: A timeout occurred while communicating + with the LaMetric device. + + """ + url = URL.build(scheme="https", host=self.host, path="/") + + if self.session is None: + self.session = aiohttp.ClientSession() + self._close_session = True + + try: + async with ( + asyncio.timeout(self.request_timeout), + self.session.get(url, ssl=False) as response, + ): + return response.status == HTTPStatus.OK + except TimeoutError as exception: + msg = ( + "Timeout occurred while connecting to the LaMetric device" + f" at {self.host}" + ) + raise LaMetricConnectionTimeoutError(msg) from exception + except (aiohttp.ClientError, socket.gaierror) as exception: + msg = ( + "Error occurred while communicating with the LaMetric device" + f" at {self.host}" + ) + raise LaMetricConnectionError(msg) from exception + async def request_challenge(self) -> AuthChallenge: """Ask the device to have its button pressed. diff --git a/tests/test_local_auth.py b/tests/test_local_auth.py index fca70a58..238a7577 100644 --- a/tests/test_local_auth.py +++ b/tests/test_local_auth.py @@ -253,3 +253,57 @@ async def test_internal_session(responses: aioresponses) -> None: assert challenge.resolved is True assert session is not None assert session.closed + + +@pytest.mark.parametrize( + ("status", "supported"), + [ + # A device from 2022 onward serves its web interface. + (200, True), + # An LM 37X8 TIME asks for credentials instead. + (401, False), + ], +) +async def test_supported( + responses: aioresponses, + auth: LaMetricLocalAuth, + status: int, + supported: bool, # noqa: FBT001 +) -> None: + """Test checking support, without asking the device for a button press.""" + responses.get(f"{WEB_URL}/", status=status, body="") + + assert await auth.supported() is supported + assert ("POST", URL(REQUEST_URL)) not in responses.requests + + +@pytest.mark.parametrize( + ("exception", "expected"), + [ + (TimeoutError(), LaMetricConnectionTimeoutError), + (aiohttp.ClientError(), LaMetricConnectionError), + ], +) +async def test_supported_connection_errors( + responses: aioresponses, + auth: LaMetricLocalAuth, + exception: Exception, + expected: type[Exception], +) -> None: + """Test connection problems while checking support raise connection errors.""" + responses.get(f"{WEB_URL}/", exception=exception) + + with pytest.raises(expected): + await auth.supported() + + +async def test_supported_internal_session(responses: aioresponses) -> None: + """Test checking support creates and closes its own session.""" + responses.get(f"{WEB_URL}/", status=200, body="") + + async with LaMetricLocalAuth(host="127.0.0.2") as auth: + assert await auth.supported() is True + session = auth.session + + assert session is not None + assert session.closed