evlli/hass-core
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Update more nest tests to use common fixtures (#73303)

Browse source 
Update nest tests to use fixtures
This commit is contained in:
Allen Porter 2022-06-09 22:14:43 -07:00 • committed by GitHub
parent 15621bee3f
commit 7a5fa8eb58
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
4 changed files with 357 additions and 447 deletions
Download patch file Download diff file Expand all files Collapse all files

349
tests/components/nest/test_media_source.py
View file

@ -8,11 +8,11 @@ from collections.abc import Generator
import datetime
from http import HTTPStatus
import io
from typing import Any
from unittest.mock import patch
import aiohttp
import av
from google_nest_sdm.device import Device
from google_nest_sdm.event import EventMessage
import numpy as np
import pytest
@ -27,17 +27,11 @@ from homeassistant.helpers.template import DATE_STR_FORMAT
from homeassistant.setup import async_setup_component
import homeassistant.util.dt as dt_util
from .common import (
CONFIG,
FakeSubscriber,
async_setup_sdm_platform,
create_config_entry,
)
from .common import DEVICE_ID, CreateDevice, FakeSubscriber
from tests.common import async_capture_events
DOMAIN = "nest"
DEVICE_ID = "example/api/device/id"
DEVICE_NAME = "Front"
PLATFORM = "camera"
NEST_EVENT = "nest_event"
@ -90,10 +84,42 @@ def frame_image_data(frame_i, total_frames):
return img
@pytest.fixture
def platforms() -> list[str]:
"""Fixture for platforms to setup."""
return [PLATFORM]
@pytest.fixture(autouse=True)
async def setup_media_source(hass) -> None:
"""Set up media source."""
assert await async_setup_component(hass, "media_source", {})
async def setup_components(hass) -> None:
"""Fixture to initialize the integration."""
await async_setup_component(hass, "media_source", {})
@pytest.fixture
def device_type() -> str:
"""Fixture for the type of device under test."""
return CAMERA_DEVICE_TYPE
@pytest.fixture
def device_traits() -> dict[str, Any]:
"""Fixture for the present traits of the device under test."""
return CAMERA_TRAITS
@pytest.fixture(autouse=True)
def device(
device_type: str, device_traits: dict[str, Any], create_device: CreateDevice
) -> None:
"""Fixture to create a device under test."""
return create_device.create(
raw_data={
"name": DEVICE_ID,
"type": device_type,
"traits": device_traits,
}
)
@pytest.fixture
@ -128,22 +154,23 @@ def mp4() -> io.BytesIO:
return output
async def async_setup_devices(hass, auth, device_type, traits={}, events=[]):
"""Set up the platform and prerequisites."""
devices = {
DEVICE_ID: Device.MakeDevice(
{
"name": DEVICE_ID,
"type": device_type,
"traits": traits,
},
auth=auth,
),
}
subscriber = await async_setup_sdm_platform(hass, PLATFORM, devices=devices)
# Enable feature for fetching media
@pytest.fixture(autouse=True)
def enable_prefetch(subscriber: FakeSubscriber) -> None:
"""Fixture to enable media fetching for tests to exercise."""
subscriber.cache_policy.fetch = True
return subscriber
@pytest.fixture
def cache_size() -> int:
"""Fixture for overrideing cache size."""
return 100
@pytest.fixture(autouse=True)
def apply_cache_size(cache_size):
"""Fixture for patching the cache size."""
with patch("homeassistant.components.nest.EVENT_MEDIA_CACHE_SIZE", new=cache_size):
yield
def create_event(
@ -194,17 +221,20 @@ def create_battery_event_data(
}
async def test_no_eligible_devices(hass, auth):
@pytest.mark.parametrize(
"device_type,device_traits",
[
(
"sdm.devices.types.THERMOSTAT",
{
"sdm.devices.traits.Temperature": {},
},
)
],
)
async def test_no_eligible_devices(hass, setup_platform):
"""Test a media source with no eligible camera devices."""
await async_setup_devices(
hass,
auth,
"sdm.devices.types.THERMOSTAT",
{
"sdm.devices.traits.Temperature": {},
},
)
await setup_platform()
browse = await media_source.async_browse_media(hass, f"{const.URI_SCHEME}{DOMAIN}")
assert browse.domain == DOMAIN
assert browse.identifier == ""
@ -212,10 +242,10 @@ async def test_no_eligible_devices(hass, auth):
assert not browse.children
@pytest.mark.parametrize("traits", [CAMERA_TRAITS, BATTERY_CAMERA_TRAITS])
async def test_supported_device(hass, auth, traits):
@pytest.mark.parametrize("device_traits", [CAMERA_TRAITS, BATTERY_CAMERA_TRAITS])
async def test_supported_device(hass, setup_platform):
"""Test a media source with a supported camera."""
await async_setup_devices(hass, auth, CAMERA_DEVICE_TYPE, traits)
await setup_platform()
assert len(hass.states.async_all()) == 1
camera = hass.states.get("camera.front")
@ -245,14 +275,9 @@ async def test_supported_device(hass, auth, traits):
assert len(browse.children) == 0
async def test_integration_unloaded(hass, auth):
async def test_integration_unloaded(hass, auth, setup_platform):
"""Test the media player loads, but has no devices, when config unloaded."""
await async_setup_devices(
hass,
auth,
CAMERA_DEVICE_TYPE,
CAMERA_TRAITS,
)
await setup_platform()
browse = await media_source.async_browse_media(hass, f"{const.URI_SCHEME}{DOMAIN}")
assert browse.domain == DOMAIN
@ -276,11 +301,9 @@ async def test_integration_unloaded(hass, auth):
assert len(browse.children) == 0
async def test_camera_event(hass, auth, hass_client):
async def test_camera_event(hass, hass_client, subscriber, auth, setup_platform):
"""Test a media source and image created for an event."""
subscriber = await async_setup_devices(
hass, auth, CAMERA_DEVICE_TYPE, CAMERA_TRAITS
)
await setup_platform()
assert len(hass.states.async_all()) == 1
camera = hass.states.get("camera.front")
@ -380,11 +403,9 @@ async def test_camera_event(hass, auth, hass_client):
assert media.mime_type == "image/jpeg"
async def test_event_order(hass, auth):
async def test_event_order(hass, auth, subscriber, setup_platform):
"""Test multiple events are in descending timestamp order."""
subscriber = await async_setup_devices(
hass, auth, CAMERA_DEVICE_TYPE, CAMERA_TRAITS
)
await setup_platform()
auth.responses = [
aiohttp.web.json_response(GENERATE_IMAGE_URL_RESPONSE),
@ -449,14 +470,15 @@ async def test_event_order(hass, auth):
assert not browse.children[1].can_play
async def test_multiple_image_events_in_session(hass, auth, hass_client):
async def test_multiple_image_events_in_session(
hass, auth, hass_client, subscriber, setup_platform
):
"""Test multiple events published within the same event session."""
await setup_platform()
event_session_id = "FWWVQVUdGNUlTU2V4MGV2aTNXV..."
event_timestamp1 = dt_util.now()
event_timestamp2 = event_timestamp1 + datetime.timedelta(seconds=5)
subscriber = await async_setup_devices(
hass, auth, CAMERA_DEVICE_TYPE, CAMERA_TRAITS
)
assert len(hass.states.async_all()) == 1
camera = hass.states.get("camera.front")
@ -560,13 +582,19 @@ async def test_multiple_image_events_in_session(hass, auth, hass_client):
assert contents == IMAGE_BYTES_FROM_EVENT + b"-1"
async def test_multiple_clip_preview_events_in_session(hass, auth, hass_client):
@pytest.mark.parametrize("device_traits", [BATTERY_CAMERA_TRAITS])
async def test_multiple_clip_preview_events_in_session(
hass,
auth,
hass_client,
subscriber,
setup_platform,
):
"""Test multiple events published within the same event session."""
await setup_platform()
event_timestamp1 = dt_util.now()
event_timestamp2 = event_timestamp1 + datetime.timedelta(seconds=5)
subscriber = await async_setup_devices(
hass, auth, CAMERA_DEVICE_TYPE, BATTERY_CAMERA_TRAITS
)
assert len(hass.states.async_all()) == 1
camera = hass.states.get("camera.front")
@ -656,9 +684,9 @@ async def test_multiple_clip_preview_events_in_session(hass, auth, hass_client):
assert contents == IMAGE_BYTES_FROM_EVENT
async def test_browse_invalid_device_id(hass, auth):
async def test_browse_invalid_device_id(hass, auth, setup_platform):
"""Test a media source request for an invalid device id."""
await async_setup_devices(hass, auth, CAMERA_DEVICE_TYPE, CAMERA_TRAITS)
await setup_platform()
device_registry = dr.async_get(hass)
device = device_registry.async_get_device({(DOMAIN, DEVICE_ID)})
@ -676,9 +704,9 @@ async def test_browse_invalid_device_id(hass, auth):
)
async def test_browse_invalid_event_id(hass, auth):
async def test_browse_invalid_event_id(hass, auth, setup_platform):
"""Test a media source browsing for an invalid event id."""
await async_setup_devices(hass, auth, CAMERA_DEVICE_TYPE, CAMERA_TRAITS)
await setup_platform()
device_registry = dr.async_get(hass)
device = device_registry.async_get_device({(DOMAIN, DEVICE_ID)})
@ -699,9 +727,9 @@ async def test_browse_invalid_event_id(hass, auth):
)
async def test_resolve_missing_event_id(hass, auth):
async def test_resolve_missing_event_id(hass, auth, setup_platform):
"""Test a media source request missing an event id."""
await async_setup_devices(hass, auth, CAMERA_DEVICE_TYPE, CAMERA_TRAITS)
await setup_platform()
device_registry = dr.async_get(hass)
device = device_registry.async_get_device({(DOMAIN, DEVICE_ID)})
@ -716,10 +744,9 @@ async def test_resolve_missing_event_id(hass, auth):
)
async def test_resolve_invalid_device_id(hass, auth):
async def test_resolve_invalid_device_id(hass, auth, setup_platform):
"""Test resolving media for an invalid event id."""
await async_setup_devices(hass, auth, CAMERA_DEVICE_TYPE, CAMERA_TRAITS)
await setup_platform()
with pytest.raises(Unresolvable):
await media_source.async_resolve_media(
hass,
@ -728,9 +755,9 @@ async def test_resolve_invalid_device_id(hass, auth):
)
async def test_resolve_invalid_event_id(hass, auth):
async def test_resolve_invalid_event_id(hass, auth, setup_platform):
"""Test resolving media for an invalid event id."""
await async_setup_devices(hass, auth, CAMERA_DEVICE_TYPE, CAMERA_TRAITS)
await setup_platform()
device_registry = dr.async_get(hass)
device = device_registry.async_get_device({(DOMAIN, DEVICE_ID)})
@ -750,14 +777,14 @@ async def test_resolve_invalid_event_id(hass, auth):
assert media.mime_type == "image/jpeg"
async def test_camera_event_clip_preview(hass, auth, hass_client, mp4):
@pytest.mark.parametrize("device_traits", [BATTERY_CAMERA_TRAITS])
async def test_camera_event_clip_preview(
hass, auth, hass_client, mp4, subscriber, setup_platform
):
"""Test an event for a battery camera video clip."""
subscriber = await async_setup_devices(
hass, auth, CAMERA_DEVICE_TYPE, BATTERY_CAMERA_TRAITS
)
# Capture any events published
received_events = async_capture_events(hass, NEST_EVENT)
await setup_platform()
auth.responses = [
aiohttp.web.Response(body=mp4.getvalue()),
@ -857,10 +884,11 @@ async def test_camera_event_clip_preview(hass, auth, hass_client, mp4):
await response.read() # Animated gif format not tested
async def test_event_media_render_invalid_device_id(hass, auth, hass_client):
async def test_event_media_render_invalid_device_id(
hass, auth, hass_client, setup_platform
):
"""Test event media API called with an invalid device id."""
await async_setup_devices(hass, auth, CAMERA_DEVICE_TYPE, CAMERA_TRAITS)
await setup_platform()
client = await hass_client()
response = await client.get("/api/nest/event_media/invalid-device-id")
assert response.status == HTTPStatus.NOT_FOUND, (
@ -868,10 +896,11 @@ async def test_event_media_render_invalid_device_id(hass, auth, hass_client):
)
async def test_event_media_render_invalid_event_id(hass, auth, hass_client):
async def test_event_media_render_invalid_event_id(
hass, auth, hass_client, setup_platform
):
"""Test event media API called with an invalid device id."""
await async_setup_devices(hass, auth, CAMERA_DEVICE_TYPE, CAMERA_TRAITS)
await setup_platform()
device_registry = dr.async_get(hass)
device = device_registry.async_get_device({(DOMAIN, DEVICE_ID)})
assert device
@ -884,13 +913,11 @@ async def test_event_media_render_invalid_event_id(hass, auth, hass_client):
)
async def test_event_media_failure(hass, auth, hass_client):
async def test_event_media_failure(hass, auth, hass_client, subscriber, setup_platform):
"""Test event media fetch sees a failure from the server."""
subscriber = await async_setup_devices(
hass, auth, CAMERA_DEVICE_TYPE, CAMERA_TRAITS
)
received_events = async_capture_events(hass, NEST_EVENT)
await setup_platform()
# Failure from server when fetching media
auth.responses = [
aiohttp.web.Response(status=HTTPStatus.INTERNAL_SERVER_ERROR),
@ -937,10 +964,11 @@ async def test_event_media_failure(hass, auth, hass_client):
)
async def test_media_permission_unauthorized(hass, auth, hass_client, hass_admin_user):
async def test_media_permission_unauthorized(
hass, auth, hass_client, hass_admin_user, setup_platform
):
"""Test case where user does not have permissions to view media."""
await async_setup_devices(hass, auth, CAMERA_DEVICE_TYPE, CAMERA_TRAITS)
await setup_platform()
assert len(hass.states.async_all()) == 1
camera = hass.states.get("camera.front")
assert camera is not None
@ -962,33 +990,22 @@ async def test_media_permission_unauthorized(hass, auth, hass_client, hass_admin
)
async def test_multiple_devices(hass, auth, hass_client):
async def test_multiple_devices(
hass, auth, hass_client, create_device, subscriber, setup_platform
):
"""Test events received for multiple devices."""
device_id1 = f"{DEVICE_ID}-1"
device_id2 = f"{DEVICE_ID}-2"
devices = {
device_id1: Device.MakeDevice(
{
"name": device_id1,
"type": CAMERA_DEVICE_TYPE,
"traits": CAMERA_TRAITS,
},
auth=auth,
),
device_id2: Device.MakeDevice(
{
"name": device_id2,
"type": CAMERA_DEVICE_TYPE,
"traits": CAMERA_TRAITS,
},
auth=auth,
),
}
subscriber = await async_setup_sdm_platform(hass, PLATFORM, devices=devices)
create_device.create(
raw_data={
"name": device_id2,
"type": CAMERA_DEVICE_TYPE,
"traits": CAMERA_TRAITS,
}
)
await setup_platform()
device_registry = dr.async_get(hass)
device1 = device_registry.async_get_device({(DOMAIN, device_id1)})
device1 = device_registry.async_get_device({(DOMAIN, DEVICE_ID)})
assert device1
device2 = device_registry.async_get_device({(DOMAIN, device_id2)})
assert device2
@ -1018,7 +1035,7 @@ async def test_multiple_devices(hass, auth, hass_client):
f"event-session-id-{i}",
f"event-id-{i}",
PERSON_EVENT,
device_id=device_id1,
device_id=DEVICE_ID,
)
)
await hass.async_block_till_done()
@ -1073,34 +1090,18 @@ def event_store() -> Generator[None, None, None]:
yield
async def test_media_store_persistence(hass, auth, hass_client, event_store):
@pytest.mark.parametrize("device_traits", [BATTERY_CAMERA_TRAITS])
async def test_media_store_persistence(
hass,
auth,
hass_client,
event_store,
subscriber,
setup_platform,
config_entry,
):
"""Test the disk backed media store persistence."""
nest_device = Device.MakeDevice(
{
"name": DEVICE_ID,
"type": CAMERA_DEVICE_TYPE,
"traits": BATTERY_CAMERA_TRAITS,
},
auth=auth,
)
subscriber = FakeSubscriber()
device_manager = await subscriber.async_get_device_manager()
device_manager.add_device(nest_device)
# Fetch media for events when published
subscriber.cache_policy.fetch = True
config_entry = create_config_entry()
config_entry.add_to_hass(hass)
with patch(
"homeassistant.helpers.config_entry_oauth2_flow.async_get_config_entry_implementation"
), patch("homeassistant.components.nest.PLATFORMS", [PLATFORM]), patch(
"homeassistant.components.nest.api.GoogleNestSubscriber",
return_value=subscriber,
):
assert await async_setup_component(hass, DOMAIN, CONFIG)
await hass.async_block_till_done()
await setup_platform()
device_registry = dr.async_get(hass)
device = device_registry.async_get_device({(DOMAIN, DEVICE_ID)})
@ -1154,18 +1155,8 @@ async def test_media_store_persistence(hass, auth, hass_client, event_store):
# Now rebuild the entire integration and verify that all persisted storage
# can be re-loaded from disk.
subscriber = FakeSubscriber()
device_manager = await subscriber.async_get_device_manager()
device_manager.add_device(nest_device)
with patch(
"homeassistant.helpers.config_entry_oauth2_flow.async_get_config_entry_implementation"
), patch("homeassistant.components.nest.PLATFORMS", [PLATFORM]), patch(
"homeassistant.components.nest.api.GoogleNestSubscriber",
return_value=subscriber,
):
await hass.config_entries.async_reload(config_entry.entry_id)
await hass.async_block_till_done()
await hass.config_entries.async_reload(config_entry.entry_id)
await hass.async_block_till_done()
device_registry = dr.async_get(hass)
device = device_registry.async_get_device({(DOMAIN, DEVICE_ID)})
@ -1197,11 +1188,12 @@ async def test_media_store_persistence(hass, auth, hass_client, event_store):
assert contents == IMAGE_BYTES_FROM_EVENT
async def test_media_store_save_filesystem_error(hass, auth, hass_client):
@pytest.mark.parametrize("device_traits", [BATTERY_CAMERA_TRAITS])
async def test_media_store_save_filesystem_error(
hass, auth, hass_client, subscriber, setup_platform
):
"""Test a filesystem error writing event media."""
subscriber = await async_setup_devices(
hass, auth, CAMERA_DEVICE_TYPE, BATTERY_CAMERA_TRAITS
)
await setup_platform()
auth.responses = [
aiohttp.web.Response(body=IMAGE_BYTES_FROM_EVENT),
@ -1250,11 +1242,11 @@ async def test_media_store_save_filesystem_error(hass, auth, hass_client):
)
async def test_media_store_load_filesystem_error(hass, auth, hass_client):
async def test_media_store_load_filesystem_error(
hass, auth, hass_client, subscriber, setup_platform
):
"""Test a filesystem error reading event media."""
subscriber = await async_setup_devices(
hass, auth, CAMERA_DEVICE_TYPE, BATTERY_CAMERA_TRAITS
)
await setup_platform()
assert len(hass.states.async_all()) == 1
camera = hass.states.get("camera.front")
@ -1299,17 +1291,12 @@ async def test_media_store_load_filesystem_error(hass, auth, hass_client):
)
async def test_camera_event_media_eviction(hass, auth, hass_client):
@pytest.mark.parametrize("device_traits,cache_size", [(BATTERY_CAMERA_TRAITS, 5)])
async def test_camera_event_media_eviction(
hass, auth, hass_client, subscriber, setup_platform
):
"""Test media files getting evicted from the cache."""
# Set small cache size for testing eviction
with patch("homeassistant.components.nest.EVENT_MEDIA_CACHE_SIZE", new=5):
subscriber = await async_setup_devices(
hass,
auth,
CAMERA_DEVICE_TYPE,
BATTERY_CAMERA_TRAITS,
)
await setup_platform()
device_registry = dr.async_get(hass)
device = device_registry.async_get_device({(DOMAIN, DEVICE_ID)})
@ -1384,23 +1371,9 @@ async def test_camera_event_media_eviction(hass, auth, hass_client):
await hass.async_block_till_done()
async def test_camera_image_resize(hass, auth, hass_client):
async def test_camera_image_resize(hass, auth, hass_client, subscriber, setup_platform):
"""Test scaling a thumbnail for an event image."""
event_timestamp = dt_util.now()
subscriber = await async_setup_devices(
hass,
auth,
CAMERA_DEVICE_TYPE,
CAMERA_TRAITS,
events=[
create_event(
EVENT_SESSION_ID,
EVENT_ID,
PERSON_EVENT,
timestamp=event_timestamp,
),
],
)
await setup_platform()
device_registry = dr.async_get(hass)
device = device_registry.async_get_device({(DOMAIN, DEVICE_ID)})