Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
14 changes: 6 additions & 8 deletions simplipy/websocket.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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)
Expand Down
35 changes: 35 additions & 0 deletions tests/conftest.py
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand Down
87 changes: 87 additions & 0 deletions tests/fixtures/ws_motion_event_flv_data.json
Original file line number Diff line number Diff line change
@@ -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"
}
}
}
}
}
17 changes: 16 additions & 1 deletion tests/test_websocket.py
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand All @@ -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
Expand Down
Loading