Add schemes endpoint (#179)

Author: edenhaus

go2rtc_client/rest.py

@@ -22,7 +22,7 @@
 _LOGGER = logging.getLogger(__name__)
 
 _API_PREFIX = "/api"
-_MIN_VERSION_SUPPORTED: Final = AwesomeVersion("1.9.5")
+_MIN_VERSION_SUPPORTED: Final = AwesomeVersion("1.9.12")
 _MIN_VERSION_UNSUPPORTED: Final = AwesomeVersion("2.0.0")
 
 
@@ -141,13 +141,29 @@ async def list(self) -> dict[str, Stream]:
         return _GET_STREAMS_DECODER.decode(await resp.json())
 
 
+class _SchemesClient:
+    PATH: Final = _API_PREFIX + "/schemes"
+    _DECODER = BasicDecoder(set[str])
+
+    def __init__(self, client: _BaseClient) -> None:
+        """Initialize Client."""
+        self._client = client
+
+    @handle_error
+    async def list(self) -> set[str]:
+        """List all supported schemes."""
+        resp = await self._client.request("GET", self.PATH)
+        return self._DECODER.decode(await resp.json())
+
+
 class Go2RtcRestClient:
     """Rest client for go2rtc server."""
 
     def __init__(self, websession: ClientSession, server_url: str) -> None:
         """Initialize Client."""
         self._client = _BaseClient(websession, server_url)
         self.application: Final = _ApplicationClient(self._client)
+        self.schemes: Final = _SchemesClient(self._client)
         self.streams: Final = _StreamClient(self._client)
         self.webrtc: Final = _WebRTCClient(self._client)
 

tests/__snapshots__/test_rest.ambr

@@ -9,6 +9,16 @@
     'version': '1.9.4',
   })
 # ---
+# name: test_schemes
+  set({
+    'exec',
+    'ffmpeg',
+    'rtsp',
+    'rtsps',
+    'rtspx',
+    'webrtc',
+  })
+# ---
 # name: test_streams_get[empty]
   dict({
   })

tests/test_rest.py

@@ -16,6 +16,7 @@
 from go2rtc_client.rest import (
     _API_PREFIX,
     _ApplicationClient,
+    _SchemesClient,
     _StreamClient,
     _WebRTCClient,
 )
@@ -134,16 +135,17 @@ async def test_streams_add_str(
     )
 
 
-VERSION_ERR = "server version '{}' not >= 1.9.5 and < 2.0.0"
+VERSION_ERR = "server version '{}' not >= 1.9.12 and < 2.0.0"
 
 
 @pytest.mark.parametrize(
     ("server_version", "expected_result"),
     [
         ("0.0.0", pytest.raises(Go2RtcVersionError, match=VERSION_ERR.format("0.0.0"))),
-        ("1.9.4", pytest.raises(Go2RtcVersionError, match=VERSION_ERR.format("1.9.4"))),
-        ("1.9.5", does_not_raise()),
-        ("1.9.6", does_not_raise()),
+        ("1.9.5", pytest.raises(Go2RtcVersionError, match=VERSION_ERR.format("1.9.5"))),
+        ("1.9.6", pytest.raises(Go2RtcVersionError, match=VERSION_ERR.format("1.9.6"))),
+        ("1.9.12", does_not_raise()),
+        ("1.9.13", does_not_raise()),
         ("2.0.0", pytest.raises(Go2RtcVersionError, match=VERSION_ERR.format("2.0.0"))),
         ("BLAH", pytest.raises(Go2RtcVersionError, match=VERSION_ERR.format("BLAH"))),
     ],
@@ -220,3 +222,18 @@ async def test_get_jpeg_snapshot(
     assert isinstance(resp, bytes)
 
     assert resp == image_bytes
+
+
+async def test_schemes(
+    responses: aioresponses,
+    rest_client: Go2RtcRestClient,
+    snapshot: SnapshotAssertion,
+) -> None:
+    """Test schemes."""
+    responses.get(
+        f"{URL}{_SchemesClient.PATH}",
+        status=200,
+        body=json.dumps(["webrtc", "exec", "ffmpeg", "rtsp", "rtsps", "rtspx"]),
+    )
+    resp = await rest_client.schemes.list()
+    assert resp == snapshot