Revert "Add probing a stream (#23)" (#26)

Author: edenhaus

go2rtc_client/models.py

@@ -3,7 +3,7 @@
 from __future__ import annotations
 
 from dataclasses import dataclass, field
-from typing import Generic, Literal, TypeVar
+from typing import Any, Literal
 
 from awesomeversion import AwesomeVersion
 from mashumaro import field_options
@@ -19,20 +19,6 @@ def deserialize(self, value: str) -> AwesomeVersion:
         return AwesomeVersion(value)
 
 
-_T = TypeVar("_T")
-
-
-class _EmptyListInsteadNoneSerializer(SerializationStrategy, Generic[_T]):
-    def serialize(self, value: list[_T]) -> list[_T]:
-        return value
-
-    def deserialize(self, value: list[_T] | None) -> list[_T]:
-        if value is None:
-            return []
-
-        return value
-
-
 @dataclass
 class ApplicationInfo(DataClassORJSONMixin):
     """Application info model.
@@ -53,22 +39,26 @@ class Streams(DataClassORJSONMixin):
 
 
 @dataclass
-class Stream(DataClassORJSONMixin):
+class Stream:
     """Stream model."""
 
-    producers: list[Producer] = field(
-        metadata=field_options(serialization_strategy=_EmptyListInsteadNoneSerializer())
-    )
+    producers: list[Producer]
+
+    @classmethod
+    def __pre_deserialize__(cls, d: dict[Any, Any]) -> dict[Any, Any]:
+        """Pre deserialize."""
+        # Ensure producers is always a list
+        if "producers" in d and d["producers"] is None:
+            d["producers"] = []
+
+        return d
 
 
 @dataclass
 class Producer:
     """Producer model."""
 
     url: str
-    medias: list[str] = field(
-        metadata=field_options(serialization_strategy=_EmptyListInsteadNoneSerializer())
-    )
 
 
 @dataclass

go2rtc_client/rest.py

@@ -133,19 +133,6 @@ async def list(self) -> dict[str, Stream]:
         resp = await self._client.request("GET", self.PATH)
         return _GET_STREAMS_DECODER.decode(await resp.json())
 
-    @handle_error
-    async def probe(
-        self, stream_name: str, *, audio: str | None = None, video: str | None = None
-    ) -> Stream:
-        """Probe a stream."""
-        params = {"src": stream_name}
-        if audio:
-            params["audio"] = audio
-        if video:
-            params["video"] = video
-        resp = await self._client.request("GET", self.PATH, params=params)
-        return Stream.from_dict(await resp.json())
-
 
 class Go2RtcRestClient:
     """Rest client for go2rtc server."""

tests/__snapshots__/test_rest.ambr

@@ -9,57 +9,6 @@
     'version': '1.9.4',
   })
 # ---
-# name: test_probe_success[audio and video][deserialized]
-  dict({
-    'producers': list([
-      dict({
-        'medias': list([
-          'video, recvonly, H264',
-          'audio, recvonly, MPEG4-GENERIC/16000',
-          'audio, sendonly, PCMU/8000',
-        ]),
-        'url': 'rtsp://x:[email protected]:554/Preview_01_sub',
-      }),
-    ]),
-  })
-# ---
-# name: test_probe_success[audio and video][serialized]
-  '{"producers":[{"url":"rtsp://x:[email protected]:554/Preview_01_sub","medias":["video, recvonly, H264","audio, recvonly, MPEG4-GENERIC/16000","audio, sendonly, PCMU/8000"]}]}'
-# ---
-# name: test_probe_success[audio only][deserialized]
-  dict({
-    'producers': list([
-      dict({
-        'medias': list([
-          'video, recvonly, H264',
-          'audio, recvonly, MPEG4-GENERIC/16000',
-          'audio, sendonly, PCMU/8000',
-        ]),
-        'url': 'rtsp://x:[email protected]:554/Preview_01_sub',
-      }),
-    ]),
-  })
-# ---
-# name: test_probe_success[audio only][serialized]
-  '{"producers":[{"url":"rtsp://x:[email protected]:554/Preview_01_sub","medias":["video, recvonly, H264","audio, recvonly, MPEG4-GENERIC/16000","audio, sendonly, PCMU/8000"]}]}'
-# ---
-# name: test_probe_success[video only][deserialized]
-  dict({
-    'producers': list([
-      dict({
-        'medias': list([
-          'video, recvonly, H264',
-          'audio, recvonly, MPEG4-GENERIC/16000',
-          'audio, sendonly, PCMU/8000',
-        ]),
-        'url': 'rtsp://x:[email protected]:554/Preview_01_sub',
-      }),
-    ]),
-  })
-# ---
-# name: test_probe_success[video only][serialized]
-  '{"producers":[{"url":"rtsp://x:[email protected]:554/Preview_01_sub","medias":["video, recvonly, H264","audio, recvonly, MPEG4-GENERIC/16000","audio, sendonly, PCMU/8000"]}]}'
-# ---
 # name: test_streams_get[empty]
   dict({
   })
@@ -69,10 +18,6 @@
     'camera.12mp_fluent': dict({
       'producers': list([
         dict({
-          'medias': list([
-            'video, recvonly, H264',
-            'audio, recvonly, MPEG4-GENERIC/16000',
-          ]),
           'url': 'rtsp://test:[email protected]:554/Preview_06_sub',
         }),
       ]),

tests/fixtures/probe_camera_offline.txt

@@ -11,8 +11,8 @@
 from awesomeversion import AwesomeVersion
 import pytest
 
-from go2rtc_client.exceptions import Go2RtcClientError, Go2RtcVersionError
-from go2rtc_client.models import Stream, WebRTCSdpOffer
+from go2rtc_client.exceptions import Go2RtcVersionError
+from go2rtc_client.models import WebRTCSdpOffer
 from go2rtc_client.rest import _ApplicationClient, _StreamClient, _WebRTCClient
 from tests import load_fixture
 
@@ -179,74 +179,3 @@ async def test_webrtc_offer(
         WebRTCSdpOffer("v=0..."),
     )
     assert resp == snapshot
-
-
-async def _test_probe(
-    responses: aioresponses,
-    rest_client: Go2RtcRestClient,
-    filename: str,
-    status_code: int,
-    additional_params: dict[str, str],
-) -> Stream:
-    """Test probing a stream."""
-    camera = "camera.test"
-    params = [f"{k}={v}" for k, v in additional_params.items()]
-    responses.get(
-        f"{URL}{_StreamClient.PATH}?src={camera}&{'&'.join(params)}",
-        status=status_code,
-        body=load_fixture(filename),
-    )
-    return await rest_client.streams.probe(camera, **additional_params)
-
-
-@pytest.mark.parametrize(
-    "additional_params",
-    [
-        {"audio": "all", "video": "all"},
-        {"audio": "all"},
-        {"video": "all"},
-    ],
-    ids=[
-        "audio and video",
-        "audio only",
-        "video only",
-    ],
-)
-async def test_probe_success(
-    responses: aioresponses,
-    rest_client: Go2RtcRestClient,
-    snapshot: SnapshotAssertion,
-    additional_params: dict[str, str],
-) -> None:
-    """Test probing a stream."""
-    resp = await _test_probe(
-        responses, rest_client, "probe_success.json", 200, additional_params
-    )
-    assert resp == snapshot(name="deserialized")
-    assert isinstance(resp, Stream)
-    assert resp.to_json() == snapshot(name="serialized")
-
-
-@pytest.mark.parametrize(
-    "additional_params",
-    [
-        {"audio": "all", "video": "all"},
-        {"audio": "all"},
-        {"video": "all"},
-    ],
-    ids=[
-        "audio and video",
-        "audio only",
-        "video only",
-    ],
-)
-async def test_probe_camera_offline(
-    responses: aioresponses,
-    rest_client: Go2RtcRestClient,
-    additional_params: dict[str, str],
-) -> None:
-    """Test probing a stream, where the camera is offline."""
-    with pytest.raises(Go2RtcClientError):
-        await _test_probe(
-            responses, rest_client, "probe_camera_offline.txt", 500, additional_params
-        )