From 58551ec66da1344bcbd8fd25712dd9074ea4e1e7 Mon Sep 17 00:00:00 2001 From: Allen Porter Date: Thu, 17 Feb 2022 23:08:29 -0800 Subject: [PATCH] Update nest camera tests to use common test fixture (#66192) --- tests/components/nest/test_camera_sdm.py | 279 +++++++++++++---------- 1 file changed, 160 insertions(+), 119 deletions(-) diff --git a/tests/components/nest/test_camera_sdm.py b/tests/components/nest/test_camera_sdm.py index 81b4a7cf0a6..b64e251bcf0 100644 --- a/tests/components/nest/test_camera_sdm.py +++ b/tests/components/nest/test_camera_sdm.py @@ -10,7 +10,6 @@ from http import HTTPStatus from unittest.mock import AsyncMock, Mock, patch import aiohttp -from google_nest_sdm.device import Device from google_nest_sdm.event import EventMessage import pytest @@ -21,18 +20,20 @@ from homeassistant.components.camera import ( STREAM_TYPE_HLS, STREAM_TYPE_WEB_RTC, ) +from homeassistant.components.nest.const import DOMAIN from homeassistant.components.websocket_api.const import TYPE_RESULT +from homeassistant.core import HomeAssistant from homeassistant.helpers import device_registry as dr, entity_registry as er from homeassistant.setup import async_setup_component from homeassistant.util.dt import utcnow -from .common import async_setup_sdm_platform +from .common import DEVICE_ID, CreateDevice, FakeSubscriber, PlatformSetup +from .conftest import FakeAuth from tests.common import async_fire_time_changed PLATFORM = "camera" CAMERA_DEVICE_TYPE = "sdm.devices.types.CAMERA" -DEVICE_ID = "some-device-id" DEVICE_TRAITS = { "sdm.devices.traits.Info": { "customName": "My Camera", @@ -49,7 +50,6 @@ DEVICE_TRAITS = { "sdm.devices.traits.CameraMotion": {}, } DATETIME_FORMAT = "YY-MM-DDTHH:MM:SS" -DOMAIN = "nest" MOTION_EVENT_ID = "FWWVQVUdGNUlTU2V4MGV2aTNXV..." EVENT_SESSION_ID = "CjY5Y3VKaTZwR3o4Y19YbTVfMF..." @@ -65,6 +65,45 @@ GENERATE_IMAGE_URL_RESPONSE = { IMAGE_AUTHORIZATION_HEADERS = {"Authorization": "Basic g.0.eventToken"} +@pytest.fixture +def platforms() -> list[str]: + """Fixture to set platforms used in the test.""" + return ["camera"] + + +@pytest.fixture +async def device_type() -> str: + """Fixture to set default device type used when creating devices.""" + return "sdm.devices.types.CAMERA" + + +@pytest.fixture +def camera_device(create_device: CreateDevice) -> None: + """Fixture to create a basic camera device.""" + create_device.create(DEVICE_TRAITS) + + +@pytest.fixture +def webrtc_camera_device(create_device: CreateDevice) -> None: + """Fixture to create a WebRTC camera device.""" + create_device.create( + { + "sdm.devices.traits.Info": { + "customName": "My Camera", + }, + "sdm.devices.traits.CameraLiveStream": { + "maxVideoResolution": { + "width": 640, + "height": 480, + }, + "videoCodecs": ["H264"], + "audioCodecs": ["AAC"], + "supportedProtocols": ["WEB_RTC"], + }, + } + ) + + def make_motion_event( event_id: str = MOTION_EVENT_ID, event_session_id: str = EVENT_SESSION_ID, @@ -136,21 +175,6 @@ async def async_get_image(hass, width=None, height=None): return image.content -async def async_setup_camera(hass, traits={}, auth=None): - """Set up the platform and prerequisites.""" - devices = {} - if traits: - devices[DEVICE_ID] = Device.MakeDevice( - { - "name": DEVICE_ID, - "type": CAMERA_DEVICE_TYPE, - "traits": traits, - }, - auth=auth, - ) - return await async_setup_sdm_platform(hass, PLATFORM, devices) - - async def fire_alarm(hass, point_in_time): """Fire an alarm and wait for callbacks to run.""" with patch("homeassistant.util.dt.utcnow", return_value=point_in_time): @@ -158,28 +182,33 @@ async def fire_alarm(hass, point_in_time): await hass.async_block_till_done() -async def test_no_devices(hass): +async def test_no_devices(hass: HomeAssistant, setup_platform: PlatformSetup): """Test configuration that returns no devices.""" - await async_setup_camera(hass) + await setup_platform() assert len(hass.states.async_all()) == 0 -async def test_ineligible_device(hass): +async def test_ineligible_device( + hass: HomeAssistant, setup_platform: PlatformSetup, create_device: CreateDevice +): """Test configuration with devices that do not support cameras.""" - await async_setup_camera( - hass, + create_device.create( { "sdm.devices.traits.Info": { "customName": "My Camera", }, - }, + } ) + + await setup_platform() assert len(hass.states.async_all()) == 0 -async def test_camera_device(hass): +async def test_camera_device( + hass: HomeAssistant, setup_platform: PlatformSetup, camera_device: None +): """Test a basic camera with a live stream.""" - await async_setup_camera(hass, DEVICE_TRAITS) + await setup_platform() assert len(hass.states.async_all()) == 1 camera = hass.states.get("camera.my_camera") @@ -188,7 +217,7 @@ async def test_camera_device(hass): registry = er.async_get(hass) entry = registry.async_get("camera.my_camera") - assert entry.unique_id == "some-device-id-camera" + assert entry.unique_id == f"{DEVICE_ID}-camera" assert entry.original_name == "My Camera" assert entry.domain == "camera" @@ -199,10 +228,16 @@ async def test_camera_device(hass): assert device.identifiers == {("nest", DEVICE_ID)} -async def test_camera_stream(hass, auth, mock_create_stream): +async def test_camera_stream( + hass: HomeAssistant, + setup_platform: PlatformSetup, + camera_device: None, + auth: FakeAuth, + mock_create_stream: Mock, +): """Test a basic camera and fetch its live stream.""" auth.responses = [make_stream_url_response()] - await async_setup_camera(hass, DEVICE_TRAITS, auth=auth) + await setup_platform() assert len(hass.states.async_all()) == 1 cam = hass.states.get("camera.my_camera") @@ -216,10 +251,17 @@ async def test_camera_stream(hass, auth, mock_create_stream): assert await async_get_image(hass) == IMAGE_BYTES_FROM_STREAM -async def test_camera_ws_stream(hass, auth, hass_ws_client, mock_create_stream): +async def test_camera_ws_stream( + hass, + setup_platform, + camera_device, + hass_ws_client, + auth, + mock_create_stream, +): """Test a basic camera that supports web rtc.""" auth.responses = [make_stream_url_response()] - await async_setup_camera(hass, DEVICE_TRAITS, auth=auth) + await setup_platform() assert len(hass.states.async_all()) == 1 cam = hass.states.get("camera.my_camera") @@ -245,10 +287,12 @@ async def test_camera_ws_stream(hass, auth, hass_ws_client, mock_create_stream): assert await async_get_image(hass) == IMAGE_BYTES_FROM_STREAM -async def test_camera_ws_stream_failure(hass, auth, hass_ws_client): +async def test_camera_ws_stream_failure( + hass, setup_platform, camera_device, hass_ws_client, auth +): """Test a basic camera that supports web rtc.""" auth.responses = [aiohttp.web.Response(status=HTTPStatus.BAD_REQUEST)] - await async_setup_camera(hass, DEVICE_TRAITS, auth=auth) + await setup_platform() assert len(hass.states.async_all()) == 1 cam = hass.states.get("camera.my_camera") @@ -272,21 +316,22 @@ async def test_camera_ws_stream_failure(hass, auth, hass_ws_client): assert msg["error"]["message"].startswith("Nest API error") -async def test_camera_stream_missing_trait(hass, auth): +async def test_camera_stream_missing_trait(hass, setup_platform, create_device): """Test fetching a video stream when not supported by the API.""" - traits = { - "sdm.devices.traits.Info": { - "customName": "My Camera", - }, - "sdm.devices.traits.CameraImage": { - "maxImageResolution": { - "width": 800, - "height": 600, - } - }, - } - - await async_setup_camera(hass, traits, auth=auth) + create_device.create( + { + "sdm.devices.traits.Info": { + "customName": "My Camera", + }, + "sdm.devices.traits.CameraImage": { + "maxImageResolution": { + "width": 800, + "height": 600, + } + }, + } + ) + await setup_platform() assert len(hass.states.async_all()) == 1 cam = hass.states.get("camera.my_camera") @@ -300,7 +345,12 @@ async def test_camera_stream_missing_trait(hass, auth): await async_get_image(hass) -async def test_refresh_expired_stream_token(hass, auth): +async def test_refresh_expired_stream_token( + hass: HomeAssistant, + setup_platform: PlatformSetup, + auth: FakeAuth, + camera_device: None, +): """Test a camera stream expiration and refresh.""" now = utcnow() stream_1_expiration = now + datetime.timedelta(seconds=90) @@ -314,11 +364,7 @@ async def test_refresh_expired_stream_token(hass, auth): # Stream URL #3 make_stream_url_response(stream_3_expiration, token_num=3), ] - await async_setup_camera( - hass, - DEVICE_TRAITS, - auth=auth, - ) + await setup_platform() assert await async_setup_component(hass, "stream", {}) assert len(hass.states.async_all()) == 1 @@ -375,7 +421,12 @@ async def test_refresh_expired_stream_token(hass, auth): assert hls_url == hls_url2 -async def test_stream_response_already_expired(hass, auth): +async def test_stream_response_already_expired( + hass: HomeAssistant, + auth: FakeAuth, + setup_platform: PlatformSetup, + camera_device: None, +): """Test a API response returning an expired stream url.""" now = utcnow() stream_1_expiration = now + datetime.timedelta(seconds=-90) @@ -384,7 +435,7 @@ async def test_stream_response_already_expired(hass, auth): make_stream_url_response(stream_1_expiration, token_num=1), make_stream_url_response(stream_2_expiration, token_num=2), ] - await async_setup_camera(hass, DEVICE_TRAITS, auth=auth) + await setup_platform() assert len(hass.states.async_all()) == 1 cam = hass.states.get("camera.my_camera") @@ -402,13 +453,15 @@ async def test_stream_response_already_expired(hass, auth): assert stream_source == "rtsp://some/url?auth=g.2.streamingToken" -async def test_camera_removed(hass, auth): +async def test_camera_removed( + hass: HomeAssistant, + auth: FakeAuth, + camera_device: None, + subscriber: FakeSubscriber, + setup_platform: PlatformSetup, +): """Test case where entities are removed and stream tokens revoked.""" - subscriber = await async_setup_camera( - hass, - DEVICE_TRAITS, - auth=auth, - ) + await setup_platform() # Simplify test setup subscriber.cache_policy.fetch = False @@ -431,13 +484,14 @@ async def test_camera_removed(hass, auth): assert len(hass.states.async_all()) == 0 -async def test_camera_remove_failure(hass, auth): +async def test_camera_remove_failure( + hass: HomeAssistant, + auth: FakeAuth, + camera_device: None, + setup_platform: PlatformSetup, +): """Test case where revoking the stream token fails on unload.""" - await async_setup_camera( - hass, - DEVICE_TRAITS, - auth=auth, - ) + await setup_platform() assert len(hass.states.async_all()) == 1 cam = hass.states.get("camera.my_camera") @@ -460,7 +514,12 @@ async def test_camera_remove_failure(hass, auth): assert len(hass.states.async_all()) == 0 -async def test_refresh_expired_stream_failure(hass, auth): +async def test_refresh_expired_stream_failure( + hass: HomeAssistant, + auth: FakeAuth, + setup_platform: PlatformSetup, + camera_device: None, +): """Tests a failure when refreshing the stream.""" now = utcnow() stream_1_expiration = now + datetime.timedelta(seconds=90) @@ -472,7 +531,7 @@ async def test_refresh_expired_stream_failure(hass, auth): # Next attempt to get a stream fetches a new url make_stream_url_response(expiration=stream_2_expiration, token_num=2), ] - await async_setup_camera(hass, DEVICE_TRAITS, auth=auth) + await setup_platform() assert await async_setup_component(hass, "stream", {}) assert len(hass.states.async_all()) == 1 @@ -510,7 +569,9 @@ async def test_refresh_expired_stream_failure(hass, auth): assert create_stream.called -async def test_camera_web_rtc(hass, auth, hass_ws_client): +async def test_camera_web_rtc( + hass, auth, hass_ws_client, webrtc_camera_device, setup_platform +): """Test a basic camera that supports web rtc.""" expiration = utcnow() + datetime.timedelta(seconds=100) auth.responses = [ @@ -524,21 +585,7 @@ async def test_camera_web_rtc(hass, auth, hass_ws_client): } ) ] - device_traits = { - "sdm.devices.traits.Info": { - "customName": "My Camera", - }, - "sdm.devices.traits.CameraLiveStream": { - "maxVideoResolution": { - "width": 640, - "height": 480, - }, - "videoCodecs": ["H264"], - "audioCodecs": ["AAC"], - "supportedProtocols": ["WEB_RTC"], - }, - } - await async_setup_camera(hass, device_traits, auth=auth) + await setup_platform() assert len(hass.states.async_all()) == 1 cam = hass.states.get("camera.my_camera") @@ -567,9 +614,11 @@ async def test_camera_web_rtc(hass, auth, hass_ws_client): await async_get_image(hass, width=1024, height=768) -async def test_camera_web_rtc_unsupported(hass, auth, hass_ws_client): +async def test_camera_web_rtc_unsupported( + hass, auth, hass_ws_client, camera_device, setup_platform +): """Test a basic camera that supports web rtc.""" - await async_setup_camera(hass, DEVICE_TRAITS, auth=auth) + await setup_platform() assert len(hass.states.async_all()) == 1 cam = hass.states.get("camera.my_camera") @@ -595,26 +644,14 @@ async def test_camera_web_rtc_unsupported(hass, auth, hass_ws_client): assert msg["error"]["message"].startswith("Camera does not support WebRTC") -async def test_camera_web_rtc_offer_failure(hass, auth, hass_ws_client): +async def test_camera_web_rtc_offer_failure( + hass, auth, hass_ws_client, webrtc_camera_device, setup_platform +): """Test a basic camera that supports web rtc.""" auth.responses = [ aiohttp.web.Response(status=HTTPStatus.BAD_REQUEST), ] - device_traits = { - "sdm.devices.traits.Info": { - "customName": "My Camera", - }, - "sdm.devices.traits.CameraLiveStream": { - "maxVideoResolution": { - "width": 640, - "height": 480, - }, - "videoCodecs": ["H264"], - "audioCodecs": ["AAC"], - "supportedProtocols": ["WEB_RTC"], - }, - } - await async_setup_camera(hass, device_traits, auth=auth) + await setup_platform() assert len(hass.states.async_all()) == 1 cam = hass.states.get("camera.my_camera") @@ -639,7 +676,9 @@ async def test_camera_web_rtc_offer_failure(hass, auth, hass_ws_client): assert msg["error"]["message"].startswith("Nest API error") -async def test_camera_multiple_streams(hass, auth, hass_ws_client, mock_create_stream): +async def test_camera_multiple_streams( + hass, auth, hass_ws_client, create_device, setup_platform, mock_create_stream +): """Test a camera supporting multiple stream types.""" expiration = utcnow() + datetime.timedelta(seconds=100) auth.responses = [ @@ -656,21 +695,23 @@ async def test_camera_multiple_streams(hass, auth, hass_ws_client, mock_create_s } ), ] - device_traits = { - "sdm.devices.traits.Info": { - "customName": "My Camera", - }, - "sdm.devices.traits.CameraLiveStream": { - "maxVideoResolution": { - "width": 640, - "height": 480, + create_device.create( + { + "sdm.devices.traits.Info": { + "customName": "My Camera", }, - "videoCodecs": ["H264"], - "audioCodecs": ["AAC"], - "supportedProtocols": ["WEB_RTC", "RTSP"], - }, - } - await async_setup_camera(hass, device_traits, auth=auth) + "sdm.devices.traits.CameraLiveStream": { + "maxVideoResolution": { + "width": 640, + "height": 480, + }, + "videoCodecs": ["H264"], + "audioCodecs": ["AAC"], + "supportedProtocols": ["WEB_RTC", "RTSP"], + }, + } + ) + await setup_platform() assert len(hass.states.async_all()) == 1 cam = hass.states.get("camera.my_camera")