Refactor DirecTV Integration to Async (#33114)

* switch to directv==0.1.1

* work on directv async.

* Update const.py

* Update __init__.py

* Update media_player.py

* Update __init__.py

* Update __init__.py

* Update __init__.py

* Update media_player.py

* Update test_config_flow.py

* Update media_player.py

* Update media_player.py

* work on tests and coverage.

* Update __init__.py

* Update __init__.py

* squash.
This commit is contained in:
Chris Talkington 2020-03-31 17:35:32 -05:00 committed by GitHub
parent 3566803d2e
commit b892dbc6ea
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
19 changed files with 749 additions and 845 deletions

View File

@ -1,19 +1,27 @@
"""The DirecTV integration.""" """The DirecTV integration."""
import asyncio import asyncio
from datetime import timedelta from datetime import timedelta
from typing import Dict from typing import Any, Dict
from DirectPy import DIRECTV from directv import DIRECTV, DIRECTVError
from requests.exceptions import RequestException
import voluptuous as vol import voluptuous as vol
from homeassistant.config_entries import SOURCE_IMPORT, ConfigEntry from homeassistant.config_entries import SOURCE_IMPORT, ConfigEntry
from homeassistant.const import CONF_HOST from homeassistant.const import ATTR_NAME, CONF_HOST
from homeassistant.core import HomeAssistant from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryNotReady from homeassistant.exceptions import ConfigEntryNotReady
from homeassistant.helpers import config_validation as cv from homeassistant.helpers import config_validation as cv
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from homeassistant.helpers.entity import Entity
from .const import DATA_CLIENT, DATA_LOCATIONS, DATA_VERSION_INFO, DEFAULT_PORT, DOMAIN from .const import (
ATTR_IDENTIFIERS,
ATTR_MANUFACTURER,
ATTR_MODEL,
ATTR_SOFTWARE_VERSION,
ATTR_VIA_DEVICE,
DOMAIN,
)
CONFIG_SCHEMA = vol.Schema( CONFIG_SCHEMA = vol.Schema(
{ {
@ -28,21 +36,6 @@ PLATFORMS = ["media_player"]
SCAN_INTERVAL = timedelta(seconds=30) 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, determine_state=False)
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: async def async_setup(hass: HomeAssistant, config: Dict) -> bool:
"""Set up the DirecTV component.""" """Set up the DirecTV component."""
hass.data.setdefault(DOMAIN, {}) hass.data.setdefault(DOMAIN, {})
@ -60,14 +53,14 @@ async def async_setup(hass: HomeAssistant, config: Dict) -> bool:
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool: async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Set up DirecTV from a config entry.""" """Set up DirecTV from a config entry."""
dtv = DIRECTV(entry.data[CONF_HOST], session=async_get_clientsession(hass))
try: try:
dtv_data = await hass.async_add_executor_job( await dtv.update()
get_dtv_data, hass, entry.data[CONF_HOST] except DIRECTVError:
)
except RequestException:
raise ConfigEntryNotReady raise ConfigEntryNotReady
hass.data[DOMAIN][entry.entry_id] = dtv_data hass.data[DOMAIN][entry.entry_id] = dtv
for component in PLATFORMS: for component in PLATFORMS:
hass.async_create_task( hass.async_create_task(
@ -92,3 +85,32 @@ async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
hass.data[DOMAIN].pop(entry.entry_id) hass.data[DOMAIN].pop(entry.entry_id)
return unload_ok return unload_ok
class DIRECTVEntity(Entity):
"""Defines a base DirecTV entity."""
def __init__(self, *, dtv: DIRECTV, name: str, address: str = "0") -> None:
"""Initialize the DirecTV entity."""
self._address = address
self._device_id = address if address != "0" else dtv.device.info.receiver_id
self._is_client = address != "0"
self._name = name
self.dtv = dtv
@property
def name(self) -> str:
"""Return the name of the entity."""
return self._name
@property
def device_info(self) -> Dict[str, Any]:
"""Return device information about this DirecTV receiver."""
return {
ATTR_IDENTIFIERS: {(DOMAIN, self._device_id)},
ATTR_NAME: self.name,
ATTR_MANUFACTURER: self.dtv.device.info.brand,
ATTR_MODEL: None,
ATTR_SOFTWARE_VERSION: self.dtv.device.info.version,
ATTR_VIA_DEVICE: (DOMAIN, self.dtv.device.info.receiver_id),
}

View File

@ -3,18 +3,20 @@ import logging
from typing import Any, Dict, Optional from typing import Any, Dict, Optional
from urllib.parse import urlparse from urllib.parse import urlparse
from DirectPy import DIRECTV from directv import DIRECTV, DIRECTVError
from requests.exceptions import RequestException
import voluptuous as vol import voluptuous as vol
from homeassistant.components.ssdp import ATTR_SSDP_LOCATION, ATTR_UPNP_SERIAL from homeassistant.components.ssdp import ATTR_SSDP_LOCATION, ATTR_UPNP_SERIAL
from homeassistant.config_entries import CONN_CLASS_LOCAL_POLL, ConfigFlow from homeassistant.config_entries import CONN_CLASS_LOCAL_POLL, ConfigFlow
from homeassistant.const import CONF_HOST, CONF_NAME from homeassistant.const import CONF_HOST, CONF_NAME
from homeassistant.core import callback from homeassistant.helpers.aiohttp_client import async_get_clientsession
from homeassistant.exceptions import HomeAssistantError from homeassistant.helpers.typing import (
from homeassistant.helpers.typing import DiscoveryInfoType ConfigType,
DiscoveryInfoType,
HomeAssistantType,
)
from .const import DEFAULT_PORT from .const import CONF_RECEIVER_ID
from .const import DOMAIN # pylint: disable=unused-import from .const import DOMAIN # pylint: disable=unused-import
_LOGGER = logging.getLogger(__name__) _LOGGER = logging.getLogger(__name__)
@ -22,22 +24,17 @@ _LOGGER = logging.getLogger(__name__)
ERROR_CANNOT_CONNECT = "cannot_connect" ERROR_CANNOT_CONNECT = "cannot_connect"
ERROR_UNKNOWN = "unknown" ERROR_UNKNOWN = "unknown"
DATA_SCHEMA = vol.Schema({vol.Required(CONF_HOST): str})
async def validate_input(hass: HomeAssistantType, data: dict) -> Dict[str, Any]:
def validate_input(data: Dict) -> Dict:
"""Validate the user input allows us to connect. """Validate the user input allows us to connect.
Data has the keys from DATA_SCHEMA with values provided by the user. Data has the keys from DATA_SCHEMA with values provided by the user.
""" """
dtv = DIRECTV(data["host"], DEFAULT_PORT, determine_state=False) session = async_get_clientsession(hass)
version_info = dtv.get_version() directv = DIRECTV(data[CONF_HOST], session=session)
device = await directv.update()
return { return {CONF_RECEIVER_ID: device.info.receiver_id}
"title": data["host"],
"host": data["host"],
"receiver_id": "".join(version_info["receiverId"].split()),
}
class DirecTVConfigFlow(ConfigFlow, domain=DOMAIN): class DirecTVConfigFlow(ConfigFlow, domain=DOMAIN):
@ -46,84 +43,91 @@ class DirecTVConfigFlow(ConfigFlow, domain=DOMAIN):
VERSION = 1 VERSION = 1
CONNECTION_CLASS = CONN_CLASS_LOCAL_POLL CONNECTION_CLASS = CONN_CLASS_LOCAL_POLL
@callback def __init__(self):
def _show_form(self, errors: Optional[Dict] = None) -> Dict[str, Any]: """Set up the instance."""
"""Show the form to the user.""" self.discovery_info = {}
return self.async_show_form(
step_id="user", data_schema=DATA_SCHEMA, errors=errors or {},
)
async def async_step_import( async def async_step_import(
self, user_input: Optional[Dict] = None self, user_input: Optional[ConfigType] = None
) -> Dict[str, Any]: ) -> Dict[str, Any]:
"""Handle a flow initialized by yaml file.""" """Handle a flow initiated by configuration file."""
return await self.async_step_user(user_input) return await self.async_step_user(user_input)
async def async_step_user( async def async_step_user(
self, user_input: Optional[Dict] = None self, user_input: Optional[ConfigType] = None
) -> Dict[str, Any]: ) -> Dict[str, Any]:
"""Handle a flow initialized by user.""" """Handle a flow initiated by the user."""
if not user_input: if user_input is None:
return self._show_form() return self._show_setup_form()
errors = {}
try: try:
info = await self.hass.async_add_executor_job(validate_input, user_input) info = await validate_input(self.hass, user_input)
user_input[CONF_HOST] = info[CONF_HOST] except DIRECTVError:
except RequestException: return self._show_setup_form({"base": ERROR_CANNOT_CONNECT})
errors["base"] = ERROR_CANNOT_CONNECT
return self._show_form(errors)
except Exception: # pylint: disable=broad-except except Exception: # pylint: disable=broad-except
_LOGGER.exception("Unexpected exception") _LOGGER.exception("Unexpected exception")
return self.async_abort(reason=ERROR_UNKNOWN) return self.async_abort(reason=ERROR_UNKNOWN)
await self.async_set_unique_id(info["receiver_id"]) user_input[CONF_RECEIVER_ID] = info[CONF_RECEIVER_ID]
self._abort_if_unique_id_configured()
return self.async_create_entry(title=info["title"], data=user_input) await self.async_set_unique_id(user_input[CONF_RECEIVER_ID])
self._abort_if_unique_id_configured(updates={CONF_HOST: user_input[CONF_HOST]})
return self.async_create_entry(title=user_input[CONF_HOST], data=user_input)
async def async_step_ssdp( async def async_step_ssdp(
self, discovery_info: Optional[DiscoveryInfoType] = None self, discovery_info: DiscoveryInfoType
) -> Dict[str, Any]: ) -> Dict[str, Any]:
"""Handle a flow initialized by discovery.""" """Handle SSDP discovery."""
host = urlparse(discovery_info[ATTR_SSDP_LOCATION]).hostname host = urlparse(discovery_info[ATTR_SSDP_LOCATION]).hostname
receiver_id = None
if discovery_info.get(ATTR_UPNP_SERIAL):
receiver_id = discovery_info[ATTR_UPNP_SERIAL][4:] # strips off RID- 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 # pylint: disable=no-member # https://github.com/PyCQA/pylint/issues/3167
self.context.update( self.context.update({"title_placeholders": {"name": host}})
{CONF_HOST: host, CONF_NAME: host, "title_placeholders": {"name": host}}
self.discovery_info.update(
{CONF_HOST: host, CONF_NAME: host, CONF_RECEIVER_ID: receiver_id}
) )
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: try:
await self.hass.async_add_executor_job(validate_input, user_input) info = await validate_input(self.hass, self.discovery_info)
return self.async_create_entry(title=name, data=user_input) except DIRECTVError:
except (OSError, RequestException):
return self.async_abort(reason=ERROR_CANNOT_CONNECT) return self.async_abort(reason=ERROR_CANNOT_CONNECT)
except Exception: # pylint: disable=broad-except except Exception: # pylint: disable=broad-except
_LOGGER.exception("Unexpected exception") _LOGGER.exception("Unexpected exception")
return self.async_abort(reason=ERROR_UNKNOWN) return self.async_abort(reason=ERROR_UNKNOWN)
return self.async_show_form( self.discovery_info[CONF_RECEIVER_ID] = info[CONF_RECEIVER_ID]
step_id="ssdp_confirm", description_placeholders={"name": name},
await self.async_set_unique_id(self.discovery_info[CONF_RECEIVER_ID])
self._abort_if_unique_id_configured(
updates={CONF_HOST: self.discovery_info[CONF_HOST]}
) )
return await self.async_step_ssdp_confirm()
class CannotConnect(HomeAssistantError): async def async_step_ssdp_confirm(
"""Error to indicate we cannot connect.""" self, user_input: ConfigType = None
) -> Dict[str, Any]:
"""Handle a confirmation flow initiated by SSDP."""
if user_input is None:
return self.async_show_form(
step_id="ssdp_confirm",
description_placeholders={"name": self.discovery_info[CONF_NAME]},
errors={},
)
return self.async_create_entry(
title=self.discovery_info[CONF_NAME], data=self.discovery_info,
)
def _show_setup_form(self, errors: Optional[Dict] = None) -> Dict[str, Any]:
"""Show the setup form to the user."""
return self.async_show_form(
step_id="user",
data_schema=vol.Schema({vol.Required(CONF_HOST): str}),
errors=errors or {},
)

View File

@ -2,19 +2,19 @@
DOMAIN = "directv" DOMAIN = "directv"
# Attributes
ATTR_IDENTIFIERS = "identifiers"
ATTR_MANUFACTURER = "manufacturer"
ATTR_MEDIA_CURRENTLY_RECORDING = "media_currently_recording" ATTR_MEDIA_CURRENTLY_RECORDING = "media_currently_recording"
ATTR_MEDIA_RATING = "media_rating" ATTR_MEDIA_RATING = "media_rating"
ATTR_MEDIA_RECORDED = "media_recorded" ATTR_MEDIA_RECORDED = "media_recorded"
ATTR_MEDIA_START_TIME = "media_start_time" ATTR_MEDIA_START_TIME = "media_start_time"
ATTR_MODEL = "model"
ATTR_SOFTWARE_VERSION = "sw_version"
ATTR_VIA_DEVICE = "via_device"
DATA_CLIENT = "client" CONF_RECEIVER_ID = "receiver_id"
DATA_LOCATIONS = "locations"
DATA_VERSION_INFO = "version_info"
DEFAULT_DEVICE = "0" DEFAULT_DEVICE = "0"
DEFAULT_MANUFACTURER = "DirecTV"
DEFAULT_NAME = "DirecTV Receiver" DEFAULT_NAME = "DirecTV Receiver"
DEFAULT_PORT = 8080 DEFAULT_PORT = 8080
MODEL_HOST = "DirecTV Host"
MODEL_CLIENT = "DirecTV Client"

View File

@ -2,9 +2,10 @@
"domain": "directv", "domain": "directv",
"name": "DirecTV", "name": "DirecTV",
"documentation": "https://www.home-assistant.io/integrations/directv", "documentation": "https://www.home-assistant.io/integrations/directv",
"requirements": ["directpy==0.7"], "requirements": ["directv==0.2.0"],
"dependencies": [], "dependencies": [],
"codeowners": ["@ctalkington"], "codeowners": ["@ctalkington"],
"quality_scale": "gold",
"config_flow": true, "config_flow": true,
"ssdp": [ "ssdp": [
{ {

View File

@ -1,12 +1,10 @@
"""Support for the DirecTV receivers.""" """Support for the DirecTV receivers."""
import logging import logging
from typing import Callable, Dict, List, Optional from typing import Callable, List
from DirectPy import DIRECTV from directv import DIRECTV
from requests.exceptions import RequestException
import voluptuous as vol
from homeassistant.components.media_player import PLATFORM_SCHEMA, MediaPlayerDevice from homeassistant.components.media_player import MediaPlayerDevice
from homeassistant.components.media_player.const import ( from homeassistant.components.media_player.const import (
MEDIA_TYPE_CHANNEL, MEDIA_TYPE_CHANNEL,
MEDIA_TYPE_MOVIE, MEDIA_TYPE_MOVIE,
@ -21,34 +19,17 @@ from homeassistant.components.media_player.const import (
SUPPORT_TURN_ON, SUPPORT_TURN_ON,
) )
from homeassistant.config_entries import ConfigEntry from homeassistant.config_entries import ConfigEntry
from homeassistant.const import ( from homeassistant.const import STATE_OFF, STATE_PAUSED, STATE_PLAYING
CONF_DEVICE,
CONF_HOST,
CONF_NAME,
CONF_PORT,
STATE_OFF,
STATE_PAUSED,
STATE_PLAYING,
)
from homeassistant.helpers import config_validation as cv
from homeassistant.helpers.typing import HomeAssistantType from homeassistant.helpers.typing import HomeAssistantType
from homeassistant.util import dt as dt_util from homeassistant.util import dt as dt_util
from . import DIRECTVEntity
from .const import ( from .const import (
ATTR_MEDIA_CURRENTLY_RECORDING, ATTR_MEDIA_CURRENTLY_RECORDING,
ATTR_MEDIA_RATING, ATTR_MEDIA_RATING,
ATTR_MEDIA_RECORDED, ATTR_MEDIA_RECORDED,
ATTR_MEDIA_START_TIME, ATTR_MEDIA_START_TIME,
DATA_CLIENT,
DATA_LOCATIONS,
DATA_VERSION_INFO,
DEFAULT_DEVICE,
DEFAULT_MANUFACTURER,
DEFAULT_NAME,
DEFAULT_PORT,
DOMAIN, DOMAIN,
MODEL_CLIENT,
MODEL_HOST,
) )
_LOGGER = logging.getLogger(__name__) _LOGGER = logging.getLogger(__name__)
@ -73,15 +54,6 @@ SUPPORT_DTV_CLIENT = (
| SUPPORT_PLAY | SUPPORT_PLAY
) )
PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend(
{
vol.Required(CONF_HOST): cv.string,
vol.Optional(CONF_NAME, default=DEFAULT_NAME): cv.string,
vol.Optional(CONF_PORT, default=DEFAULT_PORT): cv.port,
vol.Optional(CONF_DEVICE, default=DEFAULT_DEVICE): cv.string,
}
)
async def async_setup_entry( async def async_setup_entry(
hass: HomeAssistantType, hass: HomeAssistantType,
@ -89,139 +61,57 @@ async def async_setup_entry(
async_add_entities: Callable[[List, bool], None], async_add_entities: Callable[[List, bool], None],
) -> bool: ) -> bool:
"""Set up the DirecTV config entry.""" """Set up the DirecTV config entry."""
locations = hass.data[DOMAIN][entry.entry_id][DATA_LOCATIONS] dtv = hass.data[DOMAIN][entry.entry_id]
version_info = hass.data[DOMAIN][entry.entry_id][DATA_VERSION_INFO]
entities = [] entities = []
for loc in locations["locations"]: for location in dtv.device.locations:
if "locationName" not in loc or "clientAddr" not in loc:
continue
if loc["clientAddr"] != "0":
dtv = DIRECTV(
entry.data[CONF_HOST],
DEFAULT_PORT,
loc["clientAddr"],
determine_state=False,
)
else:
dtv = hass.data[DOMAIN][entry.entry_id][DATA_CLIENT]
entities.append( entities.append(
DirecTvDevice( DIRECTVMediaPlayer(
str.title(loc["locationName"]), loc["clientAddr"], dtv, version_info, dtv=dtv, name=str.title(location.name), address=location.address,
) )
) )
async_add_entities(entities, True) async_add_entities(entities, True)
class DirecTvDevice(MediaPlayerDevice): class DIRECTVMediaPlayer(DIRECTVEntity, MediaPlayerDevice):
"""Representation of a DirecTV receiver on the network.""" """Representation of a DirecTV receiver on the network."""
def __init__( def __init__(self, *, dtv: DIRECTV, name: str, address: str = "0") -> None:
self, """Initialize DirecTV media player."""
name: str, super().__init__(
device: str, dtv=dtv, name=name, address=address,
dtv: DIRECTV, )
version_info: Optional[Dict] = None,
enabled_default: bool = True,
):
"""Initialize the device."""
self.dtv = dtv
self._name = name
self._unique_id = None
self._is_standby = True
self._current = None
self._last_update = None
self._paused = None
self._last_position = None
self._is_recorded = None
self._is_client = device != "0"
self._assumed_state = None self._assumed_state = None
self._available = False self._available = False
self._enabled_default = enabled_default
self._first_error_timestamp = None
self._model = None
self._receiver_id = None
self._software_version = None
if self._is_client:
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"]
def update(self):
"""Retrieve latest state."""
_LOGGER.debug("%s: Updating status", self.entity_id)
try:
self._available = True
self._is_standby = self.dtv.get_standby()
if self._is_standby:
self._current = None
self._is_recorded = None self._is_recorded = None
self._paused = None self._is_standby = True
self._assumed_state = False
self._last_position = None self._last_position = None
self._last_update = None self._last_update = None
else: self._paused = None
self._current = self.dtv.get_tuned() self._program = None
if self._current["status"]["code"] == 200: self._state = None
self._first_error_timestamp = None
self._is_recorded = self._current.get("uniqueId") is not None async def async_update(self):
self._paused = self._last_position == self._current["offset"] """Retrieve latest state."""
self._state = await self.dtv.state(self._address)
self._available = self._state.available
self._is_standby = self._state.standby
self._program = self._state.program
if self._is_standby:
self._assumed_state = False
self._is_recorded = None
self._last_position = None
self._last_update = None
self._paused = None
elif self._program is not None:
self._paused = self._last_position == self._program.position
self._is_recorded = self._program.recorded
self._last_position = self._program.position
self._last_update = self._state.at
self._assumed_state = self._is_recorded self._assumed_state = self._is_recorded
self._last_position = self._current["offset"]
self._last_update = (
dt_util.utcnow()
if not self._paused or self._last_update is None
else self._last_update
)
else:
# If an error is received then only set to unavailable if
# this started at least 1 minute ago.
log_message = f"{self.entity_id}: Invalid status {self._current['status']['code']} received"
if self._check_state_available():
_LOGGER.debug(log_message)
else:
_LOGGER.error(log_message)
except RequestException as exception:
_LOGGER.error(
"%s: Request error trying to update current status: %s",
self.entity_id,
exception,
)
self._check_state_available()
except Exception as exception:
_LOGGER.error(
"%s: Exception trying to update current status: %s",
self.entity_id,
exception,
)
self._available = False
if not self._first_error_timestamp:
self._first_error_timestamp = dt_util.utcnow()
raise
def _check_state_available(self):
"""Set to unavailable if issue been occurring over 1 minute."""
if not self._first_error_timestamp:
self._first_error_timestamp = dt_util.utcnow()
else:
tdelta = dt_util.utcnow() - self._first_error_timestamp
if tdelta.total_seconds() >= 60:
self._available = False
return self._available
@property @property
def device_state_attributes(self): def device_state_attributes(self):
@ -243,24 +133,10 @@ class DirecTvDevice(MediaPlayerDevice):
@property @property
def unique_id(self): def unique_id(self):
"""Return a unique ID to use for this media player.""" """Return a unique ID to use for this media player."""
return self._unique_id if self._address == "0":
return self.dtv.device.info.receiver_id
@property return self._address
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 # MediaPlayerDevice properties and methods
@property @property
@ -290,29 +166,30 @@ class DirecTvDevice(MediaPlayerDevice):
@property @property
def media_content_id(self): def media_content_id(self):
"""Return the content ID of current playing media.""" """Return the content ID of current playing media."""
if self._is_standby: if self._is_standby or self._program is None:
return None return None
return self._current["programId"] return self._program.program_id
@property @property
def media_content_type(self): def media_content_type(self):
"""Return the content type of current playing media.""" """Return the content type of current playing media."""
if self._is_standby: if self._is_standby or self._program is None:
return None return None
if "episodeTitle" in self._current: known_types = [MEDIA_TYPE_MOVIE, MEDIA_TYPE_TVSHOW]
return MEDIA_TYPE_TVSHOW if self._program.program_type in known_types:
return self._program.program_type
return MEDIA_TYPE_MOVIE return MEDIA_TYPE_MOVIE
@property @property
def media_duration(self): def media_duration(self):
"""Return the duration of current playing media in seconds.""" """Return the duration of current playing media in seconds."""
if self._is_standby: if self._is_standby or self._program is None:
return None return None
return self._current["duration"] return self._program.duration
@property @property
def media_position(self): def media_position(self):
@ -324,10 +201,7 @@ class DirecTvDevice(MediaPlayerDevice):
@property @property
def media_position_updated_at(self): def media_position_updated_at(self):
"""When was the position of the current playing media valid. """When was the position of the current playing media valid."""
Returns value from homeassistant.util.dt.utcnow().
"""
if self._is_standby: if self._is_standby:
return None return None
@ -336,34 +210,34 @@ class DirecTvDevice(MediaPlayerDevice):
@property @property
def media_title(self): def media_title(self):
"""Return the title of current playing media.""" """Return the title of current playing media."""
if self._is_standby: if self._is_standby or self._program is None:
return None return None
return self._current["title"] return self._program.title
@property @property
def media_series_title(self): def media_series_title(self):
"""Return the title of current episode of TV show.""" """Return the title of current episode of TV show."""
if self._is_standby: if self._is_standby or self._program is None:
return None return None
return self._current.get("episodeTitle") return self._program.episode_title
@property @property
def media_channel(self): def media_channel(self):
"""Return the channel current playing media.""" """Return the channel current playing media."""
if self._is_standby: if self._is_standby or self._program is None:
return None return None
return f"{self._current['callsign']} ({self._current['major']})" return f"{self._program.channel_name} ({self._program.channel})"
@property @property
def source(self): def source(self):
"""Name of the current input source.""" """Name of the current input source."""
if self._is_standby: if self._is_standby or self._program is None:
return None return None
return self._current["major"] return self._program.channel
@property @property
def supported_features(self): def supported_features(self):
@ -373,18 +247,18 @@ class DirecTvDevice(MediaPlayerDevice):
@property @property
def media_currently_recording(self): def media_currently_recording(self):
"""If the media is currently being recorded or not.""" """If the media is currently being recorded or not."""
if self._is_standby: if self._is_standby or self._program is None:
return None return None
return self._current["isRecording"] return self._program.recording
@property @property
def media_rating(self): def media_rating(self):
"""TV Rating of the current playing media.""" """TV Rating of the current playing media."""
if self._is_standby: if self._is_standby or self._program is None:
return None return None
return self._current["rating"] return self._program.rating
@property @property
def media_recorded(self): def media_recorded(self):
@ -397,53 +271,53 @@ class DirecTvDevice(MediaPlayerDevice):
@property @property
def media_start_time(self): def media_start_time(self):
"""Start time the program aired.""" """Start time the program aired."""
if self._is_standby: if self._is_standby or self._program is None:
return None return None
return dt_util.as_local(dt_util.utc_from_timestamp(self._current["startTime"])) return dt_util.as_local(self._program.start_time)
def turn_on(self): async def async_turn_on(self):
"""Turn on the receiver.""" """Turn on the receiver."""
if self._is_client: if self._is_client:
raise NotImplementedError() raise NotImplementedError()
_LOGGER.debug("Turn on %s", self._name) _LOGGER.debug("Turn on %s", self._name)
self.dtv.key_press("poweron") await self.dtv.remote("poweron", self._address)
def turn_off(self): async def async_turn_off(self):
"""Turn off the receiver.""" """Turn off the receiver."""
if self._is_client: if self._is_client:
raise NotImplementedError() raise NotImplementedError()
_LOGGER.debug("Turn off %s", self._name) _LOGGER.debug("Turn off %s", self._name)
self.dtv.key_press("poweroff") await self.dtv.remote("poweroff", self._address)
def media_play(self): async def async_media_play(self):
"""Send play command.""" """Send play command."""
_LOGGER.debug("Play on %s", self._name) _LOGGER.debug("Play on %s", self._name)
self.dtv.key_press("play") await self.dtv.remote("play", self._address)
def media_pause(self): async def async_media_pause(self):
"""Send pause command.""" """Send pause command."""
_LOGGER.debug("Pause on %s", self._name) _LOGGER.debug("Pause on %s", self._name)
self.dtv.key_press("pause") await self.dtv.remote("pause", self._address)
def media_stop(self): async def async_media_stop(self):
"""Send stop command.""" """Send stop command."""
_LOGGER.debug("Stop on %s", self._name) _LOGGER.debug("Stop on %s", self._name)
self.dtv.key_press("stop") await self.dtv.remote("stop", self._address)
def media_previous_track(self): async def async_media_previous_track(self):
"""Send rewind command.""" """Send rewind command."""
_LOGGER.debug("Rewind on %s", self._name) _LOGGER.debug("Rewind on %s", self._name)
self.dtv.key_press("rew") await self.dtv.remote("rew", self._address)
def media_next_track(self): async def async_media_next_track(self):
"""Send fast forward command.""" """Send fast forward command."""
_LOGGER.debug("Fast forward on %s", self._name) _LOGGER.debug("Fast forward on %s", self._name)
self.dtv.key_press("ffwd") await self.dtv.remote("ffwd", self._address)
def play_media(self, media_type, media_id, **kwargs): async def async_play_media(self, media_type, media_id, **kwargs):
"""Select input source.""" """Select input source."""
if media_type != MEDIA_TYPE_CHANNEL: if media_type != MEDIA_TYPE_CHANNEL:
_LOGGER.error( _LOGGER.error(
@ -454,4 +328,4 @@ class DirecTvDevice(MediaPlayerDevice):
return return
_LOGGER.debug("Changing channel on %s to %s", self._name, media_id) _LOGGER.debug("Changing channel on %s to %s", self._name, media_id)
self.dtv.tune_channel(media_id) await self.dtv.tune(media_id, self._address)

View File

@ -447,7 +447,7 @@ deluge-client==1.7.1
denonavr==0.8.1 denonavr==0.8.1
# homeassistant.components.directv # homeassistant.components.directv
directpy==0.7 directv==0.2.0
# homeassistant.components.discogs # homeassistant.components.discogs
discogs_client==2.2.2 discogs_client==2.2.2

View File

@ -178,7 +178,7 @@ defusedxml==0.6.0
denonavr==0.8.1 denonavr==0.8.1
# homeassistant.components.directv # homeassistant.components.directv
directpy==0.7 directv==0.2.0
# homeassistant.components.updater # homeassistant.components.updater
distro==1.4.0 distro==1.4.0

View File

@ -1,183 +1,94 @@
"""Tests for the DirecTV component.""" """Tests for the DirecTV component."""
from DirectPy import DIRECTV from homeassistant.components.directv.const import CONF_RECEIVER_ID, DOMAIN
from homeassistant.components.ssdp import ATTR_SSDP_LOCATION
from homeassistant.components.directv.const import DOMAIN
from homeassistant.const import CONF_HOST from homeassistant.const import CONF_HOST
from homeassistant.helpers.typing import HomeAssistantType from homeassistant.helpers.typing import HomeAssistantType
from tests.common import MockConfigEntry from tests.common import MockConfigEntry, load_fixture
from tests.test_util.aiohttp import AiohttpClientMocker
CLIENT_NAME = "Bedroom Client"
CLIENT_ADDRESS = "2CA17D1CD30X"
DEFAULT_DEVICE = "0"
HOST = "127.0.0.1" HOST = "127.0.0.1"
MAIN_NAME = "Main DVR"
RECEIVER_ID = "028877455858" RECEIVER_ID = "028877455858"
SSDP_LOCATION = "http://127.0.0.1/" SSDP_LOCATION = "http://127.0.0.1/"
UPNP_SERIAL = "RID-028877455858" 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_CONFIG = {DOMAIN: [{CONF_HOST: HOST}]}
MOCK_SSDP_DISCOVERY_INFO = {ATTR_SSDP_LOCATION: SSDP_LOCATION}
MOCK_GET_LOCATIONS = { MOCK_USER_INPUT = {CONF_HOST: HOST}
"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(DIRECTV): def mock_connection(aioclient_mock: AiohttpClientMocker) -> None:
"""A fake DirecTV DVR device.""" """Mock the DirecTV connection for Home Assistant."""
aioclient_mock.get(
def __init__(self, ip, port=8080, clientAddr="0", determine_state=False): f"http://{HOST}:8080/info/getVersion",
"""Initialize the fake DirecTV device.""" text=load_fixture("directv/info-get-version.json"),
super().__init__( headers={"Content-Type": "application/json"},
ip=ip, port=port, clientAddr=clientAddr, determine_state=determine_state,
) )
self._play = False aioclient_mock.get(
self._standby = True f"http://{HOST}:8080/info/getLocations",
text=load_fixture("directv/info-get-locations.json"),
headers={"Content-Type": "application/json"},
)
if self.clientAddr == CLIENT_ADDRESS: aioclient_mock.get(
self.attributes = RECORDING f"http://{HOST}:8080/info/mode",
self._standby = False params={"clientAddr": "9XXXXXXXXXX9"},
else: status=500,
self.attributes = LIVE text=load_fixture("directv/info-mode-error.json"),
headers={"Content-Type": "application/json"},
)
def get_locations(self): aioclient_mock.get(
"""Mock for get_locations method.""" f"http://{HOST}:8080/info/mode",
return MOCK_GET_LOCATIONS text=load_fixture("directv/info-mode.json"),
headers={"Content-Type": "application/json"},
)
def get_serial_num(self): aioclient_mock.get(
"""Mock for get_serial_num method.""" f"http://{HOST}:8080/remote/processKey",
test_serial_num = { text=load_fixture("directv/remote-process-key.json"),
"serialNum": "9999999999", headers={"Content-Type": "application/json"},
"status": { )
"code": 200,
"commandResult": 0,
"msg": "OK.",
"query": "/info/getSerialNum",
},
}
return test_serial_num aioclient_mock.get(
f"http://{HOST}:8080/tv/tune",
text=load_fixture("directv/tv-tune.json"),
headers={"Content-Type": "application/json"},
)
def get_standby(self): aioclient_mock.get(
"""Mock for get_standby method.""" f"http://{HOST}:8080/tv/getTuned",
return self._standby params={"clientAddr": "2CA17D1CD30X"},
text=load_fixture("directv/tv-get-tuned.json"),
headers={"Content-Type": "application/json"},
)
def get_tuned(self): aioclient_mock.get(
"""Mock for get_tuned method.""" f"http://{HOST}:8080/tv/getTuned",
if self._play: text=load_fixture("directv/tv-get-tuned-movie.json"),
self.attributes["offset"] = self.attributes["offset"] + 1 headers={"Content-Type": "application/json"},
)
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( async def setup_integration(
hass: HomeAssistantType, skip_entry_setup: bool = False hass: HomeAssistantType,
aioclient_mock: AiohttpClientMocker,
skip_entry_setup: bool = False,
setup_error: bool = False,
) -> MockConfigEntry: ) -> MockConfigEntry:
"""Set up the DirecTV integration in Home Assistant.""" """Set up the DirecTV integration in Home Assistant."""
if setup_error:
aioclient_mock.get(
f"http://{HOST}:8080/info/getVersion", status=500,
)
else:
mock_connection(aioclient_mock)
entry = MockConfigEntry( entry = MockConfigEntry(
domain=DOMAIN, unique_id=RECEIVER_ID, data={CONF_HOST: HOST} domain=DOMAIN,
unique_id=RECEIVER_ID,
data={CONF_HOST: HOST, CONF_RECEIVER_ID: RECEIVER_ID},
) )
entry.add_to_hass(hass) entry.add_to_hass(hass)

View File

@ -1,11 +1,9 @@
"""Test the DirecTV config flow.""" """Test the DirecTV config flow."""
from typing import Any, Dict, Optional from aiohttp import ClientError as HTTPClientError
from asynctest import patch from asynctest import patch
from requests.exceptions import RequestException
from homeassistant.components.directv.const import DOMAIN from homeassistant.components.directv.const import CONF_RECEIVER_ID, DOMAIN
from homeassistant.components.ssdp import ATTR_SSDP_LOCATION, ATTR_UPNP_SERIAL from homeassistant.components.ssdp import ATTR_UPNP_SERIAL
from homeassistant.config_entries import SOURCE_IMPORT, SOURCE_SSDP, SOURCE_USER from homeassistant.config_entries import SOURCE_IMPORT, SOURCE_SSDP, SOURCE_USER
from homeassistant.const import CONF_HOST, CONF_NAME, CONF_SOURCE from homeassistant.const import CONF_HOST, CONF_NAME, CONF_SOURCE
from homeassistant.data_entry_flow import ( from homeassistant.data_entry_flow import (
@ -14,219 +12,259 @@ from homeassistant.data_entry_flow import (
RESULT_TYPE_FORM, RESULT_TYPE_FORM,
) )
from homeassistant.helpers.typing import HomeAssistantType from homeassistant.helpers.typing import HomeAssistantType
from homeassistant.setup import async_setup_component
from tests.common import MockConfigEntry
from tests.components.directv import ( from tests.components.directv import (
HOST, HOST,
MOCK_SSDP_DISCOVERY_INFO,
MOCK_USER_INPUT,
RECEIVER_ID, RECEIVER_ID,
SSDP_LOCATION,
UPNP_SERIAL, UPNP_SERIAL,
MockDirectvClass, mock_connection,
setup_integration,
) )
from tests.test_util.aiohttp import AiohttpClientMocker
async def async_configure_flow( async def test_show_user_form(hass: HomeAssistantType) -> None:
hass: HomeAssistantType, flow_id: str, user_input: Optional[Dict] = None """Test that the user set up form is served."""
) -> 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( result = await hass.config_entries.flow.async_init(
DOMAIN, context={CONF_SOURCE: SOURCE_USER} DOMAIN, context={CONF_SOURCE: SOURCE_USER},
) )
assert result["step_id"] == "user"
assert result["type"] == RESULT_TYPE_FORM 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: async def test_show_ssdp_form(
"""Test we handle cannot connect error.""" hass: HomeAssistantType, aioclient_mock: AiohttpClientMocker
) -> None:
"""Test that the ssdp confirmation form is served."""
mock_connection(aioclient_mock)
discovery_info = MOCK_SSDP_DISCOVERY_INFO.copy()
result = await hass.config_entries.flow.async_init( result = await hass.config_entries.flow.async_init(
DOMAIN, context={CONF_SOURCE: SOURCE_USER} DOMAIN, context={CONF_SOURCE: SOURCE_SSDP}, data=discovery_info
)
with patch(
"tests.components.directv.test_config_flow.MockDirectvClass.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(
"tests.components.directv.test_config_flow.MockDirectvClass.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_ABORT
assert result["reason"] == "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["type"] == RESULT_TYPE_FORM
assert result["step_id"] == "ssdp_confirm" assert result["step_id"] == "ssdp_confirm"
assert result["description_placeholders"] == {CONF_NAME: HOST} assert result["description_placeholders"] == {CONF_NAME: HOST}
async def test_cannot_connect(
hass: HomeAssistantType, aioclient_mock: AiohttpClientMocker
) -> None:
"""Test we show user form on connection error."""
aioclient_mock.get("http://127.0.0.1:8080/info/getVersion", exc=HTTPClientError)
user_input = MOCK_USER_INPUT.copy()
result = await hass.config_entries.flow.async_init(
DOMAIN, context={CONF_SOURCE: SOURCE_USER}, data=user_input,
)
assert result["type"] == RESULT_TYPE_FORM
assert result["step_id"] == "user"
assert result["errors"] == {"base": "cannot_connect"}
async def test_ssdp_cannot_connect(
hass: HomeAssistantType, aioclient_mock: AiohttpClientMocker
) -> None:
"""Test we abort SSDP flow on connection error."""
aioclient_mock.get("http://127.0.0.1:8080/info/getVersion", exc=HTTPClientError)
discovery_info = MOCK_SSDP_DISCOVERY_INFO.copy()
result = await hass.config_entries.flow.async_init(
DOMAIN, context={CONF_SOURCE: SOURCE_SSDP}, data=discovery_info,
)
assert result["type"] == RESULT_TYPE_ABORT
assert result["reason"] == "cannot_connect"
async def test_ssdp_confirm_cannot_connect(
hass: HomeAssistantType, aioclient_mock: AiohttpClientMocker
) -> None:
"""Test we abort SSDP flow on connection error."""
aioclient_mock.get("http://127.0.0.1:8080/info/getVersion", exc=HTTPClientError)
discovery_info = MOCK_SSDP_DISCOVERY_INFO.copy()
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={CONF_SOURCE: SOURCE_SSDP, CONF_HOST: HOST, CONF_NAME: HOST},
data=discovery_info,
)
assert result["type"] == RESULT_TYPE_ABORT
assert result["reason"] == "cannot_connect"
async def test_user_device_exists_abort(
hass: HomeAssistantType, aioclient_mock: AiohttpClientMocker
) -> None:
"""Test we abort user flow if DirecTV receiver already configured."""
await setup_integration(hass, aioclient_mock)
user_input = MOCK_USER_INPUT.copy()
result = await hass.config_entries.flow.async_init(
DOMAIN, context={CONF_SOURCE: SOURCE_USER}, data=user_input,
)
assert result["type"] == RESULT_TYPE_ABORT
assert result["reason"] == "already_configured"
async def test_ssdp_device_exists_abort(
hass: HomeAssistantType, aioclient_mock: AiohttpClientMocker
) -> None:
"""Test we abort SSDP flow if DirecTV receiver already configured."""
await setup_integration(hass, aioclient_mock)
discovery_info = MOCK_SSDP_DISCOVERY_INFO.copy()
result = await hass.config_entries.flow.async_init(
DOMAIN, context={CONF_SOURCE: SOURCE_SSDP}, data=discovery_info,
)
assert result["type"] == RESULT_TYPE_ABORT
assert result["reason"] == "already_configured"
async def test_ssdp_with_receiver_id_device_exists_abort(
hass: HomeAssistantType, aioclient_mock: AiohttpClientMocker
) -> None:
"""Test we abort SSDP flow if DirecTV receiver already configured."""
await setup_integration(hass, aioclient_mock)
discovery_info = MOCK_SSDP_DISCOVERY_INFO.copy()
discovery_info[ATTR_UPNP_SERIAL] = UPNP_SERIAL
result = await hass.config_entries.flow.async_init(
DOMAIN, context={CONF_SOURCE: SOURCE_SSDP}, data=discovery_info,
)
assert result["type"] == RESULT_TYPE_ABORT
assert result["reason"] == "already_configured"
async def test_unknown_error(
hass: HomeAssistantType, aioclient_mock: AiohttpClientMocker
) -> None:
"""Test we show user form on unknown error."""
user_input = MOCK_USER_INPUT.copy()
with patch( with patch(
"homeassistant.components.directv.async_setup", return_value=True "homeassistant.components.directv.config_flow.DIRECTV.update",
) as mock_setup, patch( side_effect=Exception,
"homeassistant.components.directv.async_setup_entry", return_value=True, ):
) as mock_setup_entry: result = await hass.config_entries.flow.async_init(
result = await async_configure_flow(hass, result["flow_id"], {}) DOMAIN, context={CONF_SOURCE: SOURCE_USER}, data=user_input,
)
assert result["type"] == RESULT_TYPE_ABORT
assert result["reason"] == "unknown"
async def test_ssdp_unknown_error(
hass: HomeAssistantType, aioclient_mock: AiohttpClientMocker
) -> None:
"""Test we abort SSDP flow on unknown error."""
discovery_info = MOCK_SSDP_DISCOVERY_INFO.copy()
with patch(
"homeassistant.components.directv.config_flow.DIRECTV.update",
side_effect=Exception,
):
result = await hass.config_entries.flow.async_init(
DOMAIN, context={CONF_SOURCE: SOURCE_SSDP}, data=discovery_info,
)
assert result["type"] == RESULT_TYPE_ABORT
assert result["reason"] == "unknown"
async def test_ssdp_confirm_unknown_error(
hass: HomeAssistantType, aioclient_mock: AiohttpClientMocker
) -> None:
"""Test we abort SSDP flow on unknown error."""
discovery_info = MOCK_SSDP_DISCOVERY_INFO.copy()
with patch(
"homeassistant.components.directv.config_flow.DIRECTV.update",
side_effect=Exception,
):
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={CONF_SOURCE: SOURCE_SSDP, CONF_HOST: HOST, CONF_NAME: HOST},
data=discovery_info,
)
assert result["type"] == RESULT_TYPE_ABORT
assert result["reason"] == "unknown"
async def test_full_import_flow_implementation(
hass: HomeAssistantType, aioclient_mock: AiohttpClientMocker
) -> None:
"""Test the full manual user flow from start to finish."""
mock_connection(aioclient_mock)
user_input = MOCK_USER_INPUT.copy()
result = await hass.config_entries.flow.async_init(
DOMAIN, context={CONF_SOURCE: SOURCE_IMPORT}, data=user_input,
)
assert result["type"] == RESULT_TYPE_CREATE_ENTRY assert result["type"] == RESULT_TYPE_CREATE_ENTRY
assert result["title"] == HOST assert result["title"] == HOST
assert result["data"] == {CONF_HOST: HOST}
await hass.async_block_till_done() assert result["data"]
assert len(mock_setup.mock_calls) == 1 assert result["data"][CONF_HOST] == HOST
assert len(mock_setup_entry.mock_calls) == 1 assert result["data"][CONF_RECEIVER_ID] == RECEIVER_ID
async def test_ssdp_discovery_confirm_abort(hass: HomeAssistantType) -> None: async def test_full_user_flow_implementation(
"""Test we handle SSDP confirm cannot connect error.""" hass: HomeAssistantType, aioclient_mock: AiohttpClientMocker
) -> None:
"""Test the full manual user flow from start to finish."""
mock_connection(aioclient_mock)
result = await hass.config_entries.flow.async_init( result = await hass.config_entries.flow.async_init(
DOMAIN, DOMAIN, context={CONF_SOURCE: SOURCE_USER},
context={CONF_SOURCE: SOURCE_SSDP},
data={ATTR_SSDP_LOCATION: SSDP_LOCATION, ATTR_UPNP_SERIAL: UPNP_SERIAL},
) )
with patch( assert result["type"] == RESULT_TYPE_FORM
"tests.components.directv.test_config_flow.MockDirectvClass.get_version", assert result["step_id"] == "user"
side_effect=RequestException,
) as mock_validate_input:
result = await async_configure_flow(hass, result["flow_id"], {})
assert result["type"] == RESULT_TYPE_ABORT user_input = MOCK_USER_INPUT.copy()
result = await hass.config_entries.flow.async_configure(
await hass.async_block_till_done() result["flow_id"], user_input=user_input,
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( assert result["type"] == RESULT_TYPE_CREATE_ENTRY
"tests.components.directv.test_config_flow.MockDirectvClass.get_version", assert result["title"] == HOST
side_effect=Exception,
) as mock_validate_input:
result = await async_configure_flow(hass, result["flow_id"], {})
assert result["type"] == RESULT_TYPE_ABORT assert result["data"]
assert result["data"][CONF_HOST] == HOST
assert result["data"][CONF_RECEIVER_ID] == RECEIVER_ID
await hass.async_block_till_done()
assert len(mock_validate_input.mock_calls) == 1 async def test_full_ssdp_flow_implementation(
hass: HomeAssistantType, aioclient_mock: AiohttpClientMocker
) -> None:
"""Test the full SSDP flow from start to finish."""
mock_connection(aioclient_mock)
discovery_info = MOCK_SSDP_DISCOVERY_INFO.copy()
result = await hass.config_entries.flow.async_init(
DOMAIN, context={CONF_SOURCE: SOURCE_SSDP}, data=discovery_info
)
assert result["type"] == RESULT_TYPE_FORM
assert result["step_id"] == "ssdp_confirm"
assert result["description_placeholders"] == {CONF_NAME: HOST}
result = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input={}
)
assert result["type"] == RESULT_TYPE_CREATE_ENTRY
assert result["title"] == HOST
assert result["data"]
assert result["data"][CONF_HOST] == HOST
assert result["data"][CONF_RECEIVER_ID] == RECEIVER_ID

View File

@ -1,7 +1,4 @@
"""Tests for the Roku integration.""" """Tests for the DirecTV integration."""
from asynctest import patch
from requests.exceptions import RequestException
from homeassistant.components.directv.const import DOMAIN from homeassistant.components.directv.const import DOMAIN
from homeassistant.config_entries import ( from homeassistant.config_entries import (
ENTRY_STATE_LOADED, ENTRY_STATE_LOADED,
@ -9,34 +6,36 @@ from homeassistant.config_entries import (
ENTRY_STATE_SETUP_RETRY, ENTRY_STATE_SETUP_RETRY,
) )
from homeassistant.helpers.typing import HomeAssistantType from homeassistant.helpers.typing import HomeAssistantType
from homeassistant.setup import async_setup_component
from tests.components.directv import MockDirectvClass, setup_integration from tests.components.directv import MOCK_CONFIG, mock_connection, setup_integration
from tests.test_util.aiohttp import AiohttpClientMocker
# pylint: disable=redefined-outer-name # pylint: disable=redefined-outer-name
async def test_config_entry_not_ready(hass: HomeAssistantType) -> None: async def test_setup(
hass: HomeAssistantType, aioclient_mock: AiohttpClientMocker
) -> None:
"""Test the DirecTV setup from configuration."""
mock_connection(aioclient_mock)
assert await async_setup_component(hass, DOMAIN, MOCK_CONFIG)
async def test_config_entry_not_ready(
hass: HomeAssistantType, aioclient_mock: AiohttpClientMocker
) -> None:
"""Test the DirecTV configuration entry not ready.""" """Test the DirecTV configuration entry not ready."""
with patch( entry = await setup_integration(hass, aioclient_mock, setup_error=True)
"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 assert entry.state == ENTRY_STATE_SETUP_RETRY
async def test_unload_config_entry(hass: HomeAssistantType) -> None: async def test_unload_config_entry(
hass: HomeAssistantType, aioclient_mock: AiohttpClientMocker
) -> None:
"""Test the DirecTV configuration entry unloading.""" """Test the DirecTV configuration entry unloading."""
with patch( entry = await setup_integration(hass, aioclient_mock)
"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.entry_id in hass.data[DOMAIN]
assert entry.state == ENTRY_STATE_LOADED assert entry.state == ENTRY_STATE_LOADED

View File

@ -4,7 +4,6 @@ from typing import Optional
from asynctest import patch from asynctest import patch
from pytest import fixture from pytest import fixture
from requests import RequestException
from homeassistant.components.directv.media_player import ( from homeassistant.components.directv.media_player import (
ATTR_MEDIA_CURRENTLY_RECORDING, ATTR_MEDIA_CURRENTLY_RECORDING,
@ -24,6 +23,7 @@ from homeassistant.components.media_player.const import (
ATTR_MEDIA_SERIES_TITLE, ATTR_MEDIA_SERIES_TITLE,
ATTR_MEDIA_TITLE, ATTR_MEDIA_TITLE,
DOMAIN as MP_DOMAIN, DOMAIN as MP_DOMAIN,
MEDIA_TYPE_MOVIE,
MEDIA_TYPE_TVSHOW, MEDIA_TYPE_TVSHOW,
SERVICE_PLAY_MEDIA, SERVICE_PLAY_MEDIA,
SUPPORT_NEXT_TRACK, SUPPORT_NEXT_TRACK,
@ -44,7 +44,6 @@ from homeassistant.const import (
SERVICE_MEDIA_STOP, SERVICE_MEDIA_STOP,
SERVICE_TURN_OFF, SERVICE_TURN_OFF,
SERVICE_TURN_ON, SERVICE_TURN_ON,
STATE_OFF,
STATE_PAUSED, STATE_PAUSED,
STATE_PLAYING, STATE_PLAYING,
STATE_UNAVAILABLE, STATE_UNAVAILABLE,
@ -52,18 +51,13 @@ from homeassistant.const import (
from homeassistant.helpers.typing import HomeAssistantType from homeassistant.helpers.typing import HomeAssistantType
from homeassistant.util import dt as dt_util from homeassistant.util import dt as dt_util
from tests.common import MockConfigEntry, async_fire_time_changed from tests.components.directv import setup_integration
from tests.components.directv import ( from tests.test_util.aiohttp import AiohttpClientMocker
DOMAIN,
MOCK_GET_LOCATIONS_MULTIPLE,
RECORDING,
MockDirectvClass,
setup_integration,
)
ATTR_UNIQUE_ID = "unique_id" ATTR_UNIQUE_ID = "unique_id"
CLIENT_ENTITY_ID = f"{MP_DOMAIN}.bedroom_client" CLIENT_ENTITY_ID = f"{MP_DOMAIN}.client"
MAIN_ENTITY_ID = f"{MP_DOMAIN}.main_dvr" MAIN_ENTITY_ID = f"{MP_DOMAIN}.host"
UNAVAILABLE_ENTITY_ID = f"{MP_DOMAIN}.unavailable_client"
# pylint: disable=redefined-outer-name # pylint: disable=redefined-outer-name
@ -74,29 +68,6 @@ def mock_now() -> datetime:
return dt_util.utcnow() return dt_util.utcnow()
async def setup_directv(hass: HomeAssistantType) -> MockConfigEntry:
"""Set up mock DirecTV integration."""
with patch(
"homeassistant.components.directv.DIRECTV", new=MockDirectvClass,
):
return await setup_integration(hass)
async def setup_directv_with_locations(hass: HomeAssistantType) -> MockConfigEntry:
"""Set up mock DirecTV integration."""
with patch(
"tests.components.directv.test_media_player.MockDirectvClass.get_locations",
return_value=MOCK_GET_LOCATIONS_MULTIPLE,
):
with patch(
"homeassistant.components.directv.DIRECTV", new=MockDirectvClass,
), patch(
"homeassistant.components.directv.media_player.DIRECTV",
new=MockDirectvClass,
):
return await setup_integration(hass)
async def async_turn_on( async def async_turn_on(
hass: HomeAssistantType, entity_id: Optional[str] = None hass: HomeAssistantType, entity_id: Optional[str] = None
) -> None: ) -> None:
@ -172,23 +143,21 @@ async def async_play_media(
await hass.services.async_call(MP_DOMAIN, SERVICE_PLAY_MEDIA, data) await hass.services.async_call(MP_DOMAIN, SERVICE_PLAY_MEDIA, data)
async def test_setup(hass: HomeAssistantType) -> None: async def test_setup(
hass: HomeAssistantType, aioclient_mock: AiohttpClientMocker
) -> None:
"""Test setup with basic config.""" """Test setup with basic config."""
await setup_directv(hass) await setup_integration(hass, aioclient_mock)
assert hass.states.get(MAIN_ENTITY_ID)
async def test_setup_with_multiple_locations(hass: HomeAssistantType) -> None:
"""Test setup with basic config with client location."""
await setup_directv_with_locations(hass)
assert hass.states.get(MAIN_ENTITY_ID) assert hass.states.get(MAIN_ENTITY_ID)
assert hass.states.get(CLIENT_ENTITY_ID) assert hass.states.get(CLIENT_ENTITY_ID)
assert hass.states.get(UNAVAILABLE_ENTITY_ID)
async def test_unique_id(hass: HomeAssistantType) -> None: async def test_unique_id(
hass: HomeAssistantType, aioclient_mock: AiohttpClientMocker
) -> None:
"""Test unique id.""" """Test unique id."""
await setup_directv_with_locations(hass) await setup_integration(hass, aioclient_mock)
entity_registry = await hass.helpers.entity_registry.async_get_registry() entity_registry = await hass.helpers.entity_registry.async_get_registry()
@ -198,10 +167,15 @@ async def test_unique_id(hass: HomeAssistantType) -> None:
client = entity_registry.async_get(CLIENT_ENTITY_ID) client = entity_registry.async_get(CLIENT_ENTITY_ID)
assert client.unique_id == "2CA17D1CD30X" assert client.unique_id == "2CA17D1CD30X"
unavailable_client = entity_registry.async_get(UNAVAILABLE_ENTITY_ID)
assert unavailable_client.unique_id == "9XXXXXXXXXX9"
async def test_supported_features(hass: HomeAssistantType) -> None:
async def test_supported_features(
hass: HomeAssistantType, aioclient_mock: AiohttpClientMocker
) -> None:
"""Test supported features.""" """Test supported features."""
await setup_directv_with_locations(hass) await setup_integration(hass, aioclient_mock)
# Features supported for main DVR # Features supported for main DVR
state = hass.states.get(MAIN_ENTITY_ID) state = hass.states.get(MAIN_ENTITY_ID)
@ -231,168 +205,123 @@ async def test_supported_features(hass: HomeAssistantType) -> None:
async def test_check_attributes( async def test_check_attributes(
hass: HomeAssistantType, mock_now: dt_util.dt.datetime hass: HomeAssistantType,
mock_now: dt_util.dt.datetime,
aioclient_mock: AiohttpClientMocker,
) -> None: ) -> None:
"""Test attributes.""" """Test attributes."""
await setup_directv_with_locations(hass) await setup_integration(hass, aioclient_mock)
next_update = mock_now + timedelta(minutes=5) state = hass.states.get(MAIN_ENTITY_ID)
with patch("homeassistant.util.dt.utcnow", return_value=next_update): assert state.state == STATE_PLAYING
async_fire_time_changed(hass, next_update)
await hass.async_block_till_done()
# Start playing TV assert state.attributes.get(ATTR_MEDIA_CONTENT_ID) == "17016356"
with patch("homeassistant.util.dt.utcnow", return_value=next_update): assert state.attributes.get(ATTR_MEDIA_CONTENT_TYPE) == MEDIA_TYPE_MOVIE
await async_media_play(hass, CLIENT_ENTITY_ID) assert state.attributes.get(ATTR_MEDIA_DURATION) == 7200
await hass.async_block_till_done() assert state.attributes.get(ATTR_MEDIA_POSITION) == 4437
assert state.attributes.get(ATTR_MEDIA_POSITION_UPDATED_AT)
assert state.attributes.get(ATTR_MEDIA_TITLE) == "Snow Bride"
assert state.attributes.get(ATTR_MEDIA_SERIES_TITLE) is None
assert state.attributes.get(ATTR_MEDIA_CHANNEL) == "{} ({})".format("HALLHD", "312")
assert state.attributes.get(ATTR_INPUT_SOURCE) == "312"
assert not state.attributes.get(ATTR_MEDIA_CURRENTLY_RECORDING)
assert state.attributes.get(ATTR_MEDIA_RATING) == "TV-G"
assert not state.attributes.get(ATTR_MEDIA_RECORDED)
assert state.attributes.get(ATTR_MEDIA_START_TIME) == datetime(
2020, 3, 21, 13, 0, tzinfo=dt_util.UTC
)
state = hass.states.get(CLIENT_ENTITY_ID) state = hass.states.get(CLIENT_ENTITY_ID)
assert state.state == STATE_PLAYING assert state.state == STATE_PLAYING
assert state.attributes.get(ATTR_MEDIA_CONTENT_ID) == RECORDING["programId"] assert state.attributes.get(ATTR_MEDIA_CONTENT_ID) == "4405732"
assert state.attributes.get(ATTR_MEDIA_CONTENT_TYPE) == MEDIA_TYPE_TVSHOW assert state.attributes.get(ATTR_MEDIA_CONTENT_TYPE) == MEDIA_TYPE_TVSHOW
assert state.attributes.get(ATTR_MEDIA_DURATION) == RECORDING["duration"] assert state.attributes.get(ATTR_MEDIA_DURATION) == 1791
assert state.attributes.get(ATTR_MEDIA_POSITION) == 2 assert state.attributes.get(ATTR_MEDIA_POSITION) == 263
assert state.attributes.get(ATTR_MEDIA_POSITION_UPDATED_AT) == next_update assert state.attributes.get(ATTR_MEDIA_POSITION_UPDATED_AT)
assert state.attributes.get(ATTR_MEDIA_TITLE) == RECORDING["title"] assert state.attributes.get(ATTR_MEDIA_TITLE) == "Tyler's Ultimate"
assert state.attributes.get(ATTR_MEDIA_SERIES_TITLE) == RECORDING["episodeTitle"] assert state.attributes.get(ATTR_MEDIA_SERIES_TITLE) == "Spaghetti and Clam Sauce"
assert state.attributes.get(ATTR_MEDIA_CHANNEL) == "{} ({})".format( assert state.attributes.get(ATTR_MEDIA_CHANNEL) == "{} ({})".format("FOODHD", "231")
RECORDING["callsign"], RECORDING["major"] assert state.attributes.get(ATTR_INPUT_SOURCE) == "231"
) assert not state.attributes.get(ATTR_MEDIA_CURRENTLY_RECORDING)
assert state.attributes.get(ATTR_INPUT_SOURCE) == RECORDING["major"] assert state.attributes.get(ATTR_MEDIA_RATING) == "No Rating"
assert (
state.attributes.get(ATTR_MEDIA_CURRENTLY_RECORDING) == RECORDING["isRecording"]
)
assert state.attributes.get(ATTR_MEDIA_RATING) == RECORDING["rating"]
assert state.attributes.get(ATTR_MEDIA_RECORDED) assert state.attributes.get(ATTR_MEDIA_RECORDED)
assert state.attributes.get(ATTR_MEDIA_START_TIME) == datetime( assert state.attributes.get(ATTR_MEDIA_START_TIME) == datetime(
2018, 11, 10, 19, 0, tzinfo=dt_util.UTC 2010, 7, 5, 15, 0, 8, tzinfo=dt_util.UTC
) )
state = hass.states.get(UNAVAILABLE_ENTITY_ID)
assert state.state == STATE_UNAVAILABLE
async def test_attributes_paused(
hass: HomeAssistantType,
mock_now: dt_util.dt.datetime,
aioclient_mock: AiohttpClientMocker,
):
"""Test attributes while paused."""
await setup_integration(hass, aioclient_mock)
state = hass.states.get(CLIENT_ENTITY_ID)
last_updated = state.attributes.get(ATTR_MEDIA_POSITION_UPDATED_AT)
# Test to make sure that ATTR_MEDIA_POSITION_UPDATED_AT is not # Test to make sure that ATTR_MEDIA_POSITION_UPDATED_AT is not
# updated if TV is paused. # updated if TV is paused.
with patch( with patch(
"homeassistant.util.dt.utcnow", return_value=next_update + timedelta(minutes=5) "homeassistant.util.dt.utcnow", return_value=mock_now + timedelta(minutes=5)
): ):
await async_media_pause(hass, CLIENT_ENTITY_ID) await async_media_pause(hass, CLIENT_ENTITY_ID)
await hass.async_block_till_done() await hass.async_block_till_done()
state = hass.states.get(CLIENT_ENTITY_ID) state = hass.states.get(CLIENT_ENTITY_ID)
assert state.state == STATE_PAUSED assert state.state == STATE_PAUSED
assert state.attributes.get(ATTR_MEDIA_POSITION_UPDATED_AT) == next_update assert state.attributes.get(ATTR_MEDIA_POSITION_UPDATED_AT) == last_updated
async def test_main_services( async def test_main_services(
hass: HomeAssistantType, mock_now: dt_util.dt.datetime hass: HomeAssistantType,
mock_now: dt_util.dt.datetime,
aioclient_mock: AiohttpClientMocker,
) -> None: ) -> None:
"""Test the different services.""" """Test the different services."""
await setup_directv(hass) await setup_integration(hass, aioclient_mock)
next_update = mock_now + timedelta(minutes=5) with patch("directv.DIRECTV.remote") as remote_mock:
with patch("homeassistant.util.dt.utcnow", return_value=next_update):
async_fire_time_changed(hass, next_update)
await hass.async_block_till_done()
# DVR starts in off state.
state = hass.states.get(MAIN_ENTITY_ID)
assert state.state == STATE_OFF
# 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
# 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
# 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
# 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
# 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
# Turn main DVR off.
await async_turn_off(hass, MAIN_ENTITY_ID) await async_turn_off(hass, MAIN_ENTITY_ID)
await hass.async_block_till_done() await hass.async_block_till_done()
state = hass.states.get(MAIN_ENTITY_ID) remote_mock.assert_called_once_with("poweroff", "0")
assert state.state == STATE_OFF
with patch("directv.DIRECTV.remote") as remote_mock:
async def test_available( await async_turn_on(hass, MAIN_ENTITY_ID)
hass: HomeAssistantType, mock_now: dt_util.dt.datetime
) -> None:
"""Test available status."""
entry = await setup_directv(hass)
next_update = mock_now + timedelta(minutes=5)
with patch("homeassistant.util.dt.utcnow", return_value=next_update):
async_fire_time_changed(hass, next_update)
await hass.async_block_till_done() await hass.async_block_till_done()
remote_mock.assert_called_once_with("poweron", "0")
# Confirm service is currently set to available. with patch("directv.DIRECTV.remote") as remote_mock:
state = hass.states.get(MAIN_ENTITY_ID) await async_media_pause(hass, 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=RequestException), patch(
"homeassistant.util.dt.utcnow", return_value=next_update
):
async_fire_time_changed(hass, next_update)
await hass.async_block_till_done() await hass.async_block_till_done()
remote_mock.assert_called_once_with("pause", "0")
state = hass.states.get(MAIN_ENTITY_ID) with patch("directv.DIRECTV.remote") as remote_mock:
assert state.state != STATE_UNAVAILABLE await async_media_play(hass, MAIN_ENTITY_ID)
# Make update fail 2nd time within 1 minute
next_update = next_update + timedelta(seconds=30)
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() await hass.async_block_till_done()
remote_mock.assert_called_once_with("play", "0")
state = hass.states.get(MAIN_ENTITY_ID) with patch("directv.DIRECTV.remote") as remote_mock:
assert state.state != STATE_UNAVAILABLE await async_media_next_track(hass, MAIN_ENTITY_ID)
# 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=RequestException), patch(
"homeassistant.util.dt.utcnow", return_value=next_update
):
async_fire_time_changed(hass, next_update)
await hass.async_block_till_done() await hass.async_block_till_done()
remote_mock.assert_called_once_with("ffwd", "0")
state = hass.states.get(MAIN_ENTITY_ID) with patch("directv.DIRECTV.remote") as remote_mock:
assert state.state == STATE_UNAVAILABLE await async_media_previous_track(hass, MAIN_ENTITY_ID)
# Recheck state, update should work again.
next_update = next_update + timedelta(minutes=5)
with patch("homeassistant.util.dt.utcnow", return_value=next_update):
async_fire_time_changed(hass, next_update)
await hass.async_block_till_done() await hass.async_block_till_done()
remote_mock.assert_called_once_with("rew", "0")
state = hass.states.get(MAIN_ENTITY_ID) with patch("directv.DIRECTV.remote") as remote_mock:
assert state.state != STATE_UNAVAILABLE await async_media_stop(hass, MAIN_ENTITY_ID)
await hass.async_block_till_done()
remote_mock.assert_called_once_with("stop", "0")
with patch("directv.DIRECTV.tune") as tune_mock:
await async_play_media(hass, "channel", 312, MAIN_ENTITY_ID)
await hass.async_block_till_done()
tune_mock.assert_called_once_with("312", "0")

View File

@ -0,0 +1,22 @@
{
"locations": [
{
"clientAddr": "0",
"locationName": "Host"
},
{
"clientAddr": "2CA17D1CD30X",
"locationName": "Client"
},
{
"clientAddr": "9XXXXXXXXXX9",
"locationName": "Unavailable Client"
}
],
"status": {
"code": 200,
"commandResult": 0,
"msg": "OK.",
"query": "/info/getLocations?callback=jsonp"
}
}

View File

@ -0,0 +1,13 @@
{
"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"
}

View File

@ -0,0 +1,8 @@
{
"status": {
"code": 500,
"commandResult": 1,
"msg": "Internal Server Error.",
"query": "/info/mode"
}
}

9
tests/fixtures/directv/info-mode.json vendored Normal file
View File

@ -0,0 +1,9 @@
{
"mode": 0,
"status": {
"code": 200,
"commandResult": 0,
"msg": "OK",
"query": "/info/mode"
}
}

View File

@ -0,0 +1,10 @@
{
"hold": "keyPress",
"key": "ANY",
"status": {
"code": 200,
"commandResult": 0,
"msg": "OK",
"query": "/remote/processKey?key=ANY&hold=keyPress"
}
}

View File

@ -0,0 +1,24 @@
{
"callsign": "HALLHD",
"date": "2013",
"duration": 7200,
"isOffAir": false,
"isPclocked": 3,
"isPpv": false,
"isRecording": false,
"isVod": false,
"major": 312,
"minor": 65535,
"offset": 4437,
"programId": "17016356",
"rating": "TV-G",
"startTime": 1584795600,
"stationId": 6580971,
"title": "Snow Bride",
"status": {
"code": 200,
"commandResult": 0,
"msg": "OK.",
"query": "/tv/getTuned"
}
}

View File

@ -0,0 +1,32 @@
{
"callsign": "FOODHD",
"date": "20070324",
"duration": 1791,
"episodeTitle": "Spaghetti and Clam Sauce",
"expiration": "0",
"expiryTime": 0,
"isOffAir": false,
"isPartial": false,
"isPclocked": 1,
"isPpv": false,
"isRecording": false,
"isViewed": true,
"isVod": false,
"keepUntilFull": true,
"major": 231,
"minor": 65535,
"offset": 263,
"programId": "4405732",
"rating": "No Rating",
"recType": 3,
"startTime": 1278342008,
"stationId": 3900976,
"status": {
"code": 200,
"commandResult": 0,
"msg": "OK.",
"query": "/tv/getTuned"
},
"title": "Tyler's Ultimate",
"uniqueId": "6728716739474078694"
}

8
tests/fixtures/directv/tv-tune.json vendored Normal file
View File

@ -0,0 +1,8 @@
{
"status": {
"code": 200,
"commandResult": 0,
"msg": "OK",
"query": "/tv/tune?major=508"
}
}