Home-Assistant
/
core
mirror of https://github.com/home-assistant/core.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Add config flow to directv (#32162) 

* initial work on config flow.

* more work on config flow.

* work on config flow and add tests. other cleanup.

* cleanup tests.

* fix test.

* isort

* Update .coveragerc

* Update test_init.py

* Update test_init.py

* Update test_init.py

* Update test_config_flow.py

* Update test_config_flow.py

* Update test_config_flow.py

* correct upnp serial format.

* improve config flow coverage.

* review tweaks.

* further review tweaks

* simplify dtv data gathering job

* lint

* black

* Update test_init.py

* Update test_init.py

* Simplify exception handling.

* Simplify exception handling.

* Update media_player.py

* Update test_media_player.py

* Update test_media_player.py

* Update test_media_player.py

* Update test_media_player.py

* Update test_media_player.py

* fix failing test.

* restore change made during debug.

* isort.

Co-authored-by: Paulus Schoutsen <paulus@home-assistant.io>
pull/32700/head
Chris Talkington 2020-03-11 14:28:38 -05:00 committed by GitHub
parent c9a9bd16fe
commit bb666b9ac6
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
14 changed files with 1062 additions and 492 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
.coveragerc
View File

@ -149,7 +149,6 @@ omit =
homeassistant/components/dht/sensor.py
homeassistant/components/digital_ocean/*
homeassistant/components/digitalloggers/switch.py
homeassistant/components/directv/media_player.py
homeassistant/components/discogs/sensor.py
homeassistant/components/discord/notify.py
homeassistant/components/dlib_face_detect/image_processing.py

26
homeassistant/components/directv/.translations/en.json Normal file
View File

@ -0,0 +1,26 @@
{
"config": {
"abort": {
"already_configured": "DirecTV receiver is already configured"
},
"error": {
"cannot_connect": "Failed to connect, please try again",
"unknown": "Unexpected error"
},
"flow_title": "DirecTV: {name}",
"step": {
"ssdp_confirm": {
"data": {},
"description": "Do you want to set up {name}?",
"title": "Connect to the DirecTV receiver"
},
"user": {
"data": {
"host": "Host or IP address"
},
"title": "Connect to the DirecTV receiver"
}
},
"title": "DirecTV"
}
}

95
homeassistant/components/directv/__init__.py
View File

@ -1 +1,94 @@
"""The directv component."""
"""The DirecTV integration."""
import asyncio
from datetime import timedelta
from typing import Dict
from DirectPy import DIRECTV
from requests.exceptions import RequestException
import voluptuous as vol
from homeassistant.config_entries import SOURCE_IMPORT, ConfigEntry
from homeassistant.const import CONF_HOST
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryNotReady
from homeassistant.helpers import config_validation as cv
from .const import DATA_CLIENT, DATA_LOCATIONS, DATA_VERSION_INFO, DEFAULT_PORT, DOMAIN
CONFIG_SCHEMA = vol.Schema(
{
DOMAIN: vol.All(
cv.ensure_list, [vol.Schema({vol.Required(CONF_HOST): cv.string})]
)
},
extra=vol.ALLOW_EXTRA,
)
PLATFORMS = ["media_player"]
SCAN_INTERVAL = timedelta(seconds=30)
def get_dtv_data(
hass: HomeAssistant, host: str, port: int = DEFAULT_PORT, client_addr: str = "0"
) -> dict:
"""Retrieve a DIRECTV instance, locations list, and version info for the receiver device."""
dtv = DIRECTV(host, port, client_addr)
locations = dtv.get_locations()
version_info = dtv.get_version()
return {
DATA_CLIENT: dtv,
DATA_LOCATIONS: locations,
DATA_VERSION_INFO: version_info,
}
async def async_setup(hass: HomeAssistant, config: Dict) -> bool:
"""Set up the DirecTV component."""
hass.data.setdefault(DOMAIN, {})
if DOMAIN in config:
for entry_config in config[DOMAIN]:
hass.async_create_task(
hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_IMPORT}, data=entry_config,
)
)
return True
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Set up DirecTV from a config entry."""
try:
dtv_data = await hass.async_add_executor_job(
get_dtv_data, hass, entry.data[CONF_HOST]
)
except RequestException:
raise ConfigEntryNotReady
hass.data[DOMAIN][entry.entry_id] = dtv_data
for component in PLATFORMS:
hass.async_create_task(
hass.config_entries.async_forward_entry_setup(entry, component)
)
return True
async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Unload a config entry."""
unload_ok = all(
await asyncio.gather(
*[
hass.config_entries.async_forward_entry_unload(entry, component)
for component in PLATFORMS
]
)
)
if unload_ok:
hass.data[DOMAIN].pop(entry.entry_id)
return unload_ok

130
homeassistant/components/directv/config_flow.py Normal file
View File

@ -0,0 +1,130 @@
"""Config flow for DirecTV."""
import logging
from typing import Any, Dict, Optional
from urllib.parse import urlparse
from DirectPy import DIRECTV
from requests.exceptions import RequestException
import voluptuous as vol
from homeassistant.components.ssdp import ATTR_SSDP_LOCATION, ATTR_UPNP_SERIAL
from homeassistant.config_entries import CONN_CLASS_LOCAL_POLL, ConfigFlow
from homeassistant.const import CONF_HOST, CONF_NAME
from homeassistant.core import callback
from homeassistant.exceptions import HomeAssistantError
from .const import DEFAULT_PORT
from .const import DOMAIN # pylint: disable=unused-import
_LOGGER = logging.getLogger(__name__)
ERROR_CANNOT_CONNECT = "cannot_connect"
ERROR_UNKNOWN = "unknown"
DATA_SCHEMA = vol.Schema({vol.Required(CONF_HOST): str})
def validate_input(data: Dict) -> Dict:
"""Validate the user input allows us to connect.
Data has the keys from DATA_SCHEMA with values provided by the user.
"""
# directpy does IO in constructor.
dtv = DIRECTV(data["host"], DEFAULT_PORT)
version_info = dtv.get_version()
return {
"title": data["host"],
"host": data["host"],
"receiver_id": "".join(version_info["receiverId"].split()),
}
class DirecTVConfigFlow(ConfigFlow, domain=DOMAIN):
"""Handle a config flow for DirecTV."""
VERSION = 1
CONNECTION_CLASS = CONN_CLASS_LOCAL_POLL
@callback
def _show_form(self, errors: Optional[Dict] = None) -> Dict[str, Any]:
"""Show the form to the user."""
return self.async_show_form(
step_id="user", data_schema=DATA_SCHEMA, errors=errors or {},
)
async def async_step_import(
self, user_input: Optional[Dict] = None
) -> Dict[str, Any]:
"""Handle a flow initialized by yaml file."""
return await self.async_step_user(user_input)
async def async_step_user(
self, user_input: Optional[Dict] = None
) -> Dict[str, Any]:
"""Handle a flow initialized by user."""
if not user_input:
return self._show_form()
errors = {}
try:
info = await self.hass.async_add_executor_job(validate_input, user_input)
user_input[CONF_HOST] = info[CONF_HOST]
except RequestException:
errors["base"] = ERROR_CANNOT_CONNECT
return self._show_form(errors)
except Exception: # pylint: disable=broad-except
_LOGGER.exception("Unexpected exception")
errors["base"] = ERROR_UNKNOWN
return self._show_form(errors)
await self.async_set_unique_id(info["receiver_id"])
self._abort_if_unique_id_configured()
return self.async_create_entry(title=info["title"], data=user_input)
async def async_step_ssdp(
self, discovery_info: Optional[Dict] = None
) -> Dict[str, Any]:
"""Handle a flow initialized by discovery."""
host = urlparse(discovery_info[ATTR_SSDP_LOCATION]).hostname
receiver_id = discovery_info[ATTR_UPNP_SERIAL][4:] # strips off RID-
await self.async_set_unique_id(receiver_id)
self._abort_if_unique_id_configured(updates={CONF_HOST: host})
# pylint: disable=no-member # https://github.com/PyCQA/pylint/issues/3167
self.context.update(
{CONF_HOST: host, CONF_NAME: host, "title_placeholders": {"name": host}}
)
return await self.async_step_ssdp_confirm()
async def async_step_ssdp_confirm(
self, user_input: Optional[Dict] = None
) -> Dict[str, Any]:
"""Handle user-confirmation of discovered device."""
# pylint: disable=no-member # https://github.com/PyCQA/pylint/issues/3167
name = self.context.get(CONF_NAME)
if user_input is not None:
# pylint: disable=no-member # https://github.com/PyCQA/pylint/issues/3167
user_input[CONF_HOST] = self.context.get(CONF_HOST)
try:
await self.hass.async_add_executor_job(validate_input, user_input)
return self.async_create_entry(title=name, data=user_input)
except (OSError, RequestException):
return self.async_abort(reason=ERROR_CANNOT_CONNECT)
except Exception: # pylint: disable=broad-except
_LOGGER.exception("Unexpected exception")
return self.async_abort(reason=ERROR_UNKNOWN)
return self.async_show_form(
step_id="ssdp_confirm", description_placeholders={"name": name},
)
class CannotConnect(HomeAssistantError):
"""Error to indicate we cannot connect."""

10
homeassistant/components/directv/const.py
View File

@ -1,12 +1,20 @@
"""Constants for the DirecTV integration."""
DOMAIN = "directv"
ATTR_MEDIA_CURRENTLY_RECORDING = "media_currently_recording"
ATTR_MEDIA_RATING = "media_rating"
ATTR_MEDIA_RECORDED = "media_recorded"
ATTR_MEDIA_START_TIME = "media_start_time"
DATA_DIRECTV = "data_directv"
DATA_CLIENT = "client"
DATA_LOCATIONS = "locations"
DATA_VERSION_INFO = "version_info"
DEFAULT_DEVICE = "0"
DEFAULT_MANUFACTURER = "DirecTV"
DEFAULT_NAME = "DirecTV Receiver"
DEFAULT_PORT = 8080
MODEL_HOST = "DirecTV Host"
MODEL_CLIENT = "DirecTV Client"

9
homeassistant/components/directv/manifest.json
View File

@ -4,5 +4,12 @@
"documentation": "https://www.home-assistant.io/integrations/directv",
"requirements": ["directpy==0.6"],
"dependencies": [],
"codeowners": ["@ctalkington"]
"codeowners": ["@ctalkington"],
"config_flow": true,
"ssdp": [
{
"manufacturer": "DIRECTV",
"deviceType": "urn:schemas-upnp-org:device:MediaServer:1"
}
]
}

205
homeassistant/components/directv/media_player.py
View File

@ -1,8 +1,9 @@
"""Support for the DirecTV receivers."""
import logging
from typing import Callable, Dict, List, Optional
from DirectPy import DIRECTV
import requests
from requests.exceptions import RequestException
import voluptuous as vol
from homeassistant.components.media_player import PLATFORM_SCHEMA, MediaPlayerDevice
@ -19,6 +20,7 @@ from homeassistant.components.media_player.const import (
SUPPORT_TURN_OFF,
SUPPORT_TURN_ON,
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import (
CONF_DEVICE,
CONF_HOST,
@ -28,18 +30,25 @@ from homeassistant.const import (
STATE_PAUSED,
STATE_PLAYING,
)
import homeassistant.helpers.config_validation as cv
import homeassistant.util.dt as dt_util
from homeassistant.helpers import config_validation as cv
from homeassistant.helpers.typing import HomeAssistantType
from homeassistant.util import dt as dt_util
from .const import (
ATTR_MEDIA_CURRENTLY_RECORDING,
ATTR_MEDIA_RATING,
ATTR_MEDIA_RECORDED,
ATTR_MEDIA_START_TIME,
DATA_DIRECTV,
DATA_CLIENT,
DATA_LOCATIONS,
DATA_VERSION_INFO,
DEFAULT_DEVICE,
DEFAULT_MANUFACTURER,
DEFAULT_NAME,
DEFAULT_PORT,
DOMAIN,
MODEL_CLIENT,
MODEL_HOST,
)
_LOGGER = logging.getLogger(__name__)
@ -74,97 +83,67 @@ PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend(
)
def setup_platform(hass, config, add_entities, discovery_info=None):
"""Set up the DirecTV platform."""
known_devices = hass.data.get(DATA_DIRECTV, set())
def get_dtv_instance(
host: str, port: int = DEFAULT_PORT, client_addr: str = "0"
) -> DIRECTV:
"""Retrieve a DIRECTV instance for the receiver or client device."""
try:
return DIRECTV(host, port, client_addr)
except RequestException as exception:
_LOGGER.debug(
"Request exception %s trying to retrieve DIRECTV instance for client address %s on device %s",
exception,
client_addr,
host,
)
return None
async def async_setup_entry(
hass: HomeAssistantType,
entry: ConfigEntry,
async_add_entities: Callable[[List, bool], None],
) -> bool:
"""Set up the DirecTV config entry."""
locations = hass.data[DOMAIN][entry.entry_id][DATA_LOCATIONS]
version_info = hass.data[DOMAIN][entry.entry_id][DATA_VERSION_INFO]
entities = []
if CONF_HOST in config:
name = config[CONF_NAME]
host = config[CONF_HOST]
port = config[CONF_PORT]
device = config[CONF_DEVICE]
for loc in locations["locations"]:
if "locationName" not in loc or "clientAddr" not in loc:
continue
_LOGGER.debug(
"Adding configured device %s with client address %s", name, device,
if loc["clientAddr"] != "0":
# directpy does IO in constructor.
dtv = await hass.async_add_executor_job(
get_dtv_instance, entry.data[CONF_HOST], DEFAULT_PORT, loc["clientAddr"]
)
else:
dtv = hass.data[DOMAIN][entry.entry_id][DATA_CLIENT]
if not dtv:
continue
entities.append(
DirecTvDevice(
str.title(loc["locationName"]), loc["clientAddr"], dtv, version_info,
)
)
dtv = DIRECTV(host, port, device)
dtv_version = _get_receiver_version(dtv)
entities.append(DirecTvDevice(name, device, dtv, dtv_version,))
known_devices.add((host, device))
elif discovery_info:
host = discovery_info.get("host")
name = f"DirecTV_{discovery_info.get('serial', '')}"
# Attempt to discover additional RVU units
_LOGGER.debug("Doing discovery of DirecTV devices on %s", host)
dtv = DIRECTV(host, DEFAULT_PORT)
try:
dtv_version = _get_receiver_version(dtv)
resp = dtv.get_locations()
except requests.exceptions.RequestException as ex:
# Bail out and just go forward with uPnP data
# Make sure that this device is not already configured
# Comparing based on host (IP) and clientAddr.
_LOGGER.debug("Request exception %s trying to get locations", ex)
resp = {"locations": [{"locationName": name, "clientAddr": DEFAULT_DEVICE}]}
_LOGGER.debug("Known devices: %s", known_devices)
for loc in resp.get("locations") or []:
if "locationName" not in loc or "clientAddr" not in loc:
continue
loc_name = str.title(loc["locationName"])
# Make sure that this device is not already configured
# Comparing based on host (IP) and clientAddr.
if (host, loc["clientAddr"]) in known_devices:
_LOGGER.debug(
"Discovered device %s on host %s with "
"client address %s is already "
"configured",
loc_name,
host,
loc["clientAddr"],
)
else:
_LOGGER.debug(
"Adding discovered device %s with client address %s",
loc_name,
loc["clientAddr"],
)
entities.append(
DirecTvDevice(
loc_name,
loc["clientAddr"],
DIRECTV(host, DEFAULT_PORT, loc["clientAddr"]),
dtv_version,
)
)
known_devices.add((host, loc["clientAddr"]))
add_entities(entities)
def _get_receiver_version(client):
"""Return the version of the DirectTV receiver."""
try:
return client.get_version()
except requests.exceptions.RequestException as ex:
_LOGGER.debug("Request exception %s trying to get receiver version", ex)
return None
async_add_entities(entities, True)
class DirecTvDevice(MediaPlayerDevice):
"""Representation of a DirecTV receiver on the network."""
def __init__(self, name, device, dtv, version_info=None):
def __init__(
self,
name: str,
device: str,
dtv: DIRECTV,
version_info: Optional[Dict] = None,
enabled_default: bool = True,
):
"""Initialize the device."""
self.dtv = dtv
self._name = name
@ -178,17 +157,32 @@ class DirecTvDevice(MediaPlayerDevice):
self._is_client = device != "0"
self._assumed_state = None
self._available = False
self._enabled_default = enabled_default
self._first_error_timestamp = None
if device != "0":
self._unique_id = device
elif version_info:
self._unique_id = "".join(version_info.get("receiverId").split())
self._model = None
self._receiver_id = None
self._software_version = None
if self._is_client:
_LOGGER.debug("Created DirecTV client %s for device %s", self._name, device)
self._model = MODEL_CLIENT
self._unique_id = device
if version_info:
self._receiver_id = "".join(version_info["receiverId"].split())
if not self._is_client:
self._unique_id = self._receiver_id
self._model = MODEL_HOST
self._software_version = version_info["stbSoftwareVersion"]
if self._is_client:
_LOGGER.debug(
"Created DirecTV media player for client %s on device %s",
self._name,
device,
)
else:
_LOGGER.debug("Created DirecTV device for %s", self._name)
_LOGGER.debug("Created DirecTV media player for device %s", self._name)
def update(self):
"""Retrieve latest state."""
@ -225,17 +219,19 @@ class DirecTvDevice(MediaPlayerDevice):
else:
_LOGGER.error(log_message)
except requests.RequestException as ex:
except RequestException as exception:
_LOGGER.error(
"%s: Request error trying to update current status: %s",
self.entity_id,
ex,
exception,
)
self._check_state_available()
except Exception as ex:
except Exception as exception:
_LOGGER.error(
"%s: Exception trying to update current status: %s", self.entity_id, ex
"%s: Exception trying to update current status: %s",
self.entity_id,
exception,
)
self._available = False
if not self._first_error_timestamp:
@ -275,6 +271,23 @@ class DirecTvDevice(MediaPlayerDevice):
"""Return a unique ID to use for this media player."""
return self._unique_id
@property
def device_info(self):
"""Return device specific attributes."""
return {
"name": self.name,
"identifiers": {(DOMAIN, self.unique_id)},
"manufacturer": DEFAULT_MANUFACTURER,
"model": self._model,
"sw_version": self._software_version,
"via_device": (DOMAIN, self._receiver_id),
}
@property
def entity_registry_enabled_default(self) -> bool:
"""Return if the entity should be enabled when first added to the entity registry."""
return self._enabled_default
# MediaPlayerDevice properties and methods
@property
def state(self):

26
homeassistant/components/directv/strings.json Normal file
View File

@ -0,0 +1,26 @@
{
"config": {
"title": "DirecTV",
"flow_title": "DirecTV: {name}",
"step": {
"ssdp_confirm": {
"data": {},
"description": "Do you want to set up {name}?",
"title": "Connect to the DirecTV receiver"
},
"user": {
"title": "Connect to the DirecTV receiver",
"data": {
"host": "Host or IP address"
}
}
},
"error": {
"cannot_connect": "Failed to connect, please try again",
"unknown": "Unexpected error"
},
"abort": {
"already_configured": "DirecTV receiver is already configured"
}
}
}

1
homeassistant/generated/config_flows.py
View File

@ -23,6 +23,7 @@ FLOWS = [
"daikin",
"deconz",
"dialogflow",
"directv",
"dynalite",
"ecobee",
"elgato",

6
homeassistant/generated/ssdp.py
View File

@ -11,6 +11,12 @@ SSDP = {
"manufacturer": "Royal Philips Electronics"
}
],
"directv": [
{
"deviceType": "urn:schemas-upnp-org:device:MediaServer:1",
"manufacturer": "DIRECTV"
}
],
"heos": [
{
"st": "urn:schemas-denon-com:device:ACT-Denon:1"

183
tests/components/directv/__init__.py
View File

@ -1 +1,182 @@
"""Tests for the directv component."""
"""Tests for the DirecTV component."""
from homeassistant.components.directv.const import DOMAIN
from homeassistant.const import CONF_HOST
from homeassistant.helpers.typing import HomeAssistantType
from tests.common import MockConfigEntry
CLIENT_NAME = "Bedroom Client"
CLIENT_ADDRESS = "2CA17D1CD30X"
DEFAULT_DEVICE = "0"
HOST = "127.0.0.1"
MAIN_NAME = "Main DVR"
RECEIVER_ID = "028877455858"
SSDP_LOCATION = "http://127.0.0.1/"
UPNP_SERIAL = "RID-028877455858"
LIVE = {
"callsign": "HASSTV",
"date": "20181110",
"duration": 3600,
"isOffAir": False,
"isPclocked": 1,
"isPpv": False,
"isRecording": False,
"isVod": False,
"major": 202,
"minor": 65535,
"offset": 1,
"programId": "102454523",
"rating": "No Rating",
"startTime": 1541876400,
"stationId": 3900947,
"title": "Using Home Assistant to automate your home",
}
RECORDING = {
"callsign": "HASSTV",
"date": "20181110",
"duration": 3600,
"isOffAir": False,
"isPclocked": 1,
"isPpv": False,
"isRecording": True,
"isVod": False,
"major": 202,
"minor": 65535,
"offset": 1,
"programId": "102454523",
"rating": "No Rating",
"startTime": 1541876400,
"stationId": 3900947,
"title": "Using Home Assistant to automate your home",
"uniqueId": "12345",
"episodeTitle": "Configure DirecTV platform.",
}
MOCK_CONFIG = {DOMAIN: [{CONF_HOST: HOST}]}
MOCK_GET_LOCATIONS = {
"locations": [{"locationName": MAIN_NAME, "clientAddr": DEFAULT_DEVICE}],
"status": {
"code": 200,
"commandResult": 0,
"msg": "OK.",
"query": "/info/getLocations",
},
}
MOCK_GET_LOCATIONS_MULTIPLE = {
"locations": [
{"locationName": MAIN_NAME, "clientAddr": DEFAULT_DEVICE},
{"locationName": CLIENT_NAME, "clientAddr": CLIENT_ADDRESS},
],
"status": {
"code": 200,
"commandResult": 0,
"msg": "OK.",
"query": "/info/getLocations",
},
}
MOCK_GET_VERSION = {
"accessCardId": "0021-1495-6572",
"receiverId": "0288 7745 5858",
"status": {
"code": 200,
"commandResult": 0,
"msg": "OK.",
"query": "/info/getVersion",
},
"stbSoftwareVersion": "0x4ed7",
"systemTime": 1281625203,
"version": "1.2",
}
class MockDirectvClass:
"""A fake DirecTV DVR device."""
def __init__(self, ip, port=8080, clientAddr="0"):
"""Initialize the fake DirecTV device."""
self._host = ip
self._port = port
self._device = clientAddr
self._standby = True
self._play = False
self.attributes = LIVE
def get_locations(self):
"""Mock for get_locations method."""
return MOCK_GET_LOCATIONS
def get_serial_num(self):
"""Mock for get_serial_num method."""
test_serial_num = {
"serialNum": "9999999999",
"status": {
"code": 200,
"commandResult": 0,
"msg": "OK.",
"query": "/info/getSerialNum",
},
}
return test_serial_num
def get_standby(self):
"""Mock for get_standby method."""
return self._standby
def get_tuned(self):
"""Mock for get_tuned method."""
if self._play:
self.attributes["offset"] = self.attributes["offset"] + 1
test_attributes = self.attributes
test_attributes["status"] = {
"code": 200,
"commandResult": 0,
"msg": "OK.",
"query": "/tv/getTuned",
}
return test_attributes
def get_version(self):
"""Mock for get_version method."""
return MOCK_GET_VERSION
def key_press(self, keypress):
"""Mock for key_press method."""
if keypress == "poweron":
self._standby = False
self._play = True
elif keypress == "poweroff":
self._standby = True
self._play = False
elif keypress == "play":
self._play = True
elif keypress == "pause" or keypress == "stop":
self._play = False
def tune_channel(self, source):
"""Mock for tune_channel method."""
self.attributes["major"] = int(source)
async def setup_integration(
hass: HomeAssistantType, skip_entry_setup: bool = False
) -> MockConfigEntry:
"""Set up the DirecTV integration in Home Assistant."""
entry = MockConfigEntry(
domain=DOMAIN, unique_id=RECEIVER_ID, data={CONF_HOST: HOST}
)
entry.add_to_hass(hass)
if not skip_entry_setup:
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
return entry

240
tests/components/directv/test_config_flow.py Normal file
View File

@ -0,0 +1,240 @@
"""Test the DirecTV config flow."""
from typing import Any, Dict, Optional
from asynctest import patch
from requests.exceptions import RequestException
from homeassistant.components.directv.const import DOMAIN
from homeassistant.components.ssdp import ATTR_SSDP_LOCATION, ATTR_UPNP_SERIAL
from homeassistant.config_entries import SOURCE_IMPORT, SOURCE_SSDP, SOURCE_USER
from homeassistant.const import CONF_HOST, CONF_NAME, CONF_SOURCE
from homeassistant.data_entry_flow import (
RESULT_TYPE_ABORT,
RESULT_TYPE_CREATE_ENTRY,
RESULT_TYPE_FORM,
)
from homeassistant.helpers.typing import HomeAssistantType
from homeassistant.setup import async_setup_component
from tests.common import MockConfigEntry
from tests.components.directv import (
HOST,
RECEIVER_ID,
SSDP_LOCATION,
UPNP_SERIAL,
MockDirectvClass,
)
async def async_configure_flow(
hass: HomeAssistantType, flow_id: str, user_input: Optional[Dict] = None
) -> Any:
"""Set up mock DirecTV integration flow."""
with patch(
"homeassistant.components.directv.config_flow.DIRECTV", new=MockDirectvClass,
):
return await hass.config_entries.flow.async_configure(
flow_id=flow_id, user_input=user_input
)
async def async_init_flow(
hass: HomeAssistantType,
handler: str = DOMAIN,
context: Optional[Dict] = None,
data: Any = None,
) -> Any:
"""Set up mock DirecTV integration flow."""
with patch(
"homeassistant.components.directv.config_flow.DIRECTV", new=MockDirectvClass,
):
return await hass.config_entries.flow.async_init(
handler=handler, context=context, data=data
)
async def test_duplicate_error(hass: HomeAssistantType) -> None:
"""Test that errors are shown when duplicates are added."""
MockConfigEntry(
domain=DOMAIN, unique_id=RECEIVER_ID, data={CONF_HOST: HOST}
).add_to_hass(hass)
result = await async_init_flow(
hass, context={CONF_SOURCE: SOURCE_IMPORT}, data={CONF_HOST: HOST}
)
assert result["type"] == RESULT_TYPE_ABORT
assert result["reason"] == "already_configured"
result = await async_init_flow(
hass, context={CONF_SOURCE: SOURCE_USER}, data={CONF_HOST: HOST}
)
assert result["type"] == RESULT_TYPE_ABORT
assert result["reason"] == "already_configured"
result = await async_init_flow(
hass,
context={CONF_SOURCE: SOURCE_SSDP},
data={ATTR_SSDP_LOCATION: SSDP_LOCATION, ATTR_UPNP_SERIAL: UPNP_SERIAL},
)
assert result["type"] == RESULT_TYPE_ABORT
assert result["reason"] == "already_configured"
async def test_form(hass: HomeAssistantType) -> None:
"""Test we get the form."""
await async_setup_component(hass, "persistent_notification", {})
result = await hass.config_entries.flow.async_init(
DOMAIN, context={CONF_SOURCE: SOURCE_USER}
)
assert result["type"] == RESULT_TYPE_FORM
assert result["errors"] == {}
with patch(
"homeassistant.components.directv.async_setup", return_value=True
) as mock_setup, patch(
"homeassistant.components.directv.async_setup_entry", return_value=True,
) as mock_setup_entry:
result = await async_configure_flow(hass, result["flow_id"], {CONF_HOST: HOST})
assert result["type"] == RESULT_TYPE_CREATE_ENTRY
assert result["title"] == HOST
assert result["data"] == {CONF_HOST: HOST}
await hass.async_block_till_done()
assert len(mock_setup.mock_calls) == 1
assert len(mock_setup_entry.mock_calls) == 1
async def test_form_cannot_connect(hass: HomeAssistantType) -> None:
"""Test we handle cannot connect error."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={CONF_SOURCE: SOURCE_USER}
)
with patch(
"homeassistant.components.directv.config_flow.DIRECTV", new=MockDirectvClass,
), patch(
"homeassistant.components.directv.config_flow.DIRECTV.get_version",
side_effect=RequestException,
) as mock_validate_input:
result = await async_configure_flow(hass, result["flow_id"], {CONF_HOST: HOST},)
assert result["type"] == RESULT_TYPE_FORM
assert result["errors"] == {"base": "cannot_connect"}
await hass.async_block_till_done()
assert len(mock_validate_input.mock_calls) == 1
async def test_form_unknown_error(hass: HomeAssistantType) -> None:
"""Test we handle unknown error."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={CONF_SOURCE: SOURCE_USER}
)
with patch(
"homeassistant.components.directv.config_flow.DIRECTV", new=MockDirectvClass,
), patch(
"homeassistant.components.directv.config_flow.DIRECTV.get_version",
side_effect=Exception,
) as mock_validate_input:
result = await async_configure_flow(hass, result["flow_id"], {CONF_HOST: HOST},)
assert result["type"] == RESULT_TYPE_FORM
assert result["errors"] == {"base": "unknown"}
await hass.async_block_till_done()
assert len(mock_validate_input.mock_calls) == 1
async def test_import(hass: HomeAssistantType) -> None:
"""Test the import step."""
with patch(
"homeassistant.components.directv.async_setup", return_value=True
) as mock_setup, patch(
"homeassistant.components.directv.async_setup_entry", return_value=True,
) as mock_setup_entry:
result = await async_init_flow(
hass, context={CONF_SOURCE: SOURCE_IMPORT}, data={CONF_HOST: HOST},
)
assert result["type"] == RESULT_TYPE_CREATE_ENTRY
assert result["title"] == HOST
assert result["data"] == {CONF_HOST: HOST}
await hass.async_block_till_done()
assert len(mock_setup.mock_calls) == 1
assert len(mock_setup_entry.mock_calls) == 1
async def test_ssdp_discovery(hass: HomeAssistantType) -> None:
"""Test the ssdp discovery step."""
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={CONF_SOURCE: SOURCE_SSDP},
data={ATTR_SSDP_LOCATION: SSDP_LOCATION, ATTR_UPNP_SERIAL: UPNP_SERIAL},
)
assert result["type"] == RESULT_TYPE_FORM
assert result["step_id"] == "ssdp_confirm"
assert result["description_placeholders"] == {CONF_NAME: HOST}
with patch(
"homeassistant.components.directv.async_setup", return_value=True
) as mock_setup, patch(
"homeassistant.components.directv.async_setup_entry", return_value=True,
) as mock_setup_entry:
result = await async_configure_flow(hass, result["flow_id"], {})
assert result["type"] == RESULT_TYPE_CREATE_ENTRY
assert result["title"] == HOST
assert result["data"] == {CONF_HOST: HOST}
await hass.async_block_till_done()
assert len(mock_setup.mock_calls) == 1
assert len(mock_setup_entry.mock_calls) == 1
async def test_ssdp_discovery_confirm_abort(hass: HomeAssistantType) -> None:
"""Test we handle SSDP confirm cannot connect error."""
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={CONF_SOURCE: SOURCE_SSDP},
data={ATTR_SSDP_LOCATION: SSDP_LOCATION, ATTR_UPNP_SERIAL: UPNP_SERIAL},
)
with patch(
"homeassistant.components.directv.config_flow.DIRECTV", new=MockDirectvClass,
), patch(
"homeassistant.components.directv.config_flow.DIRECTV.get_version",
side_effect=RequestException,
) as mock_validate_input:
result = await async_configure_flow(hass, result["flow_id"], {})
assert result["type"] == RESULT_TYPE_ABORT
await hass.async_block_till_done()
assert len(mock_validate_input.mock_calls) == 1
async def test_ssdp_discovery_confirm_unknown_error(hass: HomeAssistantType) -> None:
"""Test we handle SSDP confirm unknown error."""
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={CONF_SOURCE: SOURCE_SSDP},
data={ATTR_SSDP_LOCATION: SSDP_LOCATION, ATTR_UPNP_SERIAL: UPNP_SERIAL},
)
with patch(
"homeassistant.components.directv.config_flow.DIRECTV", new=MockDirectvClass,
), patch(
"homeassistant.components.directv.config_flow.DIRECTV.get_version",
side_effect=Exception,
) as mock_validate_input:
result = await async_configure_flow(hass, result["flow_id"], {})
assert result["type"] == RESULT_TYPE_ABORT
await hass.async_block_till_done()
assert len(mock_validate_input.mock_calls) == 1

48
tests/components/directv/test_init.py Normal file
View File

@ -0,0 +1,48 @@
"""Tests for the Roku integration."""
from asynctest import patch
from requests.exceptions import RequestException
from homeassistant.components.directv.const import DOMAIN
from homeassistant.config_entries import (
ENTRY_STATE_LOADED,
ENTRY_STATE_NOT_LOADED,
ENTRY_STATE_SETUP_RETRY,
)
from homeassistant.helpers.typing import HomeAssistantType
from tests.components.directv import MockDirectvClass, setup_integration
# pylint: disable=redefined-outer-name
async def test_config_entry_not_ready(hass: HomeAssistantType) -> None:
"""Test the DirecTV configuration entry not ready."""
with patch(
"homeassistant.components.directv.DIRECTV", new=MockDirectvClass,
), patch(
"homeassistant.components.directv.DIRECTV.get_locations",
side_effect=RequestException,
):
entry = await setup_integration(hass)
assert entry.state == ENTRY_STATE_SETUP_RETRY
async def test_unload_config_entry(hass: HomeAssistantType) -> None:
"""Test the DirecTV configuration entry unloading."""
with patch(
"homeassistant.components.directv.DIRECTV", new=MockDirectvClass,
), patch(
"homeassistant.components.directv.media_player.async_setup_entry",
return_value=True,
):
entry = await setup_integration(hass)
assert entry.entry_id in hass.data[DOMAIN]
assert entry.state == ENTRY_STATE_LOADED
await hass.config_entries.async_unload(entry.entry_id)
await hass.async_block_till_done()
assert entry.entry_id not in hass.data[DOMAIN]
assert entry.state == ENTRY_STATE_NOT_LOADED

574
tests/components/directv/test_media_player.py
View File

@ -1,17 +1,16 @@
"""The tests for the DirecTV Media player platform."""
from datetime import datetime, timedelta
from unittest.mock import call, patch
from typing import Optional
import pytest
import requests
from asynctest import patch
from pytest import fixture
from requests import RequestException
from homeassistant.components.directv.media_player import (
ATTR_MEDIA_CURRENTLY_RECORDING,
ATTR_MEDIA_RATING,
ATTR_MEDIA_RECORDED,
ATTR_MEDIA_START_TIME,
DEFAULT_DEVICE,
DEFAULT_PORT,
)
from homeassistant.components.media_player.const import (
ATTR_INPUT_SOURCE,
@ -24,7 +23,7 @@ from homeassistant.components.media_player.const import (
ATTR_MEDIA_POSITION_UPDATED_AT,
ATTR_MEDIA_SERIES_TITLE,
ATTR_MEDIA_TITLE,
DOMAIN,
DOMAIN as MP_DOMAIN,
MEDIA_TYPE_TVSHOW,
SERVICE_PLAY_MEDIA,
SUPPORT_NEXT_TRACK,
@ -38,10 +37,6 @@ from homeassistant.components.media_player.const import (
)
from homeassistant.const import (
ATTR_ENTITY_ID,
CONF_DEVICE,
CONF_HOST,
CONF_NAME,
CONF_PORT,
SERVICE_MEDIA_NEXT_TRACK,
SERVICE_MEDIA_PAUSE,
SERVICE_MEDIA_PLAY,
@ -54,184 +49,143 @@ from homeassistant.const import (
STATE_PLAYING,
STATE_UNAVAILABLE,
)
from homeassistant.helpers.discovery import async_load_platform
from homeassistant.setup import async_setup_component
import homeassistant.util.dt as dt_util
from homeassistant.helpers.typing import HomeAssistantType
from homeassistant.util import dt as dt_util
from tests.common import async_fire_time_changed
from tests.common import MockConfigEntry, async_fire_time_changed
from tests.components.directv import (
CLIENT_ADDRESS,
DOMAIN,
HOST,
MOCK_GET_LOCATIONS_MULTIPLE,
RECORDING,
MockDirectvClass,
setup_integration,
)
ATTR_UNIQUE_ID = "unique_id"
CLIENT_ENTITY_ID = "media_player.client_dvr"
MAIN_ENTITY_ID = "media_player.main_dvr"
IP_ADDRESS = "127.0.0.1"
CLIENT_ENTITY_ID = f"{MP_DOMAIN}.bedroom_client"
MAIN_ENTITY_ID = f"{MP_DOMAIN}.main_dvr"
DISCOVERY_INFO = {"host": IP_ADDRESS, "serial": 1234}
LIVE = {
"callsign": "HASSTV",
"date": "20181110",
"duration": 3600,
"isOffAir": False,
"isPclocked": 1,
"isPpv": False,
"isRecording": False,
"isVod": False,
"major": 202,
"minor": 65535,
"offset": 1,
"programId": "102454523",
"rating": "No Rating",
"startTime": 1541876400,
"stationId": 3900947,
"title": "Using Home Assistant to automate your home",
}
LOCATIONS = [{"locationName": "Main DVR", "clientAddr": DEFAULT_DEVICE}]
RECORDING = {
"callsign": "HASSTV",
"date": "20181110",
"duration": 3600,
"isOffAir": False,
"isPclocked": 1,
"isPpv": False,
"isRecording": True,
"isVod": False,
"major": 202,
"minor": 65535,
"offset": 1,
"programId": "102454523",
"rating": "No Rating",
"startTime": 1541876400,
"stationId": 3900947,
"title": "Using Home Assistant to automate your home",
"uniqueId": "12345",
"episodeTitle": "Configure DirecTV platform.",
}
WORKING_CONFIG = {
"media_player": {
"platform": "directv",
CONF_HOST: IP_ADDRESS,
CONF_NAME: "Main DVR",
CONF_PORT: DEFAULT_PORT,
CONF_DEVICE: DEFAULT_DEVICE,
}
}
# pylint: disable=redefined-outer-name
@pytest.fixture
def client_dtv():
@fixture
def client_dtv() -> MockDirectvClass:
"""Fixture for a client device."""
mocked_dtv = MockDirectvClass("mock_ip")
mocked_dtv = MockDirectvClass(HOST, clientAddr=CLIENT_ADDRESS)
mocked_dtv.attributes = RECORDING
mocked_dtv._standby = False
mocked_dtv._standby = False # pylint: disable=protected-access
return mocked_dtv
@pytest.fixture
def main_dtv():
"""Fixture for main DVR."""
return MockDirectvClass("mock_ip")
@pytest.fixture
def dtv_side_effect(client_dtv, main_dtv):
"""Fixture to create DIRECTV instance for main and client."""
def mock_dtv(ip, port, client_addr="0"):
if client_addr != "0":
mocked_dtv = client_dtv
else:
mocked_dtv = main_dtv
mocked_dtv._host = ip
mocked_dtv._port = port
mocked_dtv._device = client_addr
return mocked_dtv
return mock_dtv
@pytest.fixture
def mock_now():
@fixture
def mock_now() -> datetime:
"""Fixture for dtutil.now."""
return dt_util.utcnow()
@pytest.fixture
def platforms(hass, dtv_side_effect, mock_now):
"""Fixture for setting up test platforms."""
config = {
"media_player": [
{
"platform": "directv",
"name": "Main DVR",
"host": IP_ADDRESS,
"port": DEFAULT_PORT,
"device": DEFAULT_DEVICE,
},
{
"platform": "directv",
"name": "Client DVR",
"host": IP_ADDRESS,
"port": DEFAULT_PORT,
"device": "2CA17D1CD30X",
},
]
}
async def setup_directv(hass: HomeAssistantType) -> MockConfigEntry:
"""Set up mock DirecTV integration."""
with patch(
"homeassistant.components.directv.media_player.DIRECTV",
side_effect=dtv_side_effect,
), patch("homeassistant.util.dt.utcnow", return_value=mock_now):
hass.loop.run_until_complete(async_setup_component(hass, DOMAIN, config))
hass.loop.run_until_complete(hass.async_block_till_done())
yield
"homeassistant.components.directv.DIRECTV", new=MockDirectvClass,
):
return await setup_integration(hass)
async def async_turn_on(hass, entity_id=None):
async def setup_directv_with_instance_error(hass: HomeAssistantType) -> MockConfigEntry:
"""Set up mock DirecTV integration."""
with patch(
"homeassistant.components.directv.DIRECTV", new=MockDirectvClass,
), patch(
"homeassistant.components.directv.DIRECTV.get_locations",
return_value=MOCK_GET_LOCATIONS_MULTIPLE,
), patch(
"homeassistant.components.directv.media_player.get_dtv_instance",
return_value=None,
):
return await setup_integration(hass)
async def setup_directv_with_locations(
hass: HomeAssistantType, client_dtv: MockDirectvClass,
) -> MockConfigEntry:
"""Set up mock DirecTV integration."""
with patch(
"homeassistant.components.directv.DIRECTV", new=MockDirectvClass,
), patch(
"homeassistant.components.directv.DIRECTV.get_locations",
return_value=MOCK_GET_LOCATIONS_MULTIPLE,
), patch(
"homeassistant.components.directv.media_player.get_dtv_instance",
return_value=client_dtv,
):
return await setup_integration(hass)
async def async_turn_on(
hass: HomeAssistantType, entity_id: Optional[str] = None
) -> None:
"""Turn on specified media player or all."""
data = {ATTR_ENTITY_ID: entity_id} if entity_id else {}
await hass.services.async_call(DOMAIN, SERVICE_TURN_ON, data)
await hass.services.async_call(MP_DOMAIN, SERVICE_TURN_ON, data)
async def async_turn_off(hass, entity_id=None):
async def async_turn_off(
hass: HomeAssistantType, entity_id: Optional[str] = None
) -> None:
"""Turn off specified media player or all."""
data = {ATTR_ENTITY_ID: entity_id} if entity_id else {}
await hass.services.async_call(DOMAIN, SERVICE_TURN_OFF, data)
await hass.services.async_call(MP_DOMAIN, SERVICE_TURN_OFF, data)
async def async_media_pause(hass, entity_id=None):
async def async_media_pause(
hass: HomeAssistantType, entity_id: Optional[str] = None
) -> None:
"""Send the media player the command for pause."""
data = {ATTR_ENTITY_ID: entity_id} if entity_id else {}
await hass.services.async_call(DOMAIN, SERVICE_MEDIA_PAUSE, data)
await hass.services.async_call(MP_DOMAIN, SERVICE_MEDIA_PAUSE, data)
async def async_media_play(hass, entity_id=None):
async def async_media_play(
hass: HomeAssistantType, entity_id: Optional[str] = None
) -> None:
"""Send the media player the command for play/pause."""
data = {ATTR_ENTITY_ID: entity_id} if entity_id else {}
await hass.services.async_call(DOMAIN, SERVICE_MEDIA_PLAY, data)
await hass.services.async_call(MP_DOMAIN, SERVICE_MEDIA_PLAY, data)
async def async_media_stop(hass, entity_id=None):
async def async_media_stop(
hass: HomeAssistantType, entity_id: Optional[str] = None
) -> None:
"""Send the media player the command for stop."""
data = {ATTR_ENTITY_ID: entity_id} if entity_id else {}
await hass.services.async_call(DOMAIN, SERVICE_MEDIA_STOP, data)
await hass.services.async_call(MP_DOMAIN, SERVICE_MEDIA_STOP, data)
async def async_media_next_track(hass, entity_id=None):
async def async_media_next_track(
hass: HomeAssistantType, entity_id: Optional[str] = None
) -> None:
"""Send the media player the command for next track."""
data = {ATTR_ENTITY_ID: entity_id} if entity_id else {}
await hass.services.async_call(DOMAIN, SERVICE_MEDIA_NEXT_TRACK, data)
await hass.services.async_call(MP_DOMAIN, SERVICE_MEDIA_NEXT_TRACK, data)
async def async_media_previous_track(hass, entity_id=None):
async def async_media_previous_track(
hass: HomeAssistantType, entity_id: Optional[str] = None
) -> None:
"""Send the media player the command for prev track."""
data = {ATTR_ENTITY_ID: entity_id} if entity_id else {}
await hass.services.async_call(DOMAIN, SERVICE_MEDIA_PREVIOUS_TRACK, data)
await hass.services.async_call(MP_DOMAIN, SERVICE_MEDIA_PREVIOUS_TRACK, data)
async def async_play_media(hass, media_type, media_id, entity_id=None, enqueue=None):
async def async_play_media(
hass: HomeAssistantType,
media_type: str,
media_id: str,
entity_id: Optional[str] = None,
enqueue: Optional[str] = None,
) -> None:
"""Send the media player the command for playing media."""
data = {ATTR_MEDIA_CONTENT_TYPE: media_type, ATTR_MEDIA_CONTENT_ID: media_id}
@ -241,191 +195,37 @@ async def async_play_media(hass, media_type, media_id, entity_id=None, enqueue=N
if enqueue:
data[ATTR_MEDIA_ENQUEUE] = enqueue
await hass.services.async_call(DOMAIN, SERVICE_PLAY_MEDIA, data)
await hass.services.async_call(MP_DOMAIN, SERVICE_PLAY_MEDIA, data)
class MockDirectvClass:
"""A fake DirecTV DVR device."""
def __init__(self, ip, port=8080, clientAddr="0"):
"""Initialize the fake DirecTV device."""
self._host = ip
self._port = port
self._device = clientAddr
self._standby = True
self._play = False
self._locations = LOCATIONS
self.attributes = LIVE
def get_locations(self):
"""Mock for get_locations method."""
test_locations = {
"locations": self._locations,
"status": {
"code": 200,
"commandResult": 0,
"msg": "OK.",
"query": "/info/getLocations",
},
}
return test_locations
def get_serial_num(self):
"""Mock for get_serial_num method."""
test_serial_num = {
"serialNum": "9999999999",
"status": {
"code": 200,
"commandResult": 0,
"msg": "OK.",
"query": "/info/getSerialNum",
},
}
return test_serial_num
def get_standby(self):
"""Mock for get_standby method."""
return self._standby
def get_tuned(self):
"""Mock for get_tuned method."""
if self._play:
self.attributes["offset"] = self.attributes["offset"] + 1
test_attributes = self.attributes
test_attributes["status"] = {
"code": 200,
"commandResult": 0,
"msg": "OK.",
"query": "/tv/getTuned",
}
return test_attributes
def get_version(self):
"""Mock for get_version method."""
test_version = {
"accessCardId": "0021-1495-6572",
"receiverId": "0288 7745 5858",
"status": {
"code": 200,
"commandResult": 0,
"msg": "OK.",
"query": "/info/getVersion",
},
"stbSoftwareVersion": "0x4ed7",
"systemTime": 1281625203,
"version": "1.2",
}
return test_version
def key_press(self, keypress):
"""Mock for key_press method."""
if keypress == "poweron":
self._standby = False
self._play = True
elif keypress == "poweroff":
self._standby = True
self._play = False
elif keypress == "play":
self._play = True
elif keypress == "pause" or keypress == "stop":
self._play = False
def tune_channel(self, source):
"""Mock for tune_channel method."""
self.attributes["major"] = int(source)
async def test_setup(hass: HomeAssistantType) -> None:
"""Test setup with basic config."""
await setup_directv(hass)
assert hass.states.get(MAIN_ENTITY_ID)
async def test_setup_platform_config(hass):
"""Test setting up the platform from configuration."""
with patch(
"homeassistant.components.directv.media_player.DIRECTV", new=MockDirectvClass
):
async def test_setup_with_multiple_locations(
hass: HomeAssistantType, client_dtv: MockDirectvClass
) -> None:
"""Test setup with basic config with client location."""
await setup_directv_with_locations(hass, client_dtv)
await async_setup_component(hass, DOMAIN, WORKING_CONFIG)
await hass.async_block_till_done()
state = hass.states.get(MAIN_ENTITY_ID)
assert state
assert len(hass.states.async_entity_ids("media_player")) == 1
assert hass.states.get(MAIN_ENTITY_ID)
assert hass.states.get(CLIENT_ENTITY_ID)
async def test_setup_platform_discover(hass):
"""Test setting up the platform from discovery."""
with patch(
"homeassistant.components.directv.media_player.DIRECTV", new=MockDirectvClass
):
async def test_setup_with_instance_error(hass: HomeAssistantType) -> None:
"""Test setup with basic config with client location that results in instance error."""
await setup_directv_with_instance_error(hass)
hass.async_create_task(
async_load_platform(
hass, DOMAIN, "directv", DISCOVERY_INFO, {"media_player": {}}
)
)
await hass.async_block_till_done()
state = hass.states.get(MAIN_ENTITY_ID)
assert state
assert len(hass.states.async_entity_ids("media_player")) == 1
assert hass.states.get(MAIN_ENTITY_ID)
assert hass.states.async_entity_ids(MP_DOMAIN) == [MAIN_ENTITY_ID]
async def test_setup_platform_discover_duplicate(hass):
"""Test setting up the platform from discovery."""
with patch(
"homeassistant.components.directv.media_player.DIRECTV", new=MockDirectvClass
):
await async_setup_component(hass, DOMAIN, WORKING_CONFIG)
await hass.async_block_till_done()
hass.async_create_task(
async_load_platform(
hass, DOMAIN, "directv", DISCOVERY_INFO, {"media_player": {}}
)
)
await hass.async_block_till_done()
state = hass.states.get(MAIN_ENTITY_ID)
assert state
assert len(hass.states.async_entity_ids("media_player")) == 1
async def test_setup_platform_discover_client(hass):
"""Test setting up the platform from discovery."""
LOCATIONS.append({"locationName": "Client 1", "clientAddr": "1"})
LOCATIONS.append({"locationName": "Client 2", "clientAddr": "2"})
with patch(
"homeassistant.components.directv.media_player.DIRECTV", new=MockDirectvClass
):
await async_setup_component(hass, DOMAIN, WORKING_CONFIG)
await hass.async_block_till_done()
hass.async_create_task(
async_load_platform(
hass, DOMAIN, "directv", DISCOVERY_INFO, {"media_player": {}}
)
)
await hass.async_block_till_done()
del LOCATIONS[-1]
del LOCATIONS[-1]
state = hass.states.get(MAIN_ENTITY_ID)
assert state
state = hass.states.get("media_player.client_1")
assert state
state = hass.states.get("media_player.client_2")
assert state
assert len(hass.states.async_entity_ids("media_player")) == 3
async def test_unique_id(hass, platforms):
async def test_unique_id(hass: HomeAssistantType, client_dtv: MockDirectvClass) -> None:
"""Test unique id."""
await setup_directv_with_locations(hass, client_dtv)
entity_registry = await hass.helpers.entity_registry.async_get_registry()
main = entity_registry.async_get(MAIN_ENTITY_ID)
@ -435,8 +235,12 @@ async def test_unique_id(hass, platforms):
assert client.unique_id == "2CA17D1CD30X"
async def test_supported_features(hass, platforms):
async def test_supported_features(
hass: HomeAssistantType, client_dtv: MockDirectvClass
) -> None:
"""Test supported features."""
await setup_directv_with_locations(hass, client_dtv)
# Features supported for main DVR
state = hass.states.get(MAIN_ENTITY_ID)
assert (
@ -464,8 +268,12 @@ async def test_supported_features(hass, platforms):
)
async def test_check_attributes(hass, platforms, mock_now):
async def test_check_attributes(
hass: HomeAssistantType, mock_now: dt_util.dt.datetime, client_dtv: MockDirectvClass
) -> None:
"""Test attributes."""
await setup_directv_with_locations(hass, client_dtv)
next_update = mock_now + timedelta(minutes=5)
with patch("homeassistant.util.dt.utcnow", return_value=next_update):
async_fire_time_changed(hass, next_update)
@ -512,8 +320,12 @@ async def test_check_attributes(hass, platforms, mock_now):
assert state.attributes.get(ATTR_MEDIA_POSITION_UPDATED_AT) == next_update
async def test_main_services(hass, platforms, main_dtv, mock_now):
async def test_main_services(
hass: HomeAssistantType, mock_now: dt_util.dt.datetime, client_dtv: MockDirectvClass
) -> None:
"""Test the different services."""
await setup_directv_with_locations(hass, client_dtv)
next_update = mock_now + timedelta(minutes=5)
with patch("homeassistant.util.dt.utcnow", return_value=next_update):
async_fire_time_changed(hass, next_update)
@ -522,77 +334,50 @@ async def test_main_services(hass, platforms, main_dtv, mock_now):
state = hass.states.get(MAIN_ENTITY_ID)
assert state.state == STATE_OFF
# All these should call key_press in our class.
with patch.object(
main_dtv, "key_press", wraps=main_dtv.key_press
) as mock_key_press, patch.object(
main_dtv, "tune_channel", wraps=main_dtv.tune_channel
) as mock_tune_channel, patch.object(
main_dtv, "get_tuned", wraps=main_dtv.get_tuned
) as mock_get_tuned, patch.object(
main_dtv, "get_standby", wraps=main_dtv.get_standby
) as mock_get_standby:
# Turn main DVR on. When turning on DVR is playing.
await async_turn_on(hass, MAIN_ENTITY_ID)
await hass.async_block_till_done()
state = hass.states.get(MAIN_ENTITY_ID)
assert state.state == STATE_PLAYING
# Turn main DVR on. When turning on DVR is playing.
await async_turn_on(hass, MAIN_ENTITY_ID)
await hass.async_block_till_done()
assert mock_key_press.called
assert mock_key_press.call_args == call("poweron")
state = hass.states.get(MAIN_ENTITY_ID)
assert state.state == STATE_PLAYING
# Pause live TV.
await async_media_pause(hass, MAIN_ENTITY_ID)
await hass.async_block_till_done()
state = hass.states.get(MAIN_ENTITY_ID)
assert state.state == STATE_PAUSED
# Pause live TV.
await async_media_pause(hass, MAIN_ENTITY_ID)
await hass.async_block_till_done()
assert mock_key_press.called
assert mock_key_press.call_args == call("pause")
state = hass.states.get(MAIN_ENTITY_ID)
assert state.state == STATE_PAUSED
# Start play again for live TV.
await async_media_play(hass, MAIN_ENTITY_ID)
await hass.async_block_till_done()
state = hass.states.get(MAIN_ENTITY_ID)
assert state.state == STATE_PLAYING
# Start play again for live TV.
await async_media_play(hass, MAIN_ENTITY_ID)
await hass.async_block_till_done()
assert mock_key_press.called
assert mock_key_press.call_args == call("play")
state = hass.states.get(MAIN_ENTITY_ID)
assert state.state == STATE_PLAYING
# Change channel, currently it should be 202
assert state.attributes.get("source") == 202
await async_play_media(hass, "channel", 7, MAIN_ENTITY_ID)
await hass.async_block_till_done()
state = hass.states.get(MAIN_ENTITY_ID)
assert state.attributes.get("source") == 7
# Change channel, currently it should be 202
assert state.attributes.get("source") == 202
await async_play_media(hass, "channel", 7, MAIN_ENTITY_ID)
await hass.async_block_till_done()
assert mock_tune_channel.called
assert mock_tune_channel.call_args == call("7")
state = hass.states.get(MAIN_ENTITY_ID)
assert state.attributes.get("source") == 7
# Stop live TV.
await async_media_stop(hass, MAIN_ENTITY_ID)
await hass.async_block_till_done()
state = hass.states.get(MAIN_ENTITY_ID)
assert state.state == STATE_PAUSED
# Stop live TV.
await async_media_stop(hass, MAIN_ENTITY_ID)
await hass.async_block_till_done()
assert mock_key_press.called
assert mock_key_press.call_args == call("stop")
state = hass.states.get(MAIN_ENTITY_ID)
assert state.state == STATE_PAUSED
# Turn main DVR off.
await async_turn_off(hass, MAIN_ENTITY_ID)
await hass.async_block_till_done()
assert mock_key_press.called
assert mock_key_press.call_args == call("poweroff")
state = hass.states.get(MAIN_ENTITY_ID)
assert state.state == STATE_OFF
# There should have been 6 calls to check if DVR is in standby
assert main_dtv.get_standby.call_count == 6
assert mock_get_standby.call_count == 6
# There should be 5 calls to get current info (only 1 time it will
# not be called as DVR is in standby.)
assert main_dtv.get_tuned.call_count == 5
assert mock_get_tuned.call_count == 5
# Turn main DVR off.
await async_turn_off(hass, MAIN_ENTITY_ID)
await hass.async_block_till_done()
state = hass.states.get(MAIN_ENTITY_ID)
assert state.state == STATE_OFF
async def test_available(hass, platforms, main_dtv, mock_now):
async def test_available(
hass: HomeAssistantType, mock_now: dt_util.dt.datetime, client_dtv: MockDirectvClass
) -> None:
"""Test available status."""
entry = await setup_directv_with_locations(hass, client_dtv)
next_update = mock_now + timedelta(minutes=5)
with patch("homeassistant.util.dt.utcnow", return_value=next_update):
async_fire_time_changed(hass, next_update)
@ -602,11 +387,17 @@ async def test_available(hass, platforms, main_dtv, mock_now):
state = hass.states.get(MAIN_ENTITY_ID)
assert state.state != STATE_UNAVAILABLE
assert hass.data[DOMAIN]
assert hass.data[DOMAIN][entry.entry_id]
assert hass.data[DOMAIN][entry.entry_id]["client"]
main_dtv = hass.data[DOMAIN][entry.entry_id]["client"]
# Make update fail 1st time
next_update = next_update + timedelta(minutes=5)
with patch.object(
main_dtv, "get_standby", side_effect=requests.RequestException
), patch("homeassistant.util.dt.utcnow", return_value=next_update):
with patch.object(main_dtv, "get_standby", side_effect=RequestException), patch(
"homeassistant.util.dt.utcnow", return_value=next_update
):
async_fire_time_changed(hass, next_update)
await hass.async_block_till_done()
@ -615,9 +406,9 @@ async def test_available(hass, platforms, main_dtv, mock_now):
# Make update fail 2nd time within 1 minute
next_update = next_update + timedelta(seconds=30)
with patch.object(
main_dtv, "get_standby", side_effect=requests.RequestException
), patch("homeassistant.util.dt.utcnow", return_value=next_update):
with patch.object(main_dtv, "get_standby", side_effect=RequestException), patch(
"homeassistant.util.dt.utcnow", return_value=next_update
):
async_fire_time_changed(hass, next_update)
await hass.async_block_till_done()
@ -626,9 +417,9 @@ async def test_available(hass, platforms, main_dtv, mock_now):
# Make update fail 3rd time more then a minute after 1st failure
next_update = next_update + timedelta(minutes=1)
with patch.object(
main_dtv, "get_standby", side_effect=requests.RequestException
), patch("homeassistant.util.dt.utcnow", return_value=next_update):
with patch.object(main_dtv, "get_standby", side_effect=RequestException), patch(
"homeassistant.util.dt.utcnow", return_value=next_update
):
async_fire_time_changed(hass, next_update)
await hass.async_block_till_done()
@ -640,5 +431,6 @@ async def test_available(hass, platforms, main_dtv, mock_now):
with patch("homeassistant.util.dt.utcnow", return_value=next_update):
async_fire_time_changed(hass, next_update)
await hass.async_block_till_done()
state = hass.states.get(MAIN_ENTITY_ID)
assert state.state != STATE_UNAVAILABLE