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

Add Nest cam support for the SDM API (#42325)

Browse source 
Co-authored-by: Paulus Schoutsen <paulus@home-assistant.io>
This commit is contained in:
Allen Porter 2020-10-27 07:20:01 -07:00 • committed by GitHub
parent b6df411115
commit 8caa177ba1
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
10 changed files with 721 additions and 280 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
homeassistant/components/nest/__init__.py
View file

@ -96,7 +96,7 @@ CONFIG_SCHEMA = vol.Schema(
extra=vol.ALLOW_EXTRA,
)
PLATFORMS = ["sensor"]
PLATFORMS = ["sensor", "camera"]
# Services for the legacy API

160
homeassistant/components/nest/camera.py
View file

@ -1,150 +1,18 @@
"""Support for Nest Cameras."""
from datetime import timedelta
import logging
"""Support for Nest cameras that dispatches between API versions."""
import requests
from homeassistant.config_entries import ConfigEntry
from homeassistant.helpers.typing import HomeAssistantType
from homeassistant.components import nest
from homeassistant.components.camera import PLATFORM_SCHEMA, SUPPORT_ON_OFF, Camera
from homeassistant.util.dt import utcnow
_LOGGER = logging.getLogger(__name__)
NEST_BRAND = "Nest"
PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend({})
from .camera_legacy import async_setup_legacy_entry
from .camera_sdm import async_setup_sdm_entry
from .const import DATA_SDM
def setup_platform(hass, config, add_entities, discovery_info=None):
"""Set up a Nest Cam.
No longer in use.
"""
async def async_setup_entry(hass, entry, async_add_entities):
"""Set up a Nest sensor based on a config entry."""
camera_devices = await hass.async_add_executor_job(
hass.data[nest.DATA_NEST].cameras
)
cameras = [NestCamera(structure, device) for structure, device in camera_devices]
async_add_entities(cameras, True)
class NestCamera(Camera):
"""Representation of a Nest Camera."""
def __init__(self, structure, device):
"""Initialize a Nest Camera."""
super().__init__()
self.structure = structure
self.device = device
self._location = None
self._name = None
self._online = None
self._is_streaming = None
self._is_video_history_enabled = False
# Default to non-NestAware subscribed, but will be fixed during update
self._time_between_snapshots = timedelta(seconds=30)
self._last_image = None
self._next_snapshot_at = None
@property
def name(self):
"""Return the name of the nest, if any."""
return self._name
@property
def unique_id(self):
"""Return the serial number."""
return self.device.device_id
@property
def device_info(self):
"""Return information about the device."""
return {
"identifiers": {(nest.DOMAIN, self.device.device_id)},
"name": self.device.name_long,
"manufacturer": "Nest Labs",
"model": "Camera",
}
@property
def should_poll(self):
"""Nest camera should poll periodically."""
return True
@property
def is_recording(self):
"""Return true if the device is recording."""
return self._is_streaming
@property
def brand(self):
"""Return the brand of the camera."""
return NEST_BRAND
@property
def supported_features(self):
"""Nest Cam support turn on and off."""
return SUPPORT_ON_OFF
@property
def is_on(self):
"""Return true if on."""
return self._online and self._is_streaming
def turn_off(self):
"""Turn off camera."""
_LOGGER.debug("Turn off camera %s", self._name)
# Calling Nest API in is_streaming setter.
# device.is_streaming would not immediately change until the process
# finished in Nest Cam.
self.device.is_streaming = False
def turn_on(self):
"""Turn on camera."""
if not self._online:
_LOGGER.error("Camera %s is offline", self._name)
return
_LOGGER.debug("Turn on camera %s", self._name)
# Calling Nest API in is_streaming setter.
# device.is_streaming would not immediately change until the process
# finished in Nest Cam.
self.device.is_streaming = True
def update(self):
"""Cache value from Python-nest."""
self._location = self.device.where
self._name = self.device.name
self._online = self.device.online
self._is_streaming = self.device.is_streaming
self._is_video_history_enabled = self.device.is_video_history_enabled
if self._is_video_history_enabled:
# NestAware allowed 10/min
self._time_between_snapshots = timedelta(seconds=6)
else:
# Otherwise, 2/min
self._time_between_snapshots = timedelta(seconds=30)
def _ready_for_snapshot(self, now):
return self._next_snapshot_at is None or now > self._next_snapshot_at
def camera_image(self):
"""Return a still image response from the camera."""
now = utcnow()
if self._ready_for_snapshot(now):
url = self.device.snapshot_url
try:
response = requests.get(url)
except requests.exceptions.RequestException as error:
_LOGGER.error("Error getting camera image: %s", error)
return None
self._next_snapshot_at = now + self._time_between_snapshots
self._last_image = response.content
return self._last_image
async def async_setup_entry(
hass: HomeAssistantType, entry: ConfigEntry, async_add_entities
) -> None:
"""Set up the cameras."""
if DATA_SDM not in entry.data:
await async_setup_legacy_entry(hass, entry, async_add_entities)
return
await async_setup_sdm_entry(hass, entry, async_add_entities)

150
homeassistant/components/nest/camera_legacy.py Normal file
View file

@ -0,0 +1,150 @@
"""Support for Nest Cameras."""
from datetime import timedelta
import logging
import requests
from homeassistant.components import nest
from homeassistant.components.camera import PLATFORM_SCHEMA, SUPPORT_ON_OFF, Camera
from homeassistant.util.dt import utcnow
_LOGGER = logging.getLogger(__name__)
NEST_BRAND = "Nest"
PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend({})
def setup_platform(hass, config, add_entities, discovery_info=None):
"""Set up a Nest Cam.
No longer in use.
"""
async def async_setup_legacy_entry(hass, entry, async_add_entities):
"""Set up a Nest sensor based on a config entry."""
camera_devices = await hass.async_add_executor_job(
hass.data[nest.DATA_NEST].cameras
)
cameras = [NestCamera(structure, device) for structure, device in camera_devices]
async_add_entities(cameras, True)
class NestCamera(Camera):
"""Representation of a Nest Camera."""
def __init__(self, structure, device):
"""Initialize a Nest Camera."""
super().__init__()
self.structure = structure
self.device = device
self._location = None
self._name = None
self._online = None
self._is_streaming = None
self._is_video_history_enabled = False
# Default to non-NestAware subscribed, but will be fixed during update
self._time_between_snapshots = timedelta(seconds=30)
self._last_image = None
self._next_snapshot_at = None
@property
def name(self):
"""Return the name of the nest, if any."""
return self._name
@property
def unique_id(self):
"""Return the serial number."""
return self.device.device_id
@property
def device_info(self):
"""Return information about the device."""
return {
"identifiers": {(nest.DOMAIN, self.device.device_id)},
"name": self.device.name_long,
"manufacturer": "Nest Labs",
"model": "Camera",
}
@property
def should_poll(self):
"""Nest camera should poll periodically."""
return True
@property
def is_recording(self):
"""Return true if the device is recording."""
return self._is_streaming
@property
def brand(self):
"""Return the brand of the camera."""
return NEST_BRAND
@property
def supported_features(self):
"""Nest Cam support turn on and off."""
return SUPPORT_ON_OFF
@property
def is_on(self):
"""Return true if on."""
return self._online and self._is_streaming
def turn_off(self):
"""Turn off camera."""
_LOGGER.debug("Turn off camera %s", self._name)
# Calling Nest API in is_streaming setter.
# device.is_streaming would not immediately change until the process
# finished in Nest Cam.
self.device.is_streaming = False
def turn_on(self):
"""Turn on camera."""
if not self._online:
_LOGGER.error("Camera %s is offline", self._name)
return
_LOGGER.debug("Turn on camera %s", self._name)
# Calling Nest API in is_streaming setter.
# device.is_streaming would not immediately change until the process
# finished in Nest Cam.
self.device.is_streaming = True
def update(self):
"""Cache value from Python-nest."""
self._location = self.device.where
self._name = self.device.name
self._online = self.device.online
self._is_streaming = self.device.is_streaming
self._is_video_history_enabled = self.device.is_video_history_enabled
if self._is_video_history_enabled:
# NestAware allowed 10/min
self._time_between_snapshots = timedelta(seconds=6)
else:
# Otherwise, 2/min
self._time_between_snapshots = timedelta(seconds=30)
def _ready_for_snapshot(self, now):
return self._next_snapshot_at is None or now > self._next_snapshot_at
def camera_image(self):
"""Return a still image response from the camera."""
now = utcnow()
if self._ready_for_snapshot(now):
url = self.device.snapshot_url
try:
response = requests.get(url)
except requests.exceptions.RequestException as error:
_LOGGER.error("Error getting camera image: %s", error)
return None
self._next_snapshot_at = now + self._time_between_snapshots
self._last_image = response.content
return self._last_image

115
homeassistant/components/nest/camera_sdm.py Normal file
View file

@ -0,0 +1,115 @@
"""Support for Google Nest SDM Cameras."""
import logging
from typing import Optional
from google_nest_sdm.camera_traits import CameraImageTrait, CameraLiveStreamTrait
from google_nest_sdm.device import Device
from homeassistant.components.camera import SUPPORT_STREAM, Camera
from homeassistant.config_entries import ConfigEntry
from homeassistant.helpers.dispatcher import async_dispatcher_connect
from homeassistant.helpers.typing import HomeAssistantType
from .const import DOMAIN, SIGNAL_NEST_UPDATE
from .device_info import DeviceInfo
_LOGGER = logging.getLogger(__name__)
async def async_setup_sdm_entry(
hass: HomeAssistantType, entry: ConfigEntry, async_add_entities
) -> None:
"""Set up the cameras."""
subscriber = hass.data[DOMAIN][entry.entry_id]
device_manager = await subscriber.async_get_device_manager()
# Fetch initial data so we have data when entities subscribe.
entities = []
for device in device_manager.devices.values():
if (
CameraImageTrait.NAME in device.traits
or CameraLiveStreamTrait.NAME in device.traits
):
entities.append(NestCamera(device))
async_add_entities(entities)
class NestCamera(Camera):
"""Devices that support cameras."""
def __init__(self, device: Device):
"""Initialize the camera."""
super().__init__()
self._device = device
self._device_info = DeviceInfo(device)
@property
def should_poll(self) -> bool:
"""Disable polling since entities have state pushed via pubsub."""
return False
@property
def unique_id(self) -> Optional[str]:
"""Return a unique ID."""
# The API "name" field is a unique device identifier.
return f"{self._device.name}-camera"
@property
def name(self):
"""Return the name of the camera."""
return self._device_info.device_name
@property
def device_info(self):
"""Return device specific attributes."""
return self._device_info.device_info
@property
def brand(self):
"""Return the camera brand."""
return self._device_info.device_brand
@property
def model(self):
"""Return the camera model."""
return self._device_info.device_model
@property
def supported_features(self):
"""Flag supported features."""
features = 0
if CameraLiveStreamTrait.NAME in self._device.traits:
features = features | SUPPORT_STREAM
return features
async def stream_source(self):
"""Return the source of the stream."""
if CameraLiveStreamTrait.NAME not in self._device.traits:
return None
trait = self._device.traits[CameraLiveStreamTrait.NAME]
rtsp_stream = await trait.generate_rtsp_stream()
# Note: This is only valid for a few minutes, and probably needs
# to be improved with an occasional call to .extend_rtsp_stream() which
# returns a new rtsp_stream object.
return rtsp_stream.rtsp_stream_url
async def async_added_to_hass(self):
"""Run when entity is added to register update signal handler."""
# Event messages trigger the SIGNAL_NEST_UPDATE, which is intercepted
# here to re-fresh the signals from _device. Unregister this callback
# when the entity is removed.
self.async_on_remove(
async_dispatcher_connect(
self.hass, SIGNAL_NEST_UPDATE, self.async_write_ha_state
)
)
async def async_camera_image(self):
"""Return bytes of camera image."""
# No support for still images yet. Still images are only available
# in response to an event on the feed. For now, suppress a
# NotImplementedError in the parent class.
return None

58
homeassistant/components/nest/device_info.py Normal file
View file

@ -0,0 +1,58 @@
"""Library for extracting device specific information common to entities."""
from google_nest_sdm.device import Device
from google_nest_sdm.device_traits import InfoTrait
from .const import DOMAIN
DEVICE_TYPE_MAP = {
"sdm.devices.types.CAMERA": "Camera",
"sdm.devices.types.DISPLAY": "Display",
"sdm.devices.types.DOORBELL": "Doorbell",
"sdm.devices.types.THERMOSTAT": "Thermostat",
}
class DeviceInfo:
"""Provide device info from the SDM device, shared across platforms."""
device_brand = "Google Nest"
def __init__(self, device: Device):
"""Initialize the DeviceInfo."""
self._device = device
@property
def device_info(self):
"""Return device specific attributes."""
return {
# The API "name" field is a unique device identifier.
"identifiers": {(DOMAIN, self._device.name)},
"name": self.device_name,
"manufacturer": self.device_brand,
"model": self.device_model,
}
@property
def device_name(self):
"""Return the name of the physical device that includes the sensor."""
if InfoTrait.NAME in self._device.traits:
trait = self._device.traits[InfoTrait.NAME]
if trait.custom_name:
return trait.custom_name
# Build a name from the room/structure. Note: This room/structure name
# is not associated with a home assistant Area.
parent_relations = self._device.parent_relations
if parent_relations:
items = sorted(parent_relations.items())
names = [name for id, name in items]
return " ".join(names)
return self.device_model
@property
def device_model(self):
"""Return device model information."""
# The API intentionally returns minimal information about specific
# devices, instead relying on traits, but we can infer a generic model
# name based on the type
return DEVICE_TYPE_MAP.get(self._device.type)

40
homeassistant/components/nest/sensor_sdm.py
View file

@ -3,7 +3,7 @@
from typing import Optional
from google_nest_sdm.device import Device
from google_nest_sdm.device_traits import HumidityTrait, InfoTrait, TemperatureTrait
from google_nest_sdm.device_traits import HumidityTrait, TemperatureTrait
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import (
@ -17,6 +17,7 @@ from homeassistant.helpers.entity import Entity
from homeassistant.helpers.typing import HomeAssistantType
from .const import DOMAIN, SIGNAL_NEST_UPDATE
from .device_info import DeviceInfo
DEVICE_TYPE_MAP = {
"sdm.devices.types.CAMERA": "Camera",
@ -51,6 +52,7 @@ class SensorBase(Entity):
def __init__(self, device: Device):
"""Initialize the sensor."""
self._device = device
self._device_info = DeviceInfo(device)
@property
def should_poll(self) -> bool:
@ -63,40 +65,10 @@ class SensorBase(Entity):
# The API "name" field is a unique device identifier.
return f"{self._device.name}-{self.device_class}"
@property
def device_name(self):
"""Return the name of the physical device that includes the sensor."""
if InfoTrait.NAME in self._device.traits:
trait = self._device.traits[InfoTrait.NAME]
if trait.custom_name:
return trait.custom_name
# Build a name from the room/structure. Note: This room/structure name
# is not associated with a home assistant Area.
parent_relations = self._device.parent_relations
if parent_relations:
items = sorted(parent_relations.items())
names = [name for id, name in items]
return " ".join(names)
return self.unique_id
@property
def device_info(self):
"""Return device specific attributes."""
return {
# The API "name" field is a unique device identifier.
"identifiers": {(DOMAIN, self._device.name)},
"name": self.device_name,
"manufacturer": "Google Nest",
"model": self.device_model,
}
@property
def device_model(self):
"""Return device model information."""
# The API intentionally returns minimal information about specific
# devices, instead relying on traits, but we can infer a generic model
# name based on the type
return DEVICE_TYPE_MAP.get(self._device.type)
return self._device_info.device_info
async def async_added_to_hass(self):
"""Run when entity is added to register update signal handler."""
@ -118,7 +90,7 @@ class TemperatureSensor(SensorBase):
@property
def name(self):
"""Return the name of the sensor."""
return f"{self.device_name} Temperature"
return f"{self._device_info.device_name} Temperature"
@property
def state(self):
@ -149,7 +121,7 @@ class HumiditySensor(SensorBase):
@property
def name(self):
"""Return the name of the sensor."""
return f"{self.device_name} Humidity"
return f"{self._device_info.device_name} Humidity"
@property
def state(self):

165
tests/components/nest/camera_sdm_test.py Normal file
View file

@ -0,0 +1,165 @@
"""
Test for Nest cameras platform for the Smart Device Management API.
These tests fake out the subscriber/devicemanager, and are not using a real
pubsub subscriber.
"""
from google_nest_sdm.auth import AbstractAuth
from google_nest_sdm.device import Device
from homeassistant.components import camera
from homeassistant.components.camera import STATE_IDLE
from .common import async_setup_sdm_platform
PLATFORM = "camera"
CAMERA_DEVICE_TYPE = "sdm.devices.types.CAMERA"
DEVICE_ID = "some-device-id"
class FakeResponse:
"""A fake web response used for returning results of commands."""
def __init__(self, json):
"""Initialize the FakeResponse."""
self._json = json
def raise_for_status(self):
"""Mimics a successful response status."""
pass
async def json(self):
"""Return a dict with the response."""
return self._json
class FakeAuth(AbstractAuth):
"""Fake authentication object that returns fake responses."""
def __init__(self, response: FakeResponse):
"""Initialize the FakeAuth."""
super().__init__(None, "")
self._response = response
async def async_get_access_token(self):
"""Return a fake access token."""
return "some-token"
async def creds(self):
"""Return a fake creds."""
return None
async def request(self, method: str, url: str, **kwargs):
"""Pass through the FakeResponse."""
return self._response
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 test_no_devices(hass):
"""Test configuration that returns no devices."""
await async_setup_camera(hass)
assert len(hass.states.async_all()) == 0
async def test_ineligible_device(hass):
"""Test configuration with devices that do not support cameras."""
await async_setup_camera(
hass,
{
"sdm.devices.traits.Info": {
"customName": "My Camera",
},
},
)
assert len(hass.states.async_all()) == 0
async def test_camera_device(hass):
"""Test a basic camera with a live stream."""
await async_setup_camera(
hass,
{
"sdm.devices.traits.Info": {
"customName": "My Camera",
},
"sdm.devices.traits.CameraLiveStream": {
"maxVideoResolution": {
"width": 640,
"height": 480,
},
"videoCodecs": ["H264"],
"audioCodecs": ["AAC"],
},
},
)
assert len(hass.states.async_all()) == 1
camera = hass.states.get("camera.my_camera")
assert camera is not None
assert camera.state == STATE_IDLE
registry = await hass.helpers.entity_registry.async_get_registry()
entry = registry.async_get("camera.my_camera")
assert entry.unique_id == "some-device-id-camera"
assert entry.original_name == "My Camera"
assert entry.domain == "camera"
device_registry = await hass.helpers.device_registry.async_get_registry()
device = device_registry.async_get(entry.device_id)
assert device.name == "My Camera"
assert device.model == "Camera"
assert device.identifiers == {("nest", DEVICE_ID)}
async def test_camera_stream(hass):
"""Test a basic camera and fetch its live stream."""
response = FakeResponse(
{
"results": {
"streamUrls": {"rtspUrl": "rtsp://some/url?auth=g.0.streamingToken"},
"streamExtensionToken": "g.1.extensionToken",
"streamToken": "g.0.streamingToken",
"expiresAt": "2018-01-04T18:30:00.000Z",
},
}
)
await async_setup_camera(
hass,
{
"sdm.devices.traits.Info": {
"customName": "My Camera",
},
"sdm.devices.traits.CameraLiveStream": {
"maxVideoResolution": {
"width": 640,
"height": 480,
},
"videoCodecs": ["H264"],
"audioCodecs": ["AAC"],
},
},
auth=FakeAuth(response),
)
assert len(hass.states.async_all()) == 1
cam = hass.states.get("camera.my_camera")
assert cam is not None
assert cam.state == STATE_IDLE
stream_source = await camera.async_get_stream_source(hass, "camera.my_camera")
assert stream_source == "rtsp://some/url?auth=g.0.streamingToken"

99
tests/components/nest/common.py Normal file
View file

@ -0,0 +1,99 @@
"""Common libraries for test setup."""
import time
from google_nest_sdm.device_manager import DeviceManager
from google_nest_sdm.event import EventCallback, EventMessage
from google_nest_sdm.google_nest_subscriber import GoogleNestSubscriber
from homeassistant.components.nest import DOMAIN
from homeassistant.setup import async_setup_component
from tests.async_mock import patch
from tests.common import MockConfigEntry
CONFIG = {
"nest": {
"client_id": "some-client-id",
"client_secret": "some-client-secret",
# Required fields for using SDM API
"project_id": "some-project-id",
"subscriber_id": "some-subscriber-id",
},
}
CONFIG_ENTRY_DATA = {
"sdm": {}, # Indicates new SDM API, not legacy API
"auth_implementation": "local",
"token": {
"expires_at": time.time() + 86400,
"access_token": {
"token": "some-token",
},
},
}
class FakeDeviceManager(DeviceManager):
"""Fake DeviceManager that can supply a list of devices and structures."""
def __init__(self, devices: dict, structures: dict):
"""Initialize FakeDeviceManager."""
super().__init__()
self._devices = devices
@property
def structures(self) -> dict:
"""Override structures with fake result."""
return self._structures
@property
def devices(self) -> dict:
"""Override devices with fake result."""
return self._devices
class FakeSubscriber(GoogleNestSubscriber):
"""Fake subscriber that supplies a FakeDeviceManager."""
def __init__(self, device_manager: FakeDeviceManager):
"""Initialize Fake Subscriber."""
self._device_manager = device_manager
self._callback = None
def set_update_callback(self, callback: EventCallback):
"""Capture the callback set by Home Assistant."""
self._callback = callback
async def start_async(self) -> DeviceManager:
"""Return the fake device manager."""
return self._device_manager
async def async_get_device_manager(self) -> DeviceManager:
"""Return the fake device manager."""
return self._device_manager
def stop_async(self):
"""No-op to stop the subscriber."""
return None
def receive_event(self, event_message: EventMessage):
"""Simulate a received pubsub message, invoked by tests."""
# Update device state, then invoke HomeAssistant to refresh
self._device_manager.handle_event(event_message)
self._callback.handle_event(event_message)
async def async_setup_sdm_platform(hass, platform, devices={}, structures={}):
"""Set up the platform and prerequisites."""
MockConfigEntry(domain=DOMAIN, data=CONFIG_ENTRY_DATA).add_to_hass(hass)
device_manager = FakeDeviceManager(devices=devices, structures=structures)
subscriber = FakeSubscriber(device_manager)
with patch(
"homeassistant.helpers.config_entry_oauth2_flow.async_get_config_entry_implementation"
), patch("homeassistant.components.nest.PLATFORMS", [platform]), patch(
"homeassistant.components.nest.GoogleNestSubscriber", return_value=subscriber
):
assert await async_setup_component(hass, DOMAIN, CONFIG)
await hass.async_block_till_done()
return subscriber

103
tests/components/nest/device_info_test.py Normal file
View file

@ -0,0 +1,103 @@
"""Test for properties for devices common to all entity types."""
from google_nest_sdm.device import Device
from homeassistant.components.nest.device_info import DeviceInfo
def test_device_custom_name():
"""Test a device name from an Info trait."""
device = Device.MakeDevice(
{
"name": "some-device-id",
"type": "sdm.devices.types.DOORBELL",
"traits": {
"sdm.devices.traits.Info": {
"customName": "My Doorbell",
},
},
},
auth=None,
)
device_info = DeviceInfo(device)
assert device_info.device_name == "My Doorbell"
assert device_info.device_model == "Doorbell"
assert device_info.device_brand == "Google Nest"
assert device_info.device_info == {
"identifiers": {("nest", "some-device-id")},
"name": "My Doorbell",
"manufacturer": "Google Nest",
"model": "Doorbell",
}
def test_device_name_room():
"""Test a device name from the room name."""
device = Device.MakeDevice(
{
"name": "some-device-id",
"type": "sdm.devices.types.DOORBELL",
"parentRelations": [
{"parent": "some-structure-id", "displayName": "Some Room"}
],
},
auth=None,
)
device_info = DeviceInfo(device)
assert device_info.device_name == "Some Room"
assert device_info.device_model == "Doorbell"
assert device_info.device_brand == "Google Nest"
assert device_info.device_info == {
"identifiers": {("nest", "some-device-id")},
"name": "Some Room",
"manufacturer": "Google Nest",
"model": "Doorbell",
}
def test_device_no_name():
"""Test a device that has a name inferred from the type."""
device = Device.MakeDevice(
{"name": "some-device-id", "type": "sdm.devices.types.DOORBELL", "traits": {}},
auth=None,
)
device_info = DeviceInfo(device)
assert device_info.device_name == "Doorbell"
assert device_info.device_model == "Doorbell"
assert device_info.device_brand == "Google Nest"
assert device_info.device_info == {
"identifiers": {("nest", "some-device-id")},
"name": "Doorbell",
"manufacturer": "Google Nest",
"model": "Doorbell",
}
def test_device_invalid_type():
"""Test a device with a type name that is not recognized."""
device = Device.MakeDevice(
{
"name": "some-device-id",
"type": "sdm.devices.types.INVALID_TYPE",
"traits": {
"sdm.devices.traits.Info": {
"customName": "My Doorbell",
},
},
},
auth=None,
)
device_info = DeviceInfo(device)
assert device_info.device_name == "My Doorbell"
assert device_info.device_model is None
assert device_info.device_brand == "Google Nest"
assert device_info.device_info == {
"identifiers": {("nest", "some-device-id")},
"name": "My Doorbell",
"manufacturer": "Google Nest",
"model": None,
}

109
tests/components/nest/sensor_sdm_test.py
View file

@ -5,108 +5,19 @@ These tests fake out the subscriber/devicemanager, and are not using a real
pubsub subscriber.
"""
import time
from google_nest_sdm.device import Device
from google_nest_sdm.device_manager import DeviceManager
from google_nest_sdm.event import EventCallback, EventMessage
from google_nest_sdm.google_nest_subscriber import GoogleNestSubscriber
from google_nest_sdm.event import EventMessage
from homeassistant.components.nest import DOMAIN
from homeassistant.setup import async_setup_component
from tests.async_mock import patch
from tests.common import MockConfigEntry
from .common import async_setup_sdm_platform
PLATFORM = "sensor"
CONFIG = {
"nest": {
"client_id": "some-client-id",
"client_secret": "some-client-secret",
# Required fields for using SDM API
"project_id": "some-project-id",
"subscriber_id": "some-subscriber-id",
},
}
CONFIG_ENTRY_DATA = {
"sdm": {}, # Indicates new SDM API, not legacy API
"auth_implementation": "local",
"token": {
"expires_at": time.time() + 86400,
"access_token": {
"token": "some-token",
},
},
}
THERMOSTAT_TYPE = "sdm.devices.types.THERMOSTAT"
class FakeDeviceManager(DeviceManager):
"""Fake DeviceManager that can supply a list of devices and structures."""
def __init__(self, devices: dict, structures: dict):
"""Initialize FakeDeviceManager."""
super().__init__()
self._devices = devices
@property
def structures(self) -> dict:
"""Override structures with fake result."""
return self._structures
@property
def devices(self) -> dict:
"""Override devices with fake result."""
return self._devices
class FakeSubscriber(GoogleNestSubscriber):
"""Fake subscriber that supplies a FakeDeviceManager."""
def __init__(self, device_manager: FakeDeviceManager):
"""Initialize Fake Subscriber."""
self._device_manager = device_manager
self._callback = None
def set_update_callback(self, callback: EventCallback):
"""Capture the callback set by Home Assistant."""
self._callback = callback
async def start_async(self) -> DeviceManager:
"""Return the fake device manager."""
return self._device_manager
async def async_get_device_manager(self) -> DeviceManager:
"""Return the fake device manager."""
return self._device_manager
def stop_async(self):
"""No-op to stop the subscriber."""
return None
def receive_event(self, event_message: EventMessage):
"""Simulate a received pubsub message, invoked by tests."""
# Update device state, then invoke HomeAssistant to refresh
self._device_manager.handle_event(event_message)
self._callback.handle_event(event_message)
async def setup_sensor(hass, devices={}, structures={}):
async def async_setup_sensor(hass, devices={}, structures={}):
"""Set up the platform and prerequisites."""
MockConfigEntry(domain=DOMAIN, data=CONFIG_ENTRY_DATA).add_to_hass(hass)
device_manager = FakeDeviceManager(devices=devices, structures=structures)
subscriber = FakeSubscriber(device_manager)
with patch(
"homeassistant.helpers.config_entry_oauth2_flow.async_get_config_entry_implementation"
), patch("homeassistant.components.nest.PLATFORMS", [PLATFORM]), patch(
"homeassistant.components.nest.GoogleNestSubscriber", return_value=subscriber
):
assert await async_setup_component(hass, DOMAIN, CONFIG)
await hass.async_block_till_done()
return subscriber
return await async_setup_sdm_platform(hass, PLATFORM, devices, structures)
async def test_thermostat_device(hass):
@ -131,7 +42,7 @@ async def test_thermostat_device(hass):
auth=None,
)
}
await setup_sensor(hass, devices)
await async_setup_sensor(hass, devices)
temperature = hass.states.get("sensor.my_sensor_temperature")
assert temperature is not None
@ -156,7 +67,7 @@ async def test_thermostat_device(hass):
async def test_no_devices(hass):
"""Test no devices returned by the api."""
await setup_sensor(hass)
await async_setup_sensor(hass)
temperature = hass.states.get("sensor.my_sensor_temperature")
assert temperature is None
@ -177,7 +88,7 @@ async def test_device_no_sensor_traits(hass):
auth=None,
)
}
await setup_sensor(hass, devices)
await async_setup_sensor(hass, devices)
temperature = hass.states.get("sensor.my_sensor_temperature")
assert temperature is None
@ -205,7 +116,7 @@ async def test_device_name_from_structure(hass):
auth=None,
)
}
await setup_sensor(hass, devices)
await async_setup_sensor(hass, devices)
temperature = hass.states.get("sensor.some_room_temperature")
assert temperature is not None
@ -231,7 +142,7 @@ async def test_event_updates_sensor(hass):
auth=None,
)
}
subscriber = await setup_sensor(hass, devices)
subscriber = await async_setup_sensor(hass, devices)
temperature = hass.states.get("sensor.my_sensor_temperature")
assert temperature is not None
@ -280,7 +191,7 @@ async def test_device_with_unknown_type(hass):
auth=None,
)
}
await setup_sensor(hass, devices)
await async_setup_sensor(hass, devices)
temperature = hass.states.get("sensor.my_sensor_temperature")
assert temperature is not None