Add support for /.well-known/thread/br-rest API discovery

python_otbr_api/__init__.py

@@ -2,7 +2,7 @@
 
 from __future__ import annotations
 
-from enum import Enum
+from enum import Enum, auto
 from http import HTTPStatus
 from typing import Any
 import json
@@ -72,6 +72,15 @@ class KeyFormat(Enum):
     PASCAL_CASE = "pascal"
 
 
+class _UndefinedType(Enum):
+    """Singleton sentinel distinguishing "not probed yet" from a probed None."""
+
+    UNDEFINED = auto()
+
+
+_UNDEFINED = _UndefinedType.UNDEFINED
+
+
 class OTBRError(Exception):
     """Raised on error."""
 
@@ -111,6 +120,7 @@ def __init__(
         self._url = url
         self._timeout = timeout
         self._key_format = key_format
+        self._api_version: str | None | _UndefinedType = _UNDEFINED
 
     async def _maybe_detect_key_format(self) -> None:
         """Probe the OTBR REST API to determine the JSON key format."""
@@ -407,3 +417,36 @@ async def get_coprocessor_version(self) -> str:
             return await response.json()
         except ValueError as exc:
             raise OTBRError("unexpected API response") from exc
+
+    async def get_api_version(self) -> str | None:
+        """Get the OTBR REST API's semantic version.
+
+        Reads /.well-known/thread/br-rest (ot-br-posix PR #3330). Returns
+        None on routers that don't expose it. Raises OTBRError if the
+        endpoint exists but responds with an unexpected status or a
+        malformed body.
+        """
+        if self._api_version is not _UNDEFINED:
+            return self._api_version
+
+        response = await self._session.get(
+            f"{self._url}/.well-known/thread/br-rest",
+            timeout=aiohttp.ClientTimeout(total=self._timeout),
+        )
+
+        if response.status == HTTPStatus.NOT_FOUND:
+            self._api_version = None
+            return None
+
+        if response.status != HTTPStatus.OK:
+            raise OTBRError(f"unexpected http status {response.status}")
+
+        try:
+            data = await response.json()
+            api_version: str = data["api"]["version"]
+        except (ValueError, KeyError, TypeError) as exc:
+            raise OTBRError("unexpected API response") from exc
+
+        self._api_version = api_version
+        _LOGGER.debug("Detected OTBR REST API version: %s", api_version)
+        return api_version

tests/test_well_known.py

@@ -0,0 +1,108 @@
+"""Tests for the /.well-known/thread/br-rest API discovery endpoint."""
+
+from http import HTTPStatus
+
+import pytest
+import python_otbr_api
+
+from tests.test_util.aiohttp import AiohttpClientMocker
+
+BASE_URL = "http://core-openthread-border-router:8081"
+
+WELL_KNOWN_JSON = {
+    "api": {"version": "0.3.0", "base": "/api/"},
+    "links": [
+        {
+            "href": "/.well-known/thread/br-rest",
+            "rel": "self",
+            "type": ["application/json"],
+        },
+        {"href": "/api/node", "rel": "node", "type": ["application/vnd.api+json"]},
+        {"href": "/api/actions", "rel": "task", "type": ["application/vnd.api+json"]},
+        {
+            "href": "/api/devices",
+            "rel": "device",
+            "type": ["application/vnd.api+json"],
+        },
+        {
+            "href": "/api/diagnostics",
+            "rel": "diagnostic",
+            "type": ["application/vnd.api+json"],
+        },
+    ],
+}
+
+
+def _otbr(aioclient_mock: AiohttpClientMocker) -> python_otbr_api.OTBR:
+    return python_otbr_api.OTBR(BASE_URL, aioclient_mock.create_session())
+
+
+async def test_get_api_version(aioclient_mock: AiohttpClientMocker) -> None:
+    """A 200 with a version resource returns the advertised semver string."""
+    otbr = _otbr(aioclient_mock)
+    aioclient_mock.get(f"{BASE_URL}/.well-known/thread/br-rest", json=WELL_KNOWN_JSON)
+
+    assert await otbr.get_api_version() == "0.3.0"
+
+
+async def test_get_api_version_not_supported(
+    aioclient_mock: AiohttpClientMocker,
+) -> None:
+    """A 404 means the router predates ot-br-posix #3330: version is unknown."""
+    otbr = _otbr(aioclient_mock)
+    aioclient_mock.get(
+        f"{BASE_URL}/.well-known/thread/br-rest", status=HTTPStatus.NOT_FOUND
+    )
+
+    assert await otbr.get_api_version() is None
+
+
+async def test_get_api_version_unexpected_status(
+    aioclient_mock: AiohttpClientMocker,
+) -> None:
+    """Any other non-200 status raises OTBRError."""
+    otbr = _otbr(aioclient_mock)
+    aioclient_mock.get(
+        f"{BASE_URL}/.well-known/thread/br-rest",
+        status=HTTPStatus.INTERNAL_SERVER_ERROR,
+    )
+
+    with pytest.raises(python_otbr_api.OTBRError):
+        await otbr.get_api_version()
+
+
+async def test_get_api_version_malformed_body(
+    aioclient_mock: AiohttpClientMocker,
+) -> None:
+    """A 200 without the expected `api.version` field raises OTBRError."""
+    otbr = _otbr(aioclient_mock)
+    aioclient_mock.get(f"{BASE_URL}/.well-known/thread/br-rest", json={"api": {}})
+
+    with pytest.raises(python_otbr_api.OTBRError):
+        await otbr.get_api_version()
+
+
+async def test_get_api_version_runs_once(aioclient_mock: AiohttpClientMocker) -> None:
+    """Detection happens lazily on first call and is cached for subsequent calls."""
+    otbr = _otbr(aioclient_mock)
+    aioclient_mock.get(f"{BASE_URL}/.well-known/thread/br-rest", json=WELL_KNOWN_JSON)
+
+    assert await otbr.get_api_version() == "0.3.0"
+    assert await otbr.get_api_version() == "0.3.0"
+
+    assert aioclient_mock.call_count == 1
+
+
+async def test_get_api_version_not_supported_runs_once(
+    aioclient_mock: AiohttpClientMocker,
+) -> None:
+    """A 404 result is cached too, so repeated calls don't re-probe."""
+    otbr = _otbr(aioclient_mock)
+    aioclient_mock.get(
+        f"{BASE_URL}/.well-known/thread/br-rest", status=HTTPStatus.NOT_FOUND
+    )
+
+    assert await otbr.get_api_version() is None
+    assert await otbr.get_api_version() is None
+
+    assert aioclient_mock.call_count == 1