From b892dbc6ea0d76774d36b8b3e313ff54d620acdf Mon Sep 17 00:00:00 2001 From: Chris Talkington Date: Tue, 31 Mar 2020 17:35:32 -0500 Subject: [PATCH] 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. --- homeassistant/components/directv/__init__.py | 72 +-- .../components/directv/config_flow.py | 138 +++--- homeassistant/components/directv/const.py | 14 +- .../components/directv/manifest.json | 3 +- .../components/directv/media_player.py | 290 ++++-------- requirements_all.txt | 2 +- requirements_test_all.txt | 2 +- tests/components/directv/__init__.py | 219 +++------- tests/components/directv/test_config_flow.py | 412 ++++++++++-------- tests/components/directv/test_init.py | 41 +- tests/components/directv/test_media_player.py | 275 +++++------- .../fixtures/directv/info-get-locations.json | 22 + tests/fixtures/directv/info-get-version.json | 13 + tests/fixtures/directv/info-mode-error.json | 8 + tests/fixtures/directv/info-mode.json | 9 + .../fixtures/directv/remote-process-key.json | 10 + .../fixtures/directv/tv-get-tuned-movie.json | 24 + tests/fixtures/directv/tv-get-tuned.json | 32 ++ tests/fixtures/directv/tv-tune.json | 8 + 19 files changed, 749 insertions(+), 845 deletions(-) create mode 100644 tests/fixtures/directv/info-get-locations.json create mode 100644 tests/fixtures/directv/info-get-version.json create mode 100644 tests/fixtures/directv/info-mode-error.json create mode 100644 tests/fixtures/directv/info-mode.json create mode 100644 tests/fixtures/directv/remote-process-key.json create mode 100644 tests/fixtures/directv/tv-get-tuned-movie.json create mode 100644 tests/fixtures/directv/tv-get-tuned.json create mode 100644 tests/fixtures/directv/tv-tune.json diff --git a/homeassistant/components/directv/__init__.py b/homeassistant/components/directv/__init__.py index fc7bb78989a..0be5957a29a 100644 --- a/homeassistant/components/directv/__init__.py +++ b/homeassistant/components/directv/__init__.py @@ -1,19 +1,27 @@ """The DirecTV integration.""" import asyncio from datetime import timedelta -from typing import Dict +from typing import Any, Dict -from DirectPy import DIRECTV -from requests.exceptions import RequestException +from directv import DIRECTV, DIRECTVError import voluptuous as vol 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.exceptions import ConfigEntryNotReady 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( { @@ -28,21 +36,6 @@ 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, 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: """Set up the DirecTV component.""" 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: """Set up DirecTV from a config entry.""" + dtv = DIRECTV(entry.data[CONF_HOST], session=async_get_clientsession(hass)) + try: - dtv_data = await hass.async_add_executor_job( - get_dtv_data, hass, entry.data[CONF_HOST] - ) - except RequestException: + await dtv.update() + except DIRECTVError: raise ConfigEntryNotReady - hass.data[DOMAIN][entry.entry_id] = dtv_data + hass.data[DOMAIN][entry.entry_id] = dtv for component in PLATFORMS: 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) 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), + } diff --git a/homeassistant/components/directv/config_flow.py b/homeassistant/components/directv/config_flow.py index b7d1604622e..406f2628ee4 100644 --- a/homeassistant/components/directv/config_flow.py +++ b/homeassistant/components/directv/config_flow.py @@ -3,18 +3,20 @@ import logging from typing import Any, Dict, Optional from urllib.parse import urlparse -from DirectPy import DIRECTV -from requests.exceptions import RequestException +from directv import DIRECTV, DIRECTVError 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 homeassistant.helpers.typing import DiscoveryInfoType +from homeassistant.helpers.aiohttp_client import async_get_clientsession +from homeassistant.helpers.typing import ( + ConfigType, + DiscoveryInfoType, + HomeAssistantType, +) -from .const import DEFAULT_PORT +from .const import CONF_RECEIVER_ID from .const import DOMAIN # pylint: disable=unused-import _LOGGER = logging.getLogger(__name__) @@ -22,22 +24,17 @@ _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: +async def validate_input(hass: HomeAssistantType, data: dict) -> Dict[str, Any]: """Validate the user input allows us to connect. Data has the keys from DATA_SCHEMA with values provided by the user. """ - dtv = DIRECTV(data["host"], DEFAULT_PORT, determine_state=False) - version_info = dtv.get_version() + session = async_get_clientsession(hass) + directv = DIRECTV(data[CONF_HOST], session=session) + device = await directv.update() - return { - "title": data["host"], - "host": data["host"], - "receiver_id": "".join(version_info["receiverId"].split()), - } + return {CONF_RECEIVER_ID: device.info.receiver_id} class DirecTVConfigFlow(ConfigFlow, domain=DOMAIN): @@ -46,84 +43,91 @@ class DirecTVConfigFlow(ConfigFlow, domain=DOMAIN): 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 {}, - ) + def __init__(self): + """Set up the instance.""" + self.discovery_info = {} async def async_step_import( - self, user_input: Optional[Dict] = None + self, user_input: Optional[ConfigType] = None ) -> 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) async def async_step_user( - self, user_input: Optional[Dict] = None + self, user_input: Optional[ConfigType] = None ) -> Dict[str, Any]: - """Handle a flow initialized by user.""" - if not user_input: - return self._show_form() - - errors = {} + """Handle a flow initiated by the user.""" + if user_input is None: + return self._show_setup_form() 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) + info = await validate_input(self.hass, user_input) + except DIRECTVError: + return self._show_setup_form({"base": ERROR_CANNOT_CONNECT}) except Exception: # pylint: disable=broad-except _LOGGER.exception("Unexpected exception") return self.async_abort(reason=ERROR_UNKNOWN) - await self.async_set_unique_id(info["receiver_id"]) - self._abort_if_unique_id_configured() + user_input[CONF_RECEIVER_ID] = info[CONF_RECEIVER_ID] - 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( - self, discovery_info: Optional[DiscoveryInfoType] = None + self, discovery_info: DiscoveryInfoType ) -> Dict[str, Any]: - """Handle a flow initialized by discovery.""" + """Handle SSDP discovery.""" host = urlparse(discovery_info[ATTR_SSDP_LOCATION]).hostname - receiver_id = discovery_info[ATTR_UPNP_SERIAL][4:] # strips off RID- + receiver_id = None - await self.async_set_unique_id(receiver_id) - self._abort_if_unique_id_configured(updates={CONF_HOST: host}) + if discovery_info.get(ATTR_UPNP_SERIAL): + receiver_id = discovery_info[ATTR_UPNP_SERIAL][4:] # strips off RID- # pylint: disable=no-member # https://github.com/PyCQA/pylint/issues/3167 - self.context.update( - {CONF_HOST: host, CONF_NAME: host, "title_placeholders": {"name": host}} + self.context.update({"title_placeholders": {"name": host}}) + + self.discovery_info.update( + {CONF_HOST: host, CONF_NAME: host, CONF_RECEIVER_ID: receiver_id} + ) + + try: + info = await validate_input(self.hass, self.discovery_info) + except DIRECTVError: + 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) + + self.discovery_info[CONF_RECEIVER_ID] = info[CONF_RECEIVER_ID] + + 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() async def async_step_ssdp_confirm( - self, user_input: Optional[Dict] = None + self, user_input: ConfigType = 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) + """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={}, + ) - 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}, + return self.async_create_entry( + title=self.discovery_info[CONF_NAME], data=self.discovery_info, ) - -class CannotConnect(HomeAssistantError): - """Error to indicate we cannot connect.""" + 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 {}, + ) diff --git a/homeassistant/components/directv/const.py b/homeassistant/components/directv/const.py index e5b04ce34f6..9ad01a0179b 100644 --- a/homeassistant/components/directv/const.py +++ b/homeassistant/components/directv/const.py @@ -2,19 +2,19 @@ DOMAIN = "directv" +# Attributes +ATTR_IDENTIFIERS = "identifiers" +ATTR_MANUFACTURER = "manufacturer" ATTR_MEDIA_CURRENTLY_RECORDING = "media_currently_recording" ATTR_MEDIA_RATING = "media_rating" ATTR_MEDIA_RECORDED = "media_recorded" ATTR_MEDIA_START_TIME = "media_start_time" +ATTR_MODEL = "model" +ATTR_SOFTWARE_VERSION = "sw_version" +ATTR_VIA_DEVICE = "via_device" -DATA_CLIENT = "client" -DATA_LOCATIONS = "locations" -DATA_VERSION_INFO = "version_info" +CONF_RECEIVER_ID = "receiver_id" DEFAULT_DEVICE = "0" -DEFAULT_MANUFACTURER = "DirecTV" DEFAULT_NAME = "DirecTV Receiver" DEFAULT_PORT = 8080 - -MODEL_HOST = "DirecTV Host" -MODEL_CLIENT = "DirecTV Client" diff --git a/homeassistant/components/directv/manifest.json b/homeassistant/components/directv/manifest.json index cb8ed68b304..4a712ba053e 100644 --- a/homeassistant/components/directv/manifest.json +++ b/homeassistant/components/directv/manifest.json @@ -2,9 +2,10 @@ "domain": "directv", "name": "DirecTV", "documentation": "https://www.home-assistant.io/integrations/directv", - "requirements": ["directpy==0.7"], + "requirements": ["directv==0.2.0"], "dependencies": [], "codeowners": ["@ctalkington"], + "quality_scale": "gold", "config_flow": true, "ssdp": [ { diff --git a/homeassistant/components/directv/media_player.py b/homeassistant/components/directv/media_player.py index f487e72f694..b93577a03d6 100644 --- a/homeassistant/components/directv/media_player.py +++ b/homeassistant/components/directv/media_player.py @@ -1,12 +1,10 @@ """Support for the DirecTV receivers.""" import logging -from typing import Callable, Dict, List, Optional +from typing import Callable, List -from DirectPy import DIRECTV -from requests.exceptions import RequestException -import voluptuous as vol +from directv import DIRECTV -from homeassistant.components.media_player import PLATFORM_SCHEMA, MediaPlayerDevice +from homeassistant.components.media_player import MediaPlayerDevice from homeassistant.components.media_player.const import ( MEDIA_TYPE_CHANNEL, MEDIA_TYPE_MOVIE, @@ -21,34 +19,17 @@ from homeassistant.components.media_player.const import ( SUPPORT_TURN_ON, ) from homeassistant.config_entries import ConfigEntry -from homeassistant.const import ( - CONF_DEVICE, - CONF_HOST, - CONF_NAME, - CONF_PORT, - STATE_OFF, - STATE_PAUSED, - STATE_PLAYING, -) -from homeassistant.helpers import config_validation as cv +from homeassistant.const import STATE_OFF, STATE_PAUSED, STATE_PLAYING from homeassistant.helpers.typing import HomeAssistantType from homeassistant.util import dt as dt_util +from . import DIRECTVEntity from .const import ( ATTR_MEDIA_CURRENTLY_RECORDING, ATTR_MEDIA_RATING, ATTR_MEDIA_RECORDED, ATTR_MEDIA_START_TIME, - DATA_CLIENT, - DATA_LOCATIONS, - DATA_VERSION_INFO, - DEFAULT_DEVICE, - DEFAULT_MANUFACTURER, - DEFAULT_NAME, - DEFAULT_PORT, DOMAIN, - MODEL_CLIENT, - MODEL_HOST, ) _LOGGER = logging.getLogger(__name__) @@ -73,15 +54,6 @@ SUPPORT_DTV_CLIENT = ( | 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( hass: HomeAssistantType, @@ -89,139 +61,57 @@ async def async_setup_entry( 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] + dtv = hass.data[DOMAIN][entry.entry_id] entities = [] - for loc in locations["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] - + for location in dtv.device.locations: entities.append( - DirecTvDevice( - str.title(loc["locationName"]), loc["clientAddr"], dtv, version_info, + DIRECTVMediaPlayer( + dtv=dtv, name=str.title(location.name), address=location.address, ) ) async_add_entities(entities, True) -class DirecTvDevice(MediaPlayerDevice): +class DIRECTVMediaPlayer(DIRECTVEntity, MediaPlayerDevice): """Representation of a DirecTV receiver on the network.""" - 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 - 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" + def __init__(self, *, dtv: DIRECTV, name: str, address: str = "0") -> None: + """Initialize DirecTV media player.""" + super().__init__( + dtv=dtv, name=name, address=address, + ) + self._assumed_state = None self._available = False - self._enabled_default = enabled_default - self._first_error_timestamp = None - self._model = None - self._receiver_id = None - self._software_version = None + self._is_recorded = None + self._is_standby = True + self._last_position = None + self._last_update = None + self._paused = None + self._program = None + self._state = 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): + async def async_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._paused = None - self._assumed_state = False - self._last_position = None - self._last_update = None - else: - self._current = self.dtv.get_tuned() - if self._current["status"]["code"] == 200: - self._first_error_timestamp = None - self._is_recorded = self._current.get("uniqueId") is not None - self._paused = self._last_position == self._current["offset"] - 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) + self._state = await self.dtv.state(self._address) + self._available = self._state.available + self._is_standby = self._state.standby + self._program = self._state.program - 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 + 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 @property def device_state_attributes(self): @@ -243,24 +133,10 @@ class DirecTvDevice(MediaPlayerDevice): @property def unique_id(self): """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 - 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 + return self._address # MediaPlayerDevice properties and methods @property @@ -290,29 +166,30 @@ class DirecTvDevice(MediaPlayerDevice): @property def media_content_id(self): """Return the content ID of current playing media.""" - if self._is_standby: + if self._is_standby or self._program is None: return None - return self._current["programId"] + return self._program.program_id @property def media_content_type(self): """Return the content type of current playing media.""" - if self._is_standby: + if self._is_standby or self._program is None: return None - if "episodeTitle" in self._current: - return MEDIA_TYPE_TVSHOW + known_types = [MEDIA_TYPE_MOVIE, MEDIA_TYPE_TVSHOW] + if self._program.program_type in known_types: + return self._program.program_type return MEDIA_TYPE_MOVIE @property def media_duration(self): """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 self._current["duration"] + return self._program.duration @property def media_position(self): @@ -324,10 +201,7 @@ class DirecTvDevice(MediaPlayerDevice): @property def media_position_updated_at(self): - """When was the position of the current playing media valid. - - Returns value from homeassistant.util.dt.utcnow(). - """ + """When was the position of the current playing media valid.""" if self._is_standby: return None @@ -336,34 +210,34 @@ class DirecTvDevice(MediaPlayerDevice): @property def media_title(self): """Return the title of current playing media.""" - if self._is_standby: + if self._is_standby or self._program is None: return None - return self._current["title"] + return self._program.title @property def media_series_title(self): """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 self._current.get("episodeTitle") + return self._program.episode_title @property def media_channel(self): """Return the channel current playing media.""" - if self._is_standby: + if self._is_standby or self._program is None: return None - return f"{self._current['callsign']} ({self._current['major']})" + return f"{self._program.channel_name} ({self._program.channel})" @property def source(self): """Name of the current input source.""" - if self._is_standby: + if self._is_standby or self._program is None: return None - return self._current["major"] + return self._program.channel @property def supported_features(self): @@ -373,18 +247,18 @@ class DirecTvDevice(MediaPlayerDevice): @property def media_currently_recording(self): """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 self._current["isRecording"] + return self._program.recording @property def media_rating(self): """TV Rating of the current playing media.""" - if self._is_standby: + if self._is_standby or self._program is None: return None - return self._current["rating"] + return self._program.rating @property def media_recorded(self): @@ -397,53 +271,53 @@ class DirecTvDevice(MediaPlayerDevice): @property def media_start_time(self): """Start time the program aired.""" - if self._is_standby: + if self._is_standby or self._program is 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.""" if self._is_client: raise NotImplementedError() _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.""" if self._is_client: raise NotImplementedError() _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.""" _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.""" _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.""" _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.""" _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.""" _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.""" if media_type != MEDIA_TYPE_CHANNEL: _LOGGER.error( @@ -454,4 +328,4 @@ class DirecTvDevice(MediaPlayerDevice): return _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) diff --git a/requirements_all.txt b/requirements_all.txt index 04e3b1c99a9..9a420d98d2d 100644 --- a/requirements_all.txt +++ b/requirements_all.txt @@ -447,7 +447,7 @@ deluge-client==1.7.1 denonavr==0.8.1 # homeassistant.components.directv -directpy==0.7 +directv==0.2.0 # homeassistant.components.discogs discogs_client==2.2.2 diff --git a/requirements_test_all.txt b/requirements_test_all.txt index 2ea01f58880..84b177eb809 100644 --- a/requirements_test_all.txt +++ b/requirements_test_all.txt @@ -178,7 +178,7 @@ defusedxml==0.6.0 denonavr==0.8.1 # homeassistant.components.directv -directpy==0.7 +directv==0.2.0 # homeassistant.components.updater distro==1.4.0 diff --git a/tests/components/directv/__init__.py b/tests/components/directv/__init__.py index 876b1e311ab..cd0f72307d8 100644 --- a/tests/components/directv/__init__.py +++ b/tests/components/directv/__init__.py @@ -1,183 +1,94 @@ """Tests for the DirecTV component.""" -from DirectPy import DIRECTV - -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 from homeassistant.const import CONF_HOST 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" -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", -} +MOCK_SSDP_DISCOVERY_INFO = {ATTR_SSDP_LOCATION: SSDP_LOCATION} +MOCK_USER_INPUT = {CONF_HOST: HOST} -class MockDirectvClass(DIRECTV): - """A fake DirecTV DVR device.""" +def mock_connection(aioclient_mock: AiohttpClientMocker) -> None: + """Mock the DirecTV connection for Home Assistant.""" + aioclient_mock.get( + f"http://{HOST}:8080/info/getVersion", + text=load_fixture("directv/info-get-version.json"), + headers={"Content-Type": "application/json"}, + ) - def __init__(self, ip, port=8080, clientAddr="0", determine_state=False): - """Initialize the fake DirecTV device.""" - super().__init__( - ip=ip, port=port, clientAddr=clientAddr, determine_state=determine_state, - ) + aioclient_mock.get( + f"http://{HOST}:8080/info/getLocations", + text=load_fixture("directv/info-get-locations.json"), + headers={"Content-Type": "application/json"}, + ) - self._play = False - self._standby = True + aioclient_mock.get( + f"http://{HOST}:8080/info/mode", + params={"clientAddr": "9XXXXXXXXXX9"}, + status=500, + text=load_fixture("directv/info-mode-error.json"), + headers={"Content-Type": "application/json"}, + ) - if self.clientAddr == CLIENT_ADDRESS: - self.attributes = RECORDING - self._standby = False - else: - self.attributes = LIVE + aioclient_mock.get( + f"http://{HOST}:8080/info/mode", + text=load_fixture("directv/info-mode.json"), + headers={"Content-Type": "application/json"}, + ) - def get_locations(self): - """Mock for get_locations method.""" - return MOCK_GET_LOCATIONS + aioclient_mock.get( + f"http://{HOST}:8080/remote/processKey", + text=load_fixture("directv/remote-process-key.json"), + headers={"Content-Type": "application/json"}, + ) - 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", - }, - } + aioclient_mock.get( + f"http://{HOST}:8080/tv/tune", + text=load_fixture("directv/tv-tune.json"), + headers={"Content-Type": "application/json"}, + ) - return test_serial_num + aioclient_mock.get( + f"http://{HOST}:8080/tv/getTuned", + params={"clientAddr": "2CA17D1CD30X"}, + text=load_fixture("directv/tv-get-tuned.json"), + headers={"Content-Type": "application/json"}, + ) - 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) + aioclient_mock.get( + f"http://{HOST}:8080/tv/getTuned", + text=load_fixture("directv/tv-get-tuned-movie.json"), + headers={"Content-Type": "application/json"}, + ) 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: """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( - 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) diff --git a/tests/components/directv/test_config_flow.py b/tests/components/directv/test_config_flow.py index bd5d8b83419..c5cfec50637 100644 --- a/tests/components/directv/test_config_flow.py +++ b/tests/components/directv/test_config_flow.py @@ -1,11 +1,9 @@ """Test the DirecTV config flow.""" -from typing import Any, Dict, Optional - +from aiohttp import ClientError as HTTPClientError 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.components.directv.const import CONF_RECEIVER_ID, DOMAIN +from homeassistant.components.ssdp import 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 ( @@ -14,219 +12,259 @@ from homeassistant.data_entry_flow import ( 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, + MOCK_SSDP_DISCOVERY_INFO, + MOCK_USER_INPUT, RECEIVER_ID, - SSDP_LOCATION, UPNP_SERIAL, - MockDirectvClass, + mock_connection, + setup_integration, ) +from tests.test_util.aiohttp import AiohttpClientMocker -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", {}) +async def test_show_user_form(hass: HomeAssistantType) -> None: + """Test that the user set up form is served.""" 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["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.""" +async def test_show_ssdp_form( + 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( - DOMAIN, context={CONF_SOURCE: SOURCE_USER} - ) - - 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}, + 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} + +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( - "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"], {}) + "homeassistant.components.directv.config_flow.DIRECTV.update", + side_effect=Exception, + ): + 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"] == "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["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 + + assert result["data"] + assert result["data"][CONF_HOST] == HOST + assert result["data"][CONF_RECEIVER_ID] == RECEIVER_ID -async def test_ssdp_discovery_confirm_abort(hass: HomeAssistantType) -> None: - """Test we handle SSDP confirm cannot connect error.""" +async def test_full_user_flow_implementation( + 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( - DOMAIN, - context={CONF_SOURCE: SOURCE_SSDP}, - data={ATTR_SSDP_LOCATION: SSDP_LOCATION, ATTR_UPNP_SERIAL: UPNP_SERIAL}, + DOMAIN, context={CONF_SOURCE: SOURCE_USER}, ) - 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"], {}) + assert result["type"] == RESULT_TYPE_FORM + assert result["step_id"] == "user" - 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}, + user_input = MOCK_USER_INPUT.copy() + result = await hass.config_entries.flow.async_configure( + result["flow_id"], user_input=user_input, ) - 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"], {}) + assert result["type"] == RESULT_TYPE_CREATE_ENTRY + assert result["title"] == HOST - 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 diff --git a/tests/components/directv/test_init.py b/tests/components/directv/test_init.py index 02e97b9b015..0d806d668a0 100644 --- a/tests/components/directv/test_init.py +++ b/tests/components/directv/test_init.py @@ -1,7 +1,4 @@ -"""Tests for the Roku integration.""" -from asynctest import patch -from requests.exceptions import RequestException - +"""Tests for the DirecTV integration.""" from homeassistant.components.directv.const import DOMAIN from homeassistant.config_entries import ( ENTRY_STATE_LOADED, @@ -9,34 +6,36 @@ from homeassistant.config_entries import ( ENTRY_STATE_SETUP_RETRY, ) 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 -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.""" - with patch( - "homeassistant.components.directv.DIRECTV", new=MockDirectvClass, - ), patch( - "homeassistant.components.directv.DIRECTV.get_locations", - side_effect=RequestException, - ): - entry = await setup_integration(hass) + entry = await setup_integration(hass, aioclient_mock, setup_error=True) 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.""" - 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) + entry = await setup_integration(hass, aioclient_mock) assert entry.entry_id in hass.data[DOMAIN] assert entry.state == ENTRY_STATE_LOADED diff --git a/tests/components/directv/test_media_player.py b/tests/components/directv/test_media_player.py index f7cf63355a8..698e6ddac31 100644 --- a/tests/components/directv/test_media_player.py +++ b/tests/components/directv/test_media_player.py @@ -4,7 +4,6 @@ from typing import Optional from asynctest import patch from pytest import fixture -from requests import RequestException from homeassistant.components.directv.media_player import ( ATTR_MEDIA_CURRENTLY_RECORDING, @@ -24,6 +23,7 @@ from homeassistant.components.media_player.const import ( ATTR_MEDIA_SERIES_TITLE, ATTR_MEDIA_TITLE, DOMAIN as MP_DOMAIN, + MEDIA_TYPE_MOVIE, MEDIA_TYPE_TVSHOW, SERVICE_PLAY_MEDIA, SUPPORT_NEXT_TRACK, @@ -44,7 +44,6 @@ from homeassistant.const import ( SERVICE_MEDIA_STOP, SERVICE_TURN_OFF, SERVICE_TURN_ON, - STATE_OFF, STATE_PAUSED, STATE_PLAYING, STATE_UNAVAILABLE, @@ -52,18 +51,13 @@ from homeassistant.const import ( from homeassistant.helpers.typing import HomeAssistantType from homeassistant.util import dt as dt_util -from tests.common import MockConfigEntry, async_fire_time_changed -from tests.components.directv import ( - DOMAIN, - MOCK_GET_LOCATIONS_MULTIPLE, - RECORDING, - MockDirectvClass, - setup_integration, -) +from tests.components.directv import setup_integration +from tests.test_util.aiohttp import AiohttpClientMocker ATTR_UNIQUE_ID = "unique_id" -CLIENT_ENTITY_ID = f"{MP_DOMAIN}.bedroom_client" -MAIN_ENTITY_ID = f"{MP_DOMAIN}.main_dvr" +CLIENT_ENTITY_ID = f"{MP_DOMAIN}.client" +MAIN_ENTITY_ID = f"{MP_DOMAIN}.host" +UNAVAILABLE_ENTITY_ID = f"{MP_DOMAIN}.unavailable_client" # pylint: disable=redefined-outer-name @@ -74,29 +68,6 @@ def mock_now() -> datetime: 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( hass: HomeAssistantType, entity_id: Optional[str] = None ) -> None: @@ -172,23 +143,21 @@ async def async_play_media( 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.""" - await setup_directv(hass) - 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) - + await setup_integration(hass, aioclient_mock) assert hass.states.get(MAIN_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.""" - await setup_directv_with_locations(hass) + await setup_integration(hass, aioclient_mock) 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) 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.""" - await setup_directv_with_locations(hass) + await setup_integration(hass, aioclient_mock) # Features supported for main DVR state = hass.states.get(MAIN_ENTITY_ID) @@ -231,168 +205,123 @@ async def test_supported_features(hass: HomeAssistantType) -> None: 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: """Test attributes.""" - await setup_directv_with_locations(hass) + await setup_integration(hass, aioclient_mock) - 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() + state = hass.states.get(MAIN_ENTITY_ID) + assert state.state == STATE_PLAYING - # Start playing TV - with patch("homeassistant.util.dt.utcnow", return_value=next_update): - await async_media_play(hass, CLIENT_ENTITY_ID) - await hass.async_block_till_done() + assert state.attributes.get(ATTR_MEDIA_CONTENT_ID) == "17016356" + assert state.attributes.get(ATTR_MEDIA_CONTENT_TYPE) == MEDIA_TYPE_MOVIE + assert state.attributes.get(ATTR_MEDIA_DURATION) == 7200 + 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) 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_DURATION) == RECORDING["duration"] - assert state.attributes.get(ATTR_MEDIA_POSITION) == 2 - assert state.attributes.get(ATTR_MEDIA_POSITION_UPDATED_AT) == next_update - assert state.attributes.get(ATTR_MEDIA_TITLE) == RECORDING["title"] - assert state.attributes.get(ATTR_MEDIA_SERIES_TITLE) == RECORDING["episodeTitle"] - assert state.attributes.get(ATTR_MEDIA_CHANNEL) == "{} ({})".format( - RECORDING["callsign"], RECORDING["major"] - ) - assert state.attributes.get(ATTR_INPUT_SOURCE) == RECORDING["major"] - 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_DURATION) == 1791 + assert state.attributes.get(ATTR_MEDIA_POSITION) == 263 + assert state.attributes.get(ATTR_MEDIA_POSITION_UPDATED_AT) + assert state.attributes.get(ATTR_MEDIA_TITLE) == "Tyler's Ultimate" + assert state.attributes.get(ATTR_MEDIA_SERIES_TITLE) == "Spaghetti and Clam Sauce" + assert state.attributes.get(ATTR_MEDIA_CHANNEL) == "{} ({})".format("FOODHD", "231") + assert state.attributes.get(ATTR_INPUT_SOURCE) == "231" + assert not state.attributes.get(ATTR_MEDIA_CURRENTLY_RECORDING) + assert state.attributes.get(ATTR_MEDIA_RATING) == "No Rating" assert state.attributes.get(ATTR_MEDIA_RECORDED) 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 # updated if TV is paused. 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 hass.async_block_till_done() state = hass.states.get(CLIENT_ENTITY_ID) 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( - hass: HomeAssistantType, mock_now: dt_util.dt.datetime + hass: HomeAssistantType, + mock_now: dt_util.dt.datetime, + aioclient_mock: AiohttpClientMocker, ) -> None: """Test the different services.""" - await setup_directv(hass) + await setup_integration(hass, aioclient_mock) - next_update = mock_now + timedelta(minutes=5) - with patch("homeassistant.util.dt.utcnow", return_value=next_update): - async_fire_time_changed(hass, next_update) + with patch("directv.DIRECTV.remote") as remote_mock: + await async_turn_off(hass, MAIN_ENTITY_ID) await hass.async_block_till_done() - # DVR starts in off state. - state = hass.states.get(MAIN_ENTITY_ID) - assert state.state == STATE_OFF + remote_mock.assert_called_once_with("poweroff", "0") - # 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 hass.async_block_till_done() - state = hass.states.get(MAIN_ENTITY_ID) - assert state.state == STATE_OFF - - -async def test_available( - 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) + with patch("directv.DIRECTV.remote") as remote_mock: + await async_turn_on(hass, MAIN_ENTITY_ID) await hass.async_block_till_done() + remote_mock.assert_called_once_with("poweron", "0") - # Confirm service is currently set to available. - 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=RequestException), patch( - "homeassistant.util.dt.utcnow", return_value=next_update - ): - async_fire_time_changed(hass, next_update) + with patch("directv.DIRECTV.remote") as remote_mock: + await async_media_pause(hass, MAIN_ENTITY_ID) await hass.async_block_till_done() + remote_mock.assert_called_once_with("pause", "0") - state = hass.states.get(MAIN_ENTITY_ID) - assert state.state != STATE_UNAVAILABLE - - # 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) + with patch("directv.DIRECTV.remote") as remote_mock: + await async_media_play(hass, MAIN_ENTITY_ID) await hass.async_block_till_done() + remote_mock.assert_called_once_with("play", "0") - state = hass.states.get(MAIN_ENTITY_ID) - assert state.state != STATE_UNAVAILABLE - - # 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) + with patch("directv.DIRECTV.remote") as remote_mock: + await async_media_next_track(hass, MAIN_ENTITY_ID) await hass.async_block_till_done() + remote_mock.assert_called_once_with("ffwd", "0") - state = hass.states.get(MAIN_ENTITY_ID) - assert state.state == STATE_UNAVAILABLE - - # 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) + with patch("directv.DIRECTV.remote") as remote_mock: + await async_media_previous_track(hass, MAIN_ENTITY_ID) await hass.async_block_till_done() + remote_mock.assert_called_once_with("rew", "0") - state = hass.states.get(MAIN_ENTITY_ID) - assert state.state != STATE_UNAVAILABLE + with patch("directv.DIRECTV.remote") as remote_mock: + 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") diff --git a/tests/fixtures/directv/info-get-locations.json b/tests/fixtures/directv/info-get-locations.json new file mode 100644 index 00000000000..5279bcebefc --- /dev/null +++ b/tests/fixtures/directv/info-get-locations.json @@ -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" + } +} diff --git a/tests/fixtures/directv/info-get-version.json b/tests/fixtures/directv/info-get-version.json new file mode 100644 index 00000000000..074e1b89dd8 --- /dev/null +++ b/tests/fixtures/directv/info-get-version.json @@ -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" +} diff --git a/tests/fixtures/directv/info-mode-error.json b/tests/fixtures/directv/info-mode-error.json new file mode 100644 index 00000000000..72bc39b1f5a --- /dev/null +++ b/tests/fixtures/directv/info-mode-error.json @@ -0,0 +1,8 @@ +{ + "status": { + "code": 500, + "commandResult": 1, + "msg": "Internal Server Error.", + "query": "/info/mode" + } +} diff --git a/tests/fixtures/directv/info-mode.json b/tests/fixtures/directv/info-mode.json new file mode 100644 index 00000000000..f1c731a07aa --- /dev/null +++ b/tests/fixtures/directv/info-mode.json @@ -0,0 +1,9 @@ +{ + "mode": 0, + "status": { + "code": 200, + "commandResult": 0, + "msg": "OK", + "query": "/info/mode" + } +} diff --git a/tests/fixtures/directv/remote-process-key.json b/tests/fixtures/directv/remote-process-key.json new file mode 100644 index 00000000000..7f73e02acc7 --- /dev/null +++ b/tests/fixtures/directv/remote-process-key.json @@ -0,0 +1,10 @@ +{ + "hold": "keyPress", + "key": "ANY", + "status": { + "code": 200, + "commandResult": 0, + "msg": "OK", + "query": "/remote/processKey?key=ANY&hold=keyPress" + } +} diff --git a/tests/fixtures/directv/tv-get-tuned-movie.json b/tests/fixtures/directv/tv-get-tuned-movie.json new file mode 100644 index 00000000000..5411e7c7951 --- /dev/null +++ b/tests/fixtures/directv/tv-get-tuned-movie.json @@ -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" + } +} diff --git a/tests/fixtures/directv/tv-get-tuned.json b/tests/fixtures/directv/tv-get-tuned.json new file mode 100644 index 00000000000..dc4e4092003 --- /dev/null +++ b/tests/fixtures/directv/tv-get-tuned.json @@ -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" +} diff --git a/tests/fixtures/directv/tv-tune.json b/tests/fixtures/directv/tv-tune.json new file mode 100644 index 00000000000..39af4fe7a4e --- /dev/null +++ b/tests/fixtures/directv/tv-tune.json @@ -0,0 +1,8 @@ +{ + "status": { + "code": 200, + "commandResult": 0, + "msg": "OK", + "query": "/tv/tune?major=508" + } +}