Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
44 changes: 44 additions & 0 deletions src/demetriek/local_auth.py
Original file line number Diff line number Diff line change
Expand Up @@ -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.

Expand Down
54 changes: 54 additions & 0 deletions tests/test_local_auth.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Loading