Add method for validating the go2rtc server version is supported (#5)

* Add method for validating the go2rtc server version is supported

* Silence ruff

* Add fixture

* Use awesomeversion

* Run linters

* Improve test coverage

Author: emontnemery

go2rtc_client/models.py

@@ -5,7 +5,30 @@
 from dataclasses import dataclass, field
 from typing import Literal
 
+from awesomeversion import AwesomeVersion
+from mashumaro import field_options
 from mashumaro.mixins.orjson import DataClassORJSONMixin
+from mashumaro.types import SerializationStrategy
+
+
+class _AwesomeVersionSerializer(SerializationStrategy):
+    def serialize(self, value: AwesomeVersion) -> str:
+        return str(value)
+
+    def deserialize(self, value: str) -> AwesomeVersion:
+        return AwesomeVersion(value)
+
+
+@dataclass
+class ApplicationInfo(DataClassORJSONMixin):
+    """Application info model.
+
+    Currently only the server version is exposed.
+    """
+
+    version: AwesomeVersion = field(
+        metadata=field_options(serialization_strategy=_AwesomeVersionSerializer())
+    )
 
 
 @dataclass

go2rtc_client/rest.py

@@ -7,19 +7,21 @@
 
 from aiohttp import ClientError, ClientResponse, ClientSession
 from aiohttp.client import _RequestOptions
+from awesomeversion import AwesomeVersion
 from mashumaro.codecs.basic import BasicDecoder
 from mashumaro.mixins.dict import DataClassDictMixin
 from yarl import URL
 
 from .exceptions import handle_error
-from .models import Stream, WebRTCSdpAnswer, WebRTCSdpOffer
+from .models import ApplicationInfo, Stream, WebRTCSdpAnswer, WebRTCSdpOffer
 
 if TYPE_CHECKING:
     from collections.abc import Mapping
 
 _LOGGER = logging.getLogger(__name__)
 
 _API_PREFIX = "/api"
+_SUPPORTED_VERSION: Final = AwesomeVersion("1.9.4")
 
 
 class _BaseClient:
@@ -58,6 +60,19 @@ async def request(
         return resp
 
 
+class _ApplicationClient:
+    PATH: Final = _API_PREFIX
+
+    def __init__(self, client: _BaseClient) -> None:
+        """Initialize Client."""
+        self._client = client
+
+    async def get_info(self) -> ApplicationInfo:
+        """Get application info."""
+        resp = await self._client.request("GET", self.PATH)
+        return ApplicationInfo.from_dict(await resp.json())
+
+
 class _WebRTCClient:
     """Client for WebRTC module."""
 
@@ -123,5 +138,11 @@ class Go2RtcRestClient:
     def __init__(self, websession: ClientSession, server_url: str) -> None:
         """Initialize Client."""
         self._client = _BaseClient(websession, server_url)
+        self.application: Final = _ApplicationClient(self._client)
         self.streams: Final = _StreamClient(self._client)
         self.webrtc: Final = _WebRTCClient(self._client)
+
+    async def validate_server_version(self) -> bool:
+        """Validate the server version is compatible."""
+        application_info = await self.application.get_info()
+        return application_info.version == _SUPPORTED_VERSION

pyproject.toml

@@ -19,6 +19,7 @@ classifiers = [
 requires-python = ">=3.12.0"
 dependencies = [
     "aiohttp~=3.10",
+    "awesomeversion>=24.6.0",
     "mashumaro~=3.13",
     "orjson>=3.10.7",
 ]

tests/__snapshots__/test_rest.ambr

@@ -1,4 +1,14 @@
 # serializer version: 1
+# name: test_application_info
+  dict({
+    'version': <AwesomeVersion SemVer '1.9.4'>,
+  })
+# ---
+# name: test_application_info.1
+  dict({
+    'version': '1.9.4',
+  })
+# ---
 # name: test_streams_get[empty]
   dict({
   })

tests/fixtures/application_info_answer.json

@@ -0,0 +1,7 @@
+{
+  "config_path": "/home/erik/go2rtc/go2rtc.yaml",
+  "host": "127.0.0.1:1984",
+  "revision": "a4885c2",
+  "rtsp": { "listen": ":8554", "default_query": "video\u0026audio" },
+  "version": "1.9.4"
+}

tests/test_rest.py

@@ -2,13 +2,15 @@
 
 from __future__ import annotations
 
+import json
 from typing import TYPE_CHECKING
 
 from aiohttp.hdrs import METH_PUT
+from awesomeversion import AwesomeVersion
 import pytest
 
 from go2rtc_client.models import WebRTCSdpOffer
-from go2rtc_client.rest import _StreamClient, _WebRTCClient
+from go2rtc_client.rest import _ApplicationClient, _StreamClient, _WebRTCClient
 from tests import load_fixture
 
 from . import URL
@@ -20,6 +22,23 @@
     from go2rtc_client import Go2RtcRestClient
 
 
+async def test_application_info(
+    responses: aioresponses,
+    rest_client: Go2RtcRestClient,
+    snapshot: SnapshotAssertion,
+) -> None:
+    """Test webrtc offer."""
+    responses.get(
+        f"{URL}{_ApplicationClient.PATH}",
+        status=200,
+        body=load_fixture("application_info_answer.json"),
+    )
+    resp = await rest_client.application.get_info()
+    assert isinstance(resp.version, AwesomeVersion)
+    assert resp == snapshot
+    assert resp.to_dict() == snapshot
+
+
 @pytest.mark.parametrize(
     "filename",
     [
@@ -69,6 +88,34 @@ async def test_streams_add(
     responses.assert_called_once_with(url, method=METH_PUT, params=params)
 
 
+@pytest.mark.parametrize(
+    ("server_version", "expected_result"),
+    [
+        ("0.0.0", False),
+        ("1.9.3", False),
+        ("1.9.4", True),
+        ("1.9.5", False),
+        ("2.0.0", False),
+        ("BLAH", False),
+    ],
+)
+async def test_version_supported(
+    responses: aioresponses,
+    rest_client: Go2RtcRestClient,
+    server_version: str,
+    expected_result: bool,
+) -> None:
+    """Test webrtc offer."""
+    payload = json.loads(load_fixture("application_info_answer.json"))
+    payload["version"] = server_version
+    responses.get(
+        f"{URL}{_ApplicationClient.PATH}",
+        status=200,
+        payload=payload,
+    )
+    assert await rest_client.validate_server_version() == expected_result
+
+
 async def test_webrtc_offer(
     responses: aioresponses,
     rest_client: Go2RtcRestClient,