diff --git a/simplipy/websocket.py b/simplipy/websocket.py index a33e066c..66c6daaa 100644 --- a/simplipy/websocket.py +++ b/simplipy/websocket.py @@ -164,7 +164,7 @@ class WebsocketEvent: event_type: str | None = field(init=False) timestamp: datetime = field(init=False) - media_urls: dict[str, str] | None = field(init=False) + media_urls: dict[str, str | None] | None = field(init=False) changed_by: str | None = None sensor_name: str | None = None @@ -203,17 +203,15 @@ def __post_init__(self, event_cid: int) -> None: object.__setattr__(self, "sensor_type", None) if self._vid is not None and self._video is not None: + links = self._video[self._vid]["_links"] object.__setattr__( self, "media_urls", { - "image_url": self._video[self._vid]["_links"]["snapshot/jpg"][ - "href" - ], - "clip_url": self._video[self._vid]["_links"]["download/mp4"][ - "href" - ], - "hls_url": self._video[self._vid]["_links"]["playback/hls"]["href"], + "image_url": links["snapshot/jpg"]["href"], + "clip_url": links["download/mp4"]["href"], + "hls_url": (links.get("playback/hls") or {}).get("href"), + "flv_url": (links.get("playback/flv") or {}).get("href"), }, ) object.__setattr__(self, "_vid", None) diff --git a/tests/conftest.py b/tests/conftest.py index ab33cc8b..893eb3e6 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -420,6 +420,41 @@ def ws_motion_event_data_fixture() -> dict[str, Any]: return cast(dict[str, Any], json.loads(load_fixture("ws_motion_event_data.json"))) +@pytest.fixture(name="ws_motion_event_flv_data", scope="session") +def ws_motion_event_flv_data_fixture() -> dict[str, Any]: + """Define a fixture that returns the data payload from a FLV-only event. + + Returns: + A API response payload. + """ + return cast( + dict[str, Any], json.loads(load_fixture("ws_motion_event_flv_data.json")) + ) + + +@pytest.fixture(name="ws_motion_event_flv") +def ws_motion_event_flv_fixture( + ws_motion_event_flv_data: dict[str, Any], +) -> dict[str, Any]: + """Define a fixture to represent a FLV-only event response. + + Args: + ws_motion_event_flv_data: A mocked websocket response payload. + + Returns: + A websocket response payload. + """ + return { + "data": ws_motion_event_flv_data, + "datacontenttype": "application/json", + "id": "id:16803409109", + "source": "messagequeue", + "specversion": "1.0", + "time": "2021-09-29T23:14:46.000Z", + "type": "com.simplisafe.event.standard", + } + + @pytest.fixture(name="ws_message_hello") def ws_message_hello_fixture(ws_message_hello_data: dict[str, Any]) -> dict[str, Any]: """Define a fixture to represent the "hello" response. diff --git a/tests/fixtures/ws_motion_event_flv_data.json b/tests/fixtures/ws_motion_event_flv_data.json new file mode 100644 index 00000000..af7e44f5 --- /dev/null +++ b/tests/fixtures/ws_motion_event_flv_data.json @@ -0,0 +1,87 @@ +{ + "eventUuid": "xxx", + "eventTimestamp": 1703882325, + "eventCid": 1170, + "zoneCid": "1", + "sensorType": 17, + "sensorSerial": "f11b6abd", + "account": "abcdef12", + "userId": 12345, + "sid": 12345, + "info": "Front Door Camera Detected Motion", + "pinName": "", + "sensorName": "Front Door", + "messageSubject": "Camera Detected Motion", + "messageBody": "Front Door Camera Detected Motion on 12/29/2023 at 12:38 PM", + "eventType": "activityCam", + "timezone": 3, + "locationOffset": -480, + "internal": { + "dispatcher": "my_dispatcher" + }, + "senderId": "", + "eventId": 38552328826, + "serviceFeatures": { + "monitoring": false, + "alerts": true, + "online": true, + "hazard": false, + "video": true, + "cameras": 10, + "dispatch": false, + "proInstall": false, + "discount": 0, + "vipCS": false, + "medical": false, + "careVisit": false, + "storageDays": 30 + }, + "copsVideoOptIn": false, + "videoStartedBy": "a1b2c3d4e5f6a7b8c9d0e1f2a3b4c5d6", + "video": { + "a1b2c3d4e5f6a7b8c9d0e1f2a3b4c5d6": { + "clipId": "998877665544", + "preroll": 5, + "postroll": 60, + "cameraName": "Front Door", + "eventId": "38552328826", + "sid": 12345, + "timestamp": 1703882325, + "recordingType": "TEP", + "account": "abcdef12", + "region": "us-east-1", + "actualDuration": 0, + "status": "PENDING", + "_links": { + "_self": { + "href": "https://chronicle.us-east-1.prd.cam.simplisafe.com/v1/recordings/998877665544", + "method": "GET" + }, + "playback/flv": { + "href": "https://flv-url", + "method": "GET" + }, + "preview/mjpg": { + "href": "https://preview-url", + "method": "GET" + }, + "snapshot/mjpg": { + "href": "https://snapshot-mjpg-url", + "method": "GET" + }, + "snapshot/jpg": { + "href": "https://snapshot-jpg-url", + "method": "GET" + }, + "download/mp4": { + "href": "https://clip-url", + "method": "GET" + }, + "share": { + "href": "https://share-url", + "method": "POST" + } + } + } + } +} diff --git a/tests/test_websocket.py b/tests/test_websocket.py index 886b5d37..67740d7b 100644 --- a/tests/test_websocket.py +++ b/tests/test_websocket.py @@ -151,7 +151,7 @@ def test_create_event(ws_message_event: dict[str, Any]) -> None: def test_create_motion_event(ws_motion_event: dict[str, Any]) -> None: - """Test creating a motion event object. + """Test creating a motion event object with HLS links. Args: ws_motion_event: A websocket motion event payload with media urls. @@ -161,6 +161,21 @@ def test_create_motion_event(ws_motion_event: dict[str, Any]) -> None: assert event.media_urls["image_url"] == "https://image-url{&width}" assert event.media_urls["clip_url"] == "https://clip-url" assert event.media_urls["hls_url"] == "https://hls-url" + assert event.media_urls["flv_url"] is None + + +def test_create_motion_event_flv(ws_motion_event_flv: dict[str, Any]) -> None: + """Test creating a motion event object with FLV links (no HLS). + + Args: + ws_motion_event_flv: A websocket motion event payload with FLV-only links. + """ + event = websocket_event_from_payload(ws_motion_event_flv) + assert event.media_urls is not None + assert event.media_urls["image_url"] == "https://snapshot-jpg-url" + assert event.media_urls["clip_url"] == "https://clip-url" + assert event.media_urls["hls_url"] is None + assert event.media_urls["flv_url"] == "https://flv-url" @pytest.mark.asyncio