From 542566270208c08c06b99fd1c24d873a56454992 Mon Sep 17 00:00:00 2001 From: Franck Nijhof Date: Sat, 3 Oct 2026 12:04:24 +0000 Subject: [PATCH] Add getting the API key from the device, with a press on its button Devices from 2022 onward can hand out their API key locally: request a challenge, have someone press the button on top of the device while it shows it, and exchange the resolved challenge for the key. No LaMetric account or cloud involved. This uses the web interface of the device, which LaMetric does not document. Tested end to end on an sa8 TIME on firmware 3.2.6. An LM 37X8 TIME does not have it, and now says so. The error helper also understands the error shape of that web interface, so the reason of the device is passed on. --- README.md | 37 +++ src/demetriek/__init__.py | 4 + src/demetriek/exceptions.py | 8 +- src/demetriek/local_auth.py | 280 ++++++++++++++++++ src/demetriek/models.py | 22 ++ tests/conftest.py | 9 +- .../fixtures/auth_challenge_resolved_sa8.json | 7 + tests/fixtures/auth_challenge_sa8.json | 9 + tests/fixtures/auth_exchange_sa8.json | 10 + tests/fixtures/auth_users_sa8.json | 26 ++ tests/test_lametric.py | 2 + tests/test_local_auth.py | 255 ++++++++++++++++ 12 files changed, 666 insertions(+), 3 deletions(-) create mode 100644 src/demetriek/local_auth.py create mode 100644 tests/fixtures/auth_challenge_resolved_sa8.json create mode 100644 tests/fixtures/auth_challenge_sa8.json create mode 100644 tests/fixtures/auth_exchange_sa8.json create mode 100644 tests/fixtures/auth_users_sa8.json create mode 100644 tests/test_local_auth.py diff --git a/README.md b/README.md index 1eb8a635..9a4c30e0 100644 --- a/README.md +++ b/README.md @@ -120,6 +120,43 @@ async def main(): print(f"- {device.name} at {device.ip}") +if __name__ == "__main__": + asyncio.run(main()) +``` + +### API key from the device + +Devices from 2022 onward can hand out their API key locally, after someone +presses the button on top of the device. No LaMetric account needed. An +LM 37X8 TIME does not support this. This uses the web interface of the +device, which LaMetric does not document. + +```python +"""Asynchronous Python client for LaMetric TIME devices.""" + +import asyncio + +from demetriek import LaMetricLocalAuth + + +async def main(): + """Get the API key of a LaMetric device with a press on its button.""" + async with LaMetricLocalAuth("192.168.1.2") as auth: + challenge = await auth.request_challenge() + print(f"Press the button on your LaMetric within {challenge.duration}s") + + while challenge.state == "in-progress": + await asyncio.sleep(1) + challenge = await auth.challenge(challenge_id=challenge.challenge_id) + + if not challenge.resolved: + print("The button was not pressed in time") + return + + api_key = await auth.api_key(challenge_id=challenge.challenge_id) + print(f"The API key is {api_key}") + + if __name__ == "__main__": asyncio.run(main()) ``` diff --git a/src/demetriek/__init__.py b/src/demetriek/__init__.py index d04fe676..af6a2ebe 100644 --- a/src/demetriek/__init__.py +++ b/src/demetriek/__init__.py @@ -26,11 +26,13 @@ LaMetricConnectionTimeoutError, LaMetricError, ) +from .local_auth import LaMetricLocalAuth from .models import ( API, App, AppParameter, Audio, + AuthChallenge, Bluetooth, BluetoothLowEnergy, Chart, @@ -69,6 +71,7 @@ "App", "AppParameter", "Audio", + "AuthChallenge", "Bluetooth", "BluetoothLowEnergy", "BrightnessMode", @@ -92,6 +95,7 @@ "LaMetricConnectionTimeoutError", "LaMetricDevice", "LaMetricError", + "LaMetricLocalAuth", "LaMetricStream", "Model", "Notification", diff --git a/src/demetriek/exceptions.py b/src/demetriek/exceptions.py index 39001872..05afa41d 100644 --- a/src/demetriek/exceptions.py +++ b/src/demetriek/exceptions.py @@ -25,7 +25,8 @@ def error_message(body: str) -> str | None: """Extract the error messages from an error response of the LaMetric API. Both the device and the cloud answer errors with a body like - `{"errors": [{"message": "..."}]}`. + `{"errors": [{"message": "..."}]}`. The web interface of the device + answers with a single `{"error": {"message": "..."}}` instead. Args: ---- @@ -37,7 +38,10 @@ def error_message(body: str) -> str | None: """ try: - errors = orjson.loads(body).get("errors") + data = orjson.loads(body) + errors = data.get("errors") + if isinstance(error := data.get("error"), dict): + errors = [error] except (ValueError, AttributeError): return None diff --git a/src/demetriek/local_auth.py b/src/demetriek/local_auth.py new file mode 100644 index 00000000..184c6b69 --- /dev/null +++ b/src/demetriek/local_auth.py @@ -0,0 +1,280 @@ +"""Asynchronous Python client for LaMetric TIME devices.""" + +from __future__ import annotations + +import asyncio +import socket +from dataclasses import dataclass +from http import HTTPStatus +from typing import Any, Self + +import aiohttp +from aiohttp import hdrs +from mashumaro.exceptions import MissingField +from yarl import URL + +from .exceptions import ( + LaMetricAuthenticationError, + LaMetricConnectionError, + LaMetricConnectionTimeoutError, + LaMetricError, + error_message, +) +from .models import AuthChallenge + + +@dataclass +class LaMetricLocalAuth: + """Get the API key of a LaMetric device, with a press on its button. + + This talks to the web interface of the device, which LaMetric does not + document. Devices from 2022 onward have it, an LM 37X8 TIME does not. + + Request a challenge, have someone press the button on top of the device + while it shows it, and exchange the resolved challenge for the API key. + No LaMetric account or cloud is involved. + """ + + host: str + request_timeout: float = 8.0 + session: aiohttp.client.ClientSession | None = None + + _close_session: bool = False + + async def _request( + self, + uri: str, + *, + method: str = hdrs.METH_GET, + query: dict[str, str] | None = None, + data: dict[str, Any] | None = None, + admin_key: str | None = None, + ) -> Any: + """Handle a request to the web interface of a LaMetric device. + + Args: + ---- + uri: Request URI, for example `/api/v1/user`. + method: HTTP method to use for the request. + query: Query parameters to add to the URI. + data: Dictionary of data to send to the device. + admin_key: The web admin key, for requests that need it. + + Returns: + ------- + A Python dictionary (JSON decoded) with the response. + + Raises: + ------ + LaMetricAuthenticationError: The web admin key was not accepted. + LaMetricConnectionError: An error occurred while communicating with + the LaMetric device. + LaMetricConnectionTimeoutError: A timeout occurred while communicating + with the LaMetric device. + LaMetricError: Received an unexpected response from the device. + + """ + url = URL.build(scheme="https", host=self.host, path=uri, query=query or {}) + + headers = {"Accept": "application/json"} + if admin_key is not None: + # The web interface takes the key from a cookie, not from basic auth. + headers["Cookie"] = f"no-auth-challenge=1; authorization={admin_key}:" + + if self.session is None: + self.session = aiohttp.ClientSession() + self._close_session = True + + try: + async with asyncio.timeout(self.request_timeout): + response = await self.session.request( + method, + url, + headers=headers, + json=data, + ssl=False, + ) + body = await response.text() + + if response.status >= HTTPStatus.BAD_REQUEST: + reason = error_message(body) or response.reason + if response.status == HTTPStatus.UNAUTHORIZED: + msg = ( + f"Authentication to the LaMetric device at {self.host}" + f" failed: {reason}" + ) + raise LaMetricAuthenticationError(msg) + msg = ( + f"The LaMetric device at {self.host} returned an error" + f" ({response.status}): {reason}" + ) + raise LaMetricError(msg) + + try: + return await response.json(content_type=None) + except ValueError as exception: + msg = f"The LaMetric device at {self.host} answered with invalid JSON" + raise LaMetricError(msg) from exception + + 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 + + def _parse_challenge(self, data: Any) -> AuthChallenge: + """Parse a challenge answered by the device. + + Args: + ---- + data: The JSON decoded challenge. + + Returns: + ------- + The challenge. + + Raises: + ------ + LaMetricError: The challenge does not look like one. + + """ + try: + return AuthChallenge.from_dict(data) + except (MissingField, ValueError) as exception: + msg = ( + f"The LaMetric device at {self.host} answered with data this" + f" library does not understand: {exception}" + ) + raise LaMetricError(msg) from exception + + async def request_challenge(self) -> AuthChallenge: + """Ask the device to have its button pressed. + + The device shows the challenge on its screen, someone then has + `duration` seconds to press the button on top of the device. + + Returns + ------- + The challenge, to poll with `challenge()`. + + Raises + ------ + LaMetricError: The device does not support this, for example an + LM 37X8 TIME. + + """ + try: + response = await self._request( + "/api/v1/user/request", + method=hdrs.METH_POST, + query={"group": "web_admin"}, + ) + except LaMetricAuthenticationError as exception: + # A device without this flow asks for credentials instead. + msg = ( + f"The LaMetric device at {self.host} does not support getting" + " its API key with a press on its button" + ) + raise LaMetricError(msg) from exception + + challenge = response.get("challenge") if isinstance(response, dict) else None + return self._parse_challenge(challenge) + + async def challenge(self, *, challenge_id: str) -> AuthChallenge: + """Get the current state of a challenge. + + Args: + ---- + challenge_id: ID of the challenge, from `request_challenge()`. + + Returns: + ------- + The challenge, which is resolved once the button was pressed. + + """ + response = await self._request(f"/api/v1/user/challenge/{challenge_id}") + return self._parse_challenge(response) + + async def api_key(self, *, challenge_id: str) -> str: + """Exchange a resolved challenge for the API key of the device. + + Args: + ---- + challenge_id: ID of the resolved challenge. + + Returns: + ------- + The API key, to use with `LaMetricDevice`. + + Raises: + ------ + LaMetricError: The challenge is not resolved, or the device has + no API key. + + """ + response = await self._request( + "/api/v1/user/request/exchange", + method=hdrs.METH_POST, + data={"challenge_id": challenge_id}, + ) + + try: + admin_key = response["user"]["key"] + except (KeyError, TypeError) as exception: + msg = f"The LaMetric device at {self.host} did not hand out a key" + raise LaMetricError(msg) from exception + + # The exchange hands out a web admin key, which stays the same when + # the API key is regenerated. The API key itself is in the users. + users = await self._request("/api/v1/user", admin_key=admin_key) + keys = [ + user + for user in (users if isinstance(users, list) else []) + if isinstance(user, dict) + and user.get("group") == "integration" + and user.get("status") == "active" + and isinstance(user.get("key"), str) + ] + if not keys: + msg = ( + f"The LaMetric device at {self.host} has no API key yet," + " generate one in the LaMetric app first" + ) + raise LaMetricError(msg) + + # A key generated in the LaMetric app stays on the device, the one + # for the LaMetric account comes from the cloud. Prefer the local one. + keys.sort(key=lambda user: user.get("origin") != "local") + return keys[0]["key"] + + async def close(self) -> None: + """Close open client session.""" + if self.session and self._close_session: + await self.session.close() + + async def __aenter__(self) -> Self: + """Async enter. + + Returns + ------- + The LaMetricLocalAuth object. + + """ + return self + + async def __aexit__(self, *_exc_info: object) -> None: + """Async exit. + + Args: + ---- + _exc_info: Exec type. + + """ + await self.close() diff --git a/src/demetriek/models.py b/src/demetriek/models.py index 50b52ae4..642f5b2f 100644 --- a/src/demetriek/models.py +++ b/src/demetriek/models.py @@ -71,6 +71,28 @@ class BluetoothLowEnergy(DataClassORJSONMixin): connectable: bool | None = None +@dataclass(kw_only=True) +class AuthChallenge(DataClassORJSONMixin): + """Object holding a challenge to get the API key of an LaMetric device.""" + + challenge_id: str = field(metadata=field_options(alias="uuid")) + duration: int + + # Seen so far: "in-progress", "resolved", and "expired" once the time + # runs out. Kept as text, so a state not seen yet does not break parsing. + state: str + + @property + def resolved(self) -> bool: + """Return whether the button on the device has been pressed.""" + return self.state == "resolved" + + class Config(BaseConfig): + """AuthChallenge model configuration.""" + + allow_deserialization_not_by_alias = True + + @dataclass(kw_only=True) class Bluetooth(DataClassORJSONMixin): """Object holding the Bluetooth state of an LaMetric device.""" diff --git a/tests/conftest.py b/tests/conftest.py index 399b4501..90698429 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -13,7 +13,7 @@ from aioresponses import core as aioresponses_core from yarl import URL -from demetriek import LaMetricCloud, LaMetricDevice +from demetriek import LaMetricCloud, LaMetricDevice, LaMetricLocalAuth if TYPE_CHECKING: from collections.abc import AsyncGenerator, Generator @@ -103,3 +103,10 @@ async def cloud() -> AsyncGenerator[LaMetricCloud, None]: """Yield a LaMetric cloud client on a shared session.""" async with aiohttp.ClientSession() as session: yield LaMetricCloud(token="abc", session=session) # noqa: S106 + + +@pytest.fixture +async def auth() -> AsyncGenerator[LaMetricLocalAuth, None]: + """Yield a local auth client on a shared session.""" + async with aiohttp.ClientSession() as session: + yield LaMetricLocalAuth(host="127.0.0.2", session=session) diff --git a/tests/fixtures/auth_challenge_resolved_sa8.json b/tests/fixtures/auth_challenge_resolved_sa8.json new file mode 100644 index 00000000..f5a101ce --- /dev/null +++ b/tests/fixtures/auth_challenge_resolved_sa8.json @@ -0,0 +1,7 @@ +{ + "date": "Sat Oct 3 13:50:12 2026", + "duration": 60, + "state": "resolved", + "type": "press_key", + "uuid": "7b3636393566323030" +} diff --git a/tests/fixtures/auth_challenge_sa8.json b/tests/fixtures/auth_challenge_sa8.json new file mode 100644 index 00000000..8b6a0ea3 --- /dev/null +++ b/tests/fixtures/auth_challenge_sa8.json @@ -0,0 +1,9 @@ +{ + "challenge": { + "date": "Sat Oct 3 13:50:08 2026", + "duration": 60, + "state": "in-progress", + "type": "press_key", + "uuid": "7b3636393566323030" + } +} diff --git a/tests/fixtures/auth_exchange_sa8.json b/tests/fixtures/auth_exchange_sa8.json new file mode 100644 index 00000000..6aaaefe1 --- /dev/null +++ b/tests/fixtures/auth_exchange_sa8.json @@ -0,0 +1,10 @@ +{ + "user": { + "group": "web_admin", + "key": "webadminkey", + "name": "", + "origin": "local", + "status": "active", + "uuid": "7b3636393566323030" + } +} diff --git a/tests/fixtures/auth_users_sa8.json b/tests/fixtures/auth_users_sa8.json new file mode 100644 index 00000000..e69d45f6 --- /dev/null +++ b/tests/fixtures/auth_users_sa8.json @@ -0,0 +1,26 @@ +[ + { + "group": "guest", + "key": "guest", + "name": "", + "origin": "local", + "status": "inactive", + "uuid": "guestuuid" + }, + { + "group": "integration", + "key": "accountintegrationkey", + "name": "Account Integration Key", + "origin": "remote", + "status": "active", + "uuid": "integrationuuid" + }, + { + "group": "web_admin", + "key": "webadminkey", + "name": "", + "origin": "local", + "status": "active", + "uuid": "7b3636393566323030" + } +] diff --git a/tests/test_lametric.py b/tests/test_lametric.py index 6053bfd4..3e238359 100644 --- a/tests/test_lametric.py +++ b/tests/test_lametric.py @@ -311,6 +311,8 @@ async def test_http_error401_message( [ ('{"errors": [{"message": "Forbidden"}]}', "Forbidden"), ('{"errors": [{"message": "One"}, {"message": "Two"}]}', "One; Two"), + ('{"error": {"message": "Invalid auth"}}', "Invalid auth"), + ('{"error": "nonsense"}', None), ('{"errors": [{"code": 1}, "nonsense"]}', None), ('{"errors": "nonsense"}', None), ('{"errors": []}', None), diff --git a/tests/test_local_auth.py b/tests/test_local_auth.py new file mode 100644 index 00000000..fca70a58 --- /dev/null +++ b/tests/test_local_auth.py @@ -0,0 +1,255 @@ +"""Asynchronous Python client for LaMetric TIME devices. + +The fixtures are answers of the web interface of a real sa8 TIME on +firmware 3.2.6, with the keys and IDs replaced. +""" + +# pylint: disable=protected-access +import json + +import aiohttp +import pytest +from aioresponses import aioresponses +from yarl import URL + +from demetriek import ( + LaMetricAuthenticationError, + LaMetricConnectionError, + LaMetricConnectionTimeoutError, + LaMetricError, + LaMetricLocalAuth, +) + +from .conftest import load_fixture + +WEB_URL = "https://127.0.0.2" +CHALLENGE_ID = "7b3636393566323030" +REQUEST_URL = f"{WEB_URL}/api/v1/user/request?group=web_admin" +CHALLENGE_URL = f"{WEB_URL}/api/v1/user/challenge/{CHALLENGE_ID}" +EXCHANGE_URL = f"{WEB_URL}/api/v1/user/request/exchange" +USERS_URL = f"{WEB_URL}/api/v1/user" + + +async def test_request_challenge( + responses: aioresponses, auth: LaMetricLocalAuth +) -> None: + """Test requesting a challenge, which the device shows on its screen.""" + responses.post( + REQUEST_URL, status=200, body=load_fixture("auth_challenge_sa8.json") + ) + + challenge = await auth.request_challenge() + + assert challenge.challenge_id == CHALLENGE_ID + assert challenge.duration == 60 + assert challenge.state == "in-progress" + assert challenge.resolved is False + assert ("POST", URL(REQUEST_URL)) in responses.requests + + +async def test_request_challenge_not_supported( + responses: aioresponses, auth: LaMetricLocalAuth +) -> None: + """Test a device without this flow, like an LM 37X8 TIME, says so. + + It asks for credentials instead, which is no reason to ask for new ones. + """ + responses.post( + REQUEST_URL, + status=401, + body='{"errors":[{"message":"Authorization is required"}]}', + ) + + with pytest.raises(LaMetricError, match="does not support") as error: + await auth.request_challenge() + + assert error.type is LaMetricError + + +@pytest.mark.parametrize("body", ['{"nope": true}', "[]"]) +async def test_request_challenge_unexpected_data( + responses: aioresponses, auth: LaMetricLocalAuth, body: str +) -> None: + """Test an answer that is no challenge raises a LaMetricError.""" + responses.post(REQUEST_URL, status=200, body=body) + + with pytest.raises(LaMetricError, match="does not understand"): + await auth.request_challenge() + + +@pytest.mark.parametrize( + ("state", "resolved"), + [("in-progress", False), ("resolved", True), ("expired", False)], +) +async def test_challenge( + responses: aioresponses, + auth: LaMetricLocalAuth, + state: str, + resolved: bool, # noqa: FBT001 +) -> None: + """Test polling a challenge, including the states seen on a real device.""" + data = json.loads(load_fixture("auth_challenge_resolved_sa8.json")) + data["state"] = state + responses.get(CHALLENGE_URL, status=200, body=json.dumps(data)) + + challenge = await auth.challenge(challenge_id=CHALLENGE_ID) + + assert challenge.state == state + assert challenge.resolved is resolved + + +async def test_challenge_unknown( + responses: aioresponses, auth: LaMetricLocalAuth +) -> None: + """Test the reason of the device is passed on, in its own error shape.""" + responses.get( + CHALLENGE_URL, + status=500, + body='{"error":{ "message":"Failed to get info about challenge"}}', + ) + + with pytest.raises(LaMetricError, match="Failed to get info about challenge"): + await auth.challenge(challenge_id=CHALLENGE_ID) + + +async def test_api_key(responses: aioresponses, auth: LaMetricLocalAuth) -> None: + """Test exchanging a resolved challenge for the API key.""" + responses.post( + EXCHANGE_URL, status=200, body=load_fixture("auth_exchange_sa8.json") + ) + responses.get(USERS_URL, status=200, body=load_fixture("auth_users_sa8.json")) + + assert await auth.api_key(challenge_id=CHALLENGE_ID) == "accountintegrationkey" + + exchange = responses.requests[("POST", URL(EXCHANGE_URL))][0] + assert exchange.kwargs["json"] == {"challenge_id": CHALLENGE_ID} + + # The web interface takes the web admin key from a cookie. + users = responses.requests[("GET", URL(USERS_URL))][0] + assert users.kwargs["headers"]["Cookie"] == ( + "no-auth-challenge=1; authorization=webadminkey:" + ) + + +async def test_api_key_prefers_local( + responses: aioresponses, auth: LaMetricLocalAuth +) -> None: + """Test a key generated in the LaMetric app wins over the account one.""" + users = json.loads(load_fixture("auth_users_sa8.json")) + users.append( + { + "group": "integration", + "key": "localkey", + "name": "", + "origin": "local", + "status": "active", + "uuid": "localuuid", + } + ) + responses.post( + EXCHANGE_URL, status=200, body=load_fixture("auth_exchange_sa8.json") + ) + responses.get(USERS_URL, status=200, body=json.dumps(users)) + + assert await auth.api_key(challenge_id=CHALLENGE_ID) == "localkey" + + +async def test_api_key_none(responses: aioresponses, auth: LaMetricLocalAuth) -> None: + """Test a device without any API key yet says what to do.""" + users = [ + user + for user in json.loads(load_fixture("auth_users_sa8.json")) + if user["group"] != "integration" + ] + responses.post( + EXCHANGE_URL, status=200, body=load_fixture("auth_exchange_sa8.json") + ) + responses.get(USERS_URL, status=200, body=json.dumps(users)) + + with pytest.raises(LaMetricError, match="generate one in the LaMetric app"): + await auth.api_key(challenge_id=CHALLENGE_ID) + + +async def test_api_key_not_resolved( + responses: aioresponses, auth: LaMetricLocalAuth +) -> None: + """Test exchanging before the button was pressed passes on the reason.""" + responses.post( + EXCHANGE_URL, + status=400, + body='{"error":{ "message":"Invalid challenge status in-progress"}}', + ) + + with pytest.raises(LaMetricError, match="Invalid challenge status in-progress"): + await auth.api_key(challenge_id=CHALLENGE_ID) + + +async def test_api_key_no_key_handed_out( + responses: aioresponses, auth: LaMetricLocalAuth +) -> None: + """Test an exchange without a key raises a LaMetricError.""" + responses.post(EXCHANGE_URL, status=200, body='{"user": {}}') + + with pytest.raises(LaMetricError, match="did not hand out a key"): + await auth.api_key(challenge_id=CHALLENGE_ID) + + +async def test_api_key_admin_key_refused( + responses: aioresponses, auth: LaMetricLocalAuth +) -> None: + """Test a refused web admin key raises an authentication error.""" + responses.post( + EXCHANGE_URL, status=200, body=load_fixture("auth_exchange_sa8.json") + ) + responses.get(USERS_URL, status=401, body='{"error":{ "message":"Invalid auth"}}') + + with pytest.raises(LaMetricAuthenticationError, match="Invalid auth"): + await auth.api_key(challenge_id=CHALLENGE_ID) + + +async def test_invalid_json(responses: aioresponses, auth: LaMetricLocalAuth) -> None: + """Test a broken JSON answer raises a LaMetricError.""" + responses.get(CHALLENGE_URL, status=200, body="{") + + with pytest.raises(LaMetricError, match="invalid JSON"): + await auth.challenge(challenge_id=CHALLENGE_ID) + + +@pytest.mark.parametrize( + ("exception", "expected"), + [ + (TimeoutError(), LaMetricConnectionTimeoutError), + (aiohttp.ClientError(), LaMetricConnectionError), + ], +) +async def test_connection_errors( + responses: aioresponses, + auth: LaMetricLocalAuth, + exception: Exception, + expected: type[Exception], +) -> None: + """Test connection problems raise connection errors, without retrying. + + Requesting a challenge twice would show it twice on the device. + """ + responses.post(REQUEST_URL, exception=exception, repeat=True) + + with pytest.raises(expected): + await auth.request_challenge() + + assert len(responses.requests[("POST", URL(REQUEST_URL))]) == 1 + + +async def test_internal_session(responses: aioresponses) -> None: + """Test the client creates and closes its own session.""" + responses.get( + CHALLENGE_URL, status=200, body=load_fixture("auth_challenge_resolved_sa8.json") + ) + + async with LaMetricLocalAuth(host="127.0.0.2") as auth: + challenge = await auth.challenge(challenge_id=CHALLENGE_ID) + session = auth.session + + assert challenge.resolved is True + assert session is not None + assert session.closed