diff --git a/.coveragerc b/.coveragerc index dd89a4dfd26..2716a1fed44 100644 --- a/.coveragerc +++ b/.coveragerc @@ -149,7 +149,6 @@ omit = homeassistant/components/dht/sensor.py homeassistant/components/digital_ocean/* homeassistant/components/digitalloggers/switch.py - homeassistant/components/directv/media_player.py homeassistant/components/discogs/sensor.py homeassistant/components/discord/notify.py homeassistant/components/dlib_face_detect/image_processing.py diff --git a/homeassistant/components/directv/.translations/en.json b/homeassistant/components/directv/.translations/en.json new file mode 100644 index 00000000000..e2a8eff5783 --- /dev/null +++ b/homeassistant/components/directv/.translations/en.json @@ -0,0 +1,26 @@ +{ + "config": { + "abort": { + "already_configured": "DirecTV receiver is already configured" + }, + "error": { + "cannot_connect": "Failed to connect, please try again", + "unknown": "Unexpected error" + }, + "flow_title": "DirecTV: {name}", + "step": { + "ssdp_confirm": { + "data": {}, + "description": "Do you want to set up {name}?", + "title": "Connect to the DirecTV receiver" + }, + "user": { + "data": { + "host": "Host or IP address" + }, + "title": "Connect to the DirecTV receiver" + } + }, + "title": "DirecTV" + } +} \ No newline at end of file diff --git a/homeassistant/components/directv/__init__.py b/homeassistant/components/directv/__init__.py index 5934e1b6c51..d9f3f171992 100644 --- a/homeassistant/components/directv/__init__.py +++ b/homeassistant/components/directv/__init__.py @@ -1 +1,94 @@ -"""The directv component.""" +"""The DirecTV integration.""" +import asyncio +from datetime import timedelta +from typing import Dict + +from DirectPy import DIRECTV +from requests.exceptions import RequestException +import voluptuous as vol + +from homeassistant.config_entries import SOURCE_IMPORT, ConfigEntry +from homeassistant.const import CONF_HOST +from homeassistant.core import HomeAssistant +from homeassistant.exceptions import ConfigEntryNotReady +from homeassistant.helpers import config_validation as cv + +from .const import DATA_CLIENT, DATA_LOCATIONS, DATA_VERSION_INFO, DEFAULT_PORT, DOMAIN + +CONFIG_SCHEMA = vol.Schema( + { + DOMAIN: vol.All( + cv.ensure_list, [vol.Schema({vol.Required(CONF_HOST): cv.string})] + ) + }, + extra=vol.ALLOW_EXTRA, +) + +PLATFORMS = ["media_player"] +SCAN_INTERVAL = timedelta(seconds=30) + + +def get_dtv_data( + hass: HomeAssistant, host: str, port: int = DEFAULT_PORT, client_addr: str = "0" +) -> dict: + """Retrieve a DIRECTV instance, locations list, and version info for the receiver device.""" + dtv = DIRECTV(host, port, client_addr) + locations = dtv.get_locations() + version_info = dtv.get_version() + + return { + DATA_CLIENT: dtv, + DATA_LOCATIONS: locations, + DATA_VERSION_INFO: version_info, + } + + +async def async_setup(hass: HomeAssistant, config: Dict) -> bool: + """Set up the DirecTV component.""" + hass.data.setdefault(DOMAIN, {}) + + if DOMAIN in config: + for entry_config in config[DOMAIN]: + hass.async_create_task( + hass.config_entries.flow.async_init( + DOMAIN, context={"source": SOURCE_IMPORT}, data=entry_config, + ) + ) + + return True + + +async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool: + """Set up DirecTV from a config entry.""" + try: + dtv_data = await hass.async_add_executor_job( + get_dtv_data, hass, entry.data[CONF_HOST] + ) + except RequestException: + raise ConfigEntryNotReady + + hass.data[DOMAIN][entry.entry_id] = dtv_data + + for component in PLATFORMS: + hass.async_create_task( + hass.config_entries.async_forward_entry_setup(entry, component) + ) + + return True + + +async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool: + """Unload a config entry.""" + unload_ok = all( + await asyncio.gather( + *[ + hass.config_entries.async_forward_entry_unload(entry, component) + for component in PLATFORMS + ] + ) + ) + + if unload_ok: + hass.data[DOMAIN].pop(entry.entry_id) + + return unload_ok diff --git a/homeassistant/components/directv/config_flow.py b/homeassistant/components/directv/config_flow.py new file mode 100644 index 00000000000..27ddf2cda7b --- /dev/null +++ b/homeassistant/components/directv/config_flow.py @@ -0,0 +1,130 @@ +"""Config flow for DirecTV.""" +import logging +from typing import Any, Dict, Optional +from urllib.parse import urlparse + +from DirectPy import DIRECTV +from requests.exceptions import RequestException +import voluptuous as vol + +from homeassistant.components.ssdp import ATTR_SSDP_LOCATION, ATTR_UPNP_SERIAL +from homeassistant.config_entries import CONN_CLASS_LOCAL_POLL, ConfigFlow +from homeassistant.const import CONF_HOST, CONF_NAME +from homeassistant.core import callback +from homeassistant.exceptions import HomeAssistantError + +from .const import DEFAULT_PORT +from .const import DOMAIN # pylint: disable=unused-import + +_LOGGER = logging.getLogger(__name__) + +ERROR_CANNOT_CONNECT = "cannot_connect" +ERROR_UNKNOWN = "unknown" + +DATA_SCHEMA = vol.Schema({vol.Required(CONF_HOST): str}) + + +def validate_input(data: Dict) -> Dict: + """Validate the user input allows us to connect. + + Data has the keys from DATA_SCHEMA with values provided by the user. + """ + # directpy does IO in constructor. + dtv = DIRECTV(data["host"], DEFAULT_PORT) + version_info = dtv.get_version() + + return { + "title": data["host"], + "host": data["host"], + "receiver_id": "".join(version_info["receiverId"].split()), + } + + +class DirecTVConfigFlow(ConfigFlow, domain=DOMAIN): + """Handle a config flow for DirecTV.""" + + VERSION = 1 + CONNECTION_CLASS = CONN_CLASS_LOCAL_POLL + + @callback + def _show_form(self, errors: Optional[Dict] = None) -> Dict[str, Any]: + """Show the form to the user.""" + return self.async_show_form( + step_id="user", data_schema=DATA_SCHEMA, errors=errors or {}, + ) + + async def async_step_import( + self, user_input: Optional[Dict] = None + ) -> Dict[str, Any]: + """Handle a flow initialized by yaml file.""" + return await self.async_step_user(user_input) + + async def async_step_user( + self, user_input: Optional[Dict] = None + ) -> Dict[str, Any]: + """Handle a flow initialized by user.""" + if not user_input: + return self._show_form() + + errors = {} + + try: + info = await self.hass.async_add_executor_job(validate_input, user_input) + user_input[CONF_HOST] = info[CONF_HOST] + except RequestException: + errors["base"] = ERROR_CANNOT_CONNECT + return self._show_form(errors) + except Exception: # pylint: disable=broad-except + _LOGGER.exception("Unexpected exception") + errors["base"] = ERROR_UNKNOWN + return self._show_form(errors) + + await self.async_set_unique_id(info["receiver_id"]) + self._abort_if_unique_id_configured() + + return self.async_create_entry(title=info["title"], data=user_input) + + async def async_step_ssdp( + self, discovery_info: Optional[Dict] = None + ) -> Dict[str, Any]: + """Handle a flow initialized by discovery.""" + host = urlparse(discovery_info[ATTR_SSDP_LOCATION]).hostname + receiver_id = discovery_info[ATTR_UPNP_SERIAL][4:] # strips off RID- + + await self.async_set_unique_id(receiver_id) + self._abort_if_unique_id_configured(updates={CONF_HOST: host}) + + # pylint: disable=no-member # https://github.com/PyCQA/pylint/issues/3167 + self.context.update( + {CONF_HOST: host, CONF_NAME: host, "title_placeholders": {"name": host}} + ) + + return await self.async_step_ssdp_confirm() + + async def async_step_ssdp_confirm( + self, user_input: Optional[Dict] = None + ) -> Dict[str, Any]: + """Handle user-confirmation of discovered device.""" + # pylint: disable=no-member # https://github.com/PyCQA/pylint/issues/3167 + name = self.context.get(CONF_NAME) + + if user_input is not None: + # pylint: disable=no-member # https://github.com/PyCQA/pylint/issues/3167 + user_input[CONF_HOST] = self.context.get(CONF_HOST) + + try: + await self.hass.async_add_executor_job(validate_input, user_input) + return self.async_create_entry(title=name, data=user_input) + except (OSError, RequestException): + return self.async_abort(reason=ERROR_CANNOT_CONNECT) + except Exception: # pylint: disable=broad-except + _LOGGER.exception("Unexpected exception") + return self.async_abort(reason=ERROR_UNKNOWN) + + return self.async_show_form( + step_id="ssdp_confirm", description_placeholders={"name": name}, + ) + + +class CannotConnect(HomeAssistantError): + """Error to indicate we cannot connect.""" diff --git a/homeassistant/components/directv/const.py b/homeassistant/components/directv/const.py index 8b3ae08c526..e5b04ce34f6 100644 --- a/homeassistant/components/directv/const.py +++ b/homeassistant/components/directv/const.py @@ -1,12 +1,20 @@ """Constants for the DirecTV integration.""" +DOMAIN = "directv" + ATTR_MEDIA_CURRENTLY_RECORDING = "media_currently_recording" ATTR_MEDIA_RATING = "media_rating" ATTR_MEDIA_RECORDED = "media_recorded" ATTR_MEDIA_START_TIME = "media_start_time" -DATA_DIRECTV = "data_directv" +DATA_CLIENT = "client" +DATA_LOCATIONS = "locations" +DATA_VERSION_INFO = "version_info" DEFAULT_DEVICE = "0" +DEFAULT_MANUFACTURER = "DirecTV" DEFAULT_NAME = "DirecTV Receiver" DEFAULT_PORT = 8080 + +MODEL_HOST = "DirecTV Host" +MODEL_CLIENT = "DirecTV Client" diff --git a/homeassistant/components/directv/manifest.json b/homeassistant/components/directv/manifest.json index cfe74153f5c..7e1dffd7435 100644 --- a/homeassistant/components/directv/manifest.json +++ b/homeassistant/components/directv/manifest.json @@ -4,5 +4,12 @@ "documentation": "https://www.home-assistant.io/integrations/directv", "requirements": ["directpy==0.6"], "dependencies": [], - "codeowners": ["@ctalkington"] + "codeowners": ["@ctalkington"], + "config_flow": true, + "ssdp": [ + { + "manufacturer": "DIRECTV", + "deviceType": "urn:schemas-upnp-org:device:MediaServer:1" + } + ] } diff --git a/homeassistant/components/directv/media_player.py b/homeassistant/components/directv/media_player.py index 673e97a18af..c1c227d319d 100644 --- a/homeassistant/components/directv/media_player.py +++ b/homeassistant/components/directv/media_player.py @@ -1,8 +1,9 @@ """Support for the DirecTV receivers.""" import logging +from typing import Callable, Dict, List, Optional from DirectPy import DIRECTV -import requests +from requests.exceptions import RequestException import voluptuous as vol from homeassistant.components.media_player import PLATFORM_SCHEMA, MediaPlayerDevice @@ -19,6 +20,7 @@ from homeassistant.components.media_player.const import ( SUPPORT_TURN_OFF, SUPPORT_TURN_ON, ) +from homeassistant.config_entries import ConfigEntry from homeassistant.const import ( CONF_DEVICE, CONF_HOST, @@ -28,18 +30,25 @@ from homeassistant.const import ( STATE_PAUSED, STATE_PLAYING, ) -import homeassistant.helpers.config_validation as cv -import homeassistant.util.dt as dt_util +from homeassistant.helpers import config_validation as cv +from homeassistant.helpers.typing import HomeAssistantType +from homeassistant.util import dt as dt_util from .const import ( ATTR_MEDIA_CURRENTLY_RECORDING, ATTR_MEDIA_RATING, ATTR_MEDIA_RECORDED, ATTR_MEDIA_START_TIME, - DATA_DIRECTV, + DATA_CLIENT, + DATA_LOCATIONS, + DATA_VERSION_INFO, DEFAULT_DEVICE, + DEFAULT_MANUFACTURER, DEFAULT_NAME, DEFAULT_PORT, + DOMAIN, + MODEL_CLIENT, + MODEL_HOST, ) _LOGGER = logging.getLogger(__name__) @@ -74,97 +83,67 @@ PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend( ) -def setup_platform(hass, config, add_entities, discovery_info=None): - """Set up the DirecTV platform.""" - known_devices = hass.data.get(DATA_DIRECTV, set()) +def get_dtv_instance( + host: str, port: int = DEFAULT_PORT, client_addr: str = "0" +) -> DIRECTV: + """Retrieve a DIRECTV instance for the receiver or client device.""" + try: + return DIRECTV(host, port, client_addr) + except RequestException as exception: + _LOGGER.debug( + "Request exception %s trying to retrieve DIRECTV instance for client address %s on device %s", + exception, + client_addr, + host, + ) + return None + + +async def async_setup_entry( + hass: HomeAssistantType, + entry: ConfigEntry, + async_add_entities: Callable[[List, bool], None], +) -> bool: + """Set up the DirecTV config entry.""" + locations = hass.data[DOMAIN][entry.entry_id][DATA_LOCATIONS] + version_info = hass.data[DOMAIN][entry.entry_id][DATA_VERSION_INFO] entities = [] - if CONF_HOST in config: - name = config[CONF_NAME] - host = config[CONF_HOST] - port = config[CONF_PORT] - device = config[CONF_DEVICE] + for loc in locations["locations"]: + if "locationName" not in loc or "clientAddr" not in loc: + continue - _LOGGER.debug( - "Adding configured device %s with client address %s", name, device, + if loc["clientAddr"] != "0": + # directpy does IO in constructor. + dtv = await hass.async_add_executor_job( + get_dtv_instance, entry.data[CONF_HOST], DEFAULT_PORT, loc["clientAddr"] + ) + else: + dtv = hass.data[DOMAIN][entry.entry_id][DATA_CLIENT] + + if not dtv: + continue + + entities.append( + DirecTvDevice( + str.title(loc["locationName"]), loc["clientAddr"], dtv, version_info, + ) ) - dtv = DIRECTV(host, port, device) - dtv_version = _get_receiver_version(dtv) - - entities.append(DirecTvDevice(name, device, dtv, dtv_version,)) - known_devices.add((host, device)) - - elif discovery_info: - host = discovery_info.get("host") - name = f"DirecTV_{discovery_info.get('serial', '')}" - - # Attempt to discover additional RVU units - _LOGGER.debug("Doing discovery of DirecTV devices on %s", host) - - dtv = DIRECTV(host, DEFAULT_PORT) - - try: - dtv_version = _get_receiver_version(dtv) - resp = dtv.get_locations() - except requests.exceptions.RequestException as ex: - # Bail out and just go forward with uPnP data - # Make sure that this device is not already configured - # Comparing based on host (IP) and clientAddr. - _LOGGER.debug("Request exception %s trying to get locations", ex) - resp = {"locations": [{"locationName": name, "clientAddr": DEFAULT_DEVICE}]} - - _LOGGER.debug("Known devices: %s", known_devices) - for loc in resp.get("locations") or []: - if "locationName" not in loc or "clientAddr" not in loc: - continue - - loc_name = str.title(loc["locationName"]) - - # Make sure that this device is not already configured - # Comparing based on host (IP) and clientAddr. - if (host, loc["clientAddr"]) in known_devices: - _LOGGER.debug( - "Discovered device %s on host %s with " - "client address %s is already " - "configured", - loc_name, - host, - loc["clientAddr"], - ) - else: - _LOGGER.debug( - "Adding discovered device %s with client address %s", - loc_name, - loc["clientAddr"], - ) - - entities.append( - DirecTvDevice( - loc_name, - loc["clientAddr"], - DIRECTV(host, DEFAULT_PORT, loc["clientAddr"]), - dtv_version, - ) - ) - known_devices.add((host, loc["clientAddr"])) - - add_entities(entities) - - -def _get_receiver_version(client): - """Return the version of the DirectTV receiver.""" - try: - return client.get_version() - except requests.exceptions.RequestException as ex: - _LOGGER.debug("Request exception %s trying to get receiver version", ex) - return None + async_add_entities(entities, True) class DirecTvDevice(MediaPlayerDevice): """Representation of a DirecTV receiver on the network.""" - def __init__(self, name, device, dtv, version_info=None): + def __init__( + self, + name: str, + device: str, + dtv: DIRECTV, + version_info: Optional[Dict] = None, + enabled_default: bool = True, + ): """Initialize the device.""" self.dtv = dtv self._name = name @@ -178,17 +157,32 @@ class DirecTvDevice(MediaPlayerDevice): self._is_client = device != "0" self._assumed_state = None self._available = False + self._enabled_default = enabled_default self._first_error_timestamp = None - - if device != "0": - self._unique_id = device - elif version_info: - self._unique_id = "".join(version_info.get("receiverId").split()) + self._model = None + self._receiver_id = None + self._software_version = None if self._is_client: - _LOGGER.debug("Created DirecTV client %s for device %s", self._name, device) + self._model = MODEL_CLIENT + self._unique_id = device + + if version_info: + self._receiver_id = "".join(version_info["receiverId"].split()) + + if not self._is_client: + self._unique_id = self._receiver_id + self._model = MODEL_HOST + self._software_version = version_info["stbSoftwareVersion"] + + if self._is_client: + _LOGGER.debug( + "Created DirecTV media player for client %s on device %s", + self._name, + device, + ) else: - _LOGGER.debug("Created DirecTV device for %s", self._name) + _LOGGER.debug("Created DirecTV media player for device %s", self._name) def update(self): """Retrieve latest state.""" @@ -225,17 +219,19 @@ class DirecTvDevice(MediaPlayerDevice): else: _LOGGER.error(log_message) - except requests.RequestException as ex: + except RequestException as exception: _LOGGER.error( "%s: Request error trying to update current status: %s", self.entity_id, - ex, + exception, ) self._check_state_available() - except Exception as ex: + except Exception as exception: _LOGGER.error( - "%s: Exception trying to update current status: %s", self.entity_id, ex + "%s: Exception trying to update current status: %s", + self.entity_id, + exception, ) self._available = False if not self._first_error_timestamp: @@ -275,6 +271,23 @@ class DirecTvDevice(MediaPlayerDevice): """Return a unique ID to use for this media player.""" return self._unique_id + @property + def device_info(self): + """Return device specific attributes.""" + return { + "name": self.name, + "identifiers": {(DOMAIN, self.unique_id)}, + "manufacturer": DEFAULT_MANUFACTURER, + "model": self._model, + "sw_version": self._software_version, + "via_device": (DOMAIN, self._receiver_id), + } + + @property + def entity_registry_enabled_default(self) -> bool: + """Return if the entity should be enabled when first added to the entity registry.""" + return self._enabled_default + # MediaPlayerDevice properties and methods @property def state(self): diff --git a/homeassistant/components/directv/strings.json b/homeassistant/components/directv/strings.json new file mode 100644 index 00000000000..78316d663bd --- /dev/null +++ b/homeassistant/components/directv/strings.json @@ -0,0 +1,26 @@ +{ + "config": { + "title": "DirecTV", + "flow_title": "DirecTV: {name}", + "step": { + "ssdp_confirm": { + "data": {}, + "description": "Do you want to set up {name}?", + "title": "Connect to the DirecTV receiver" + }, + "user": { + "title": "Connect to the DirecTV receiver", + "data": { + "host": "Host or IP address" + } + } + }, + "error": { + "cannot_connect": "Failed to connect, please try again", + "unknown": "Unexpected error" + }, + "abort": { + "already_configured": "DirecTV receiver is already configured" + } + } +} diff --git a/homeassistant/generated/config_flows.py b/homeassistant/generated/config_flows.py index 91fda9f1c32..b281a322b23 100644 --- a/homeassistant/generated/config_flows.py +++ b/homeassistant/generated/config_flows.py @@ -23,6 +23,7 @@ FLOWS = [ "daikin", "deconz", "dialogflow", + "directv", "dynalite", "ecobee", "elgato", diff --git a/homeassistant/generated/ssdp.py b/homeassistant/generated/ssdp.py index 0eb9af0231d..3bf54b1d9f7 100644 --- a/homeassistant/generated/ssdp.py +++ b/homeassistant/generated/ssdp.py @@ -11,6 +11,12 @@ SSDP = { "manufacturer": "Royal Philips Electronics" } ], + "directv": [ + { + "deviceType": "urn:schemas-upnp-org:device:MediaServer:1", + "manufacturer": "DIRECTV" + } + ], "heos": [ { "st": "urn:schemas-denon-com:device:ACT-Denon:1" diff --git a/tests/components/directv/__init__.py b/tests/components/directv/__init__.py index 9a32215e53d..d7f79c76be5 100644 --- a/tests/components/directv/__init__.py +++ b/tests/components/directv/__init__.py @@ -1 +1,182 @@ -"""Tests for the directv component.""" +"""Tests for the DirecTV component.""" +from homeassistant.components.directv.const import DOMAIN +from homeassistant.const import CONF_HOST +from homeassistant.helpers.typing import HomeAssistantType + +from tests.common import MockConfigEntry + +CLIENT_NAME = "Bedroom Client" +CLIENT_ADDRESS = "2CA17D1CD30X" +DEFAULT_DEVICE = "0" +HOST = "127.0.0.1" +MAIN_NAME = "Main DVR" +RECEIVER_ID = "028877455858" +SSDP_LOCATION = "http://127.0.0.1/" +UPNP_SERIAL = "RID-028877455858" + +LIVE = { + "callsign": "HASSTV", + "date": "20181110", + "duration": 3600, + "isOffAir": False, + "isPclocked": 1, + "isPpv": False, + "isRecording": False, + "isVod": False, + "major": 202, + "minor": 65535, + "offset": 1, + "programId": "102454523", + "rating": "No Rating", + "startTime": 1541876400, + "stationId": 3900947, + "title": "Using Home Assistant to automate your home", +} + +RECORDING = { + "callsign": "HASSTV", + "date": "20181110", + "duration": 3600, + "isOffAir": False, + "isPclocked": 1, + "isPpv": False, + "isRecording": True, + "isVod": False, + "major": 202, + "minor": 65535, + "offset": 1, + "programId": "102454523", + "rating": "No Rating", + "startTime": 1541876400, + "stationId": 3900947, + "title": "Using Home Assistant to automate your home", + "uniqueId": "12345", + "episodeTitle": "Configure DirecTV platform.", +} + +MOCK_CONFIG = {DOMAIN: [{CONF_HOST: HOST}]} + +MOCK_GET_LOCATIONS = { + "locations": [{"locationName": MAIN_NAME, "clientAddr": DEFAULT_DEVICE}], + "status": { + "code": 200, + "commandResult": 0, + "msg": "OK.", + "query": "/info/getLocations", + }, +} + +MOCK_GET_LOCATIONS_MULTIPLE = { + "locations": [ + {"locationName": MAIN_NAME, "clientAddr": DEFAULT_DEVICE}, + {"locationName": CLIENT_NAME, "clientAddr": CLIENT_ADDRESS}, + ], + "status": { + "code": 200, + "commandResult": 0, + "msg": "OK.", + "query": "/info/getLocations", + }, +} + +MOCK_GET_VERSION = { + "accessCardId": "0021-1495-6572", + "receiverId": "0288 7745 5858", + "status": { + "code": 200, + "commandResult": 0, + "msg": "OK.", + "query": "/info/getVersion", + }, + "stbSoftwareVersion": "0x4ed7", + "systemTime": 1281625203, + "version": "1.2", +} + + +class MockDirectvClass: + """A fake DirecTV DVR device.""" + + def __init__(self, ip, port=8080, clientAddr="0"): + """Initialize the fake DirecTV device.""" + self._host = ip + self._port = port + self._device = clientAddr + self._standby = True + self._play = False + + self.attributes = LIVE + + def get_locations(self): + """Mock for get_locations method.""" + return MOCK_GET_LOCATIONS + + def get_serial_num(self): + """Mock for get_serial_num method.""" + test_serial_num = { + "serialNum": "9999999999", + "status": { + "code": 200, + "commandResult": 0, + "msg": "OK.", + "query": "/info/getSerialNum", + }, + } + + return test_serial_num + + def get_standby(self): + """Mock for get_standby method.""" + return self._standby + + def get_tuned(self): + """Mock for get_tuned method.""" + if self._play: + self.attributes["offset"] = self.attributes["offset"] + 1 + + test_attributes = self.attributes + test_attributes["status"] = { + "code": 200, + "commandResult": 0, + "msg": "OK.", + "query": "/tv/getTuned", + } + return test_attributes + + def get_version(self): + """Mock for get_version method.""" + return MOCK_GET_VERSION + + def key_press(self, keypress): + """Mock for key_press method.""" + if keypress == "poweron": + self._standby = False + self._play = True + elif keypress == "poweroff": + self._standby = True + self._play = False + elif keypress == "play": + self._play = True + elif keypress == "pause" or keypress == "stop": + self._play = False + + def tune_channel(self, source): + """Mock for tune_channel method.""" + self.attributes["major"] = int(source) + + +async def setup_integration( + hass: HomeAssistantType, skip_entry_setup: bool = False +) -> MockConfigEntry: + """Set up the DirecTV integration in Home Assistant.""" + entry = MockConfigEntry( + domain=DOMAIN, unique_id=RECEIVER_ID, data={CONF_HOST: HOST} + ) + + entry.add_to_hass(hass) + + if not skip_entry_setup: + await hass.config_entries.async_setup(entry.entry_id) + await hass.async_block_till_done() + + return entry diff --git a/tests/components/directv/test_config_flow.py b/tests/components/directv/test_config_flow.py new file mode 100644 index 00000000000..5516b61cd46 --- /dev/null +++ b/tests/components/directv/test_config_flow.py @@ -0,0 +1,240 @@ +"""Test the DirecTV config flow.""" +from typing import Any, Dict, Optional + +from asynctest import patch +from requests.exceptions import RequestException + +from homeassistant.components.directv.const import DOMAIN +from homeassistant.components.ssdp import ATTR_SSDP_LOCATION, ATTR_UPNP_SERIAL +from homeassistant.config_entries import SOURCE_IMPORT, SOURCE_SSDP, SOURCE_USER +from homeassistant.const import CONF_HOST, CONF_NAME, CONF_SOURCE +from homeassistant.data_entry_flow import ( + RESULT_TYPE_ABORT, + RESULT_TYPE_CREATE_ENTRY, + RESULT_TYPE_FORM, +) +from homeassistant.helpers.typing import HomeAssistantType +from homeassistant.setup import async_setup_component + +from tests.common import MockConfigEntry +from tests.components.directv import ( + HOST, + RECEIVER_ID, + SSDP_LOCATION, + UPNP_SERIAL, + MockDirectvClass, +) + + +async def async_configure_flow( + hass: HomeAssistantType, flow_id: str, user_input: Optional[Dict] = None +) -> Any: + """Set up mock DirecTV integration flow.""" + with patch( + "homeassistant.components.directv.config_flow.DIRECTV", new=MockDirectvClass, + ): + return await hass.config_entries.flow.async_configure( + flow_id=flow_id, user_input=user_input + ) + + +async def async_init_flow( + hass: HomeAssistantType, + handler: str = DOMAIN, + context: Optional[Dict] = None, + data: Any = None, +) -> Any: + """Set up mock DirecTV integration flow.""" + with patch( + "homeassistant.components.directv.config_flow.DIRECTV", new=MockDirectvClass, + ): + return await hass.config_entries.flow.async_init( + handler=handler, context=context, data=data + ) + + +async def test_duplicate_error(hass: HomeAssistantType) -> None: + """Test that errors are shown when duplicates are added.""" + MockConfigEntry( + domain=DOMAIN, unique_id=RECEIVER_ID, data={CONF_HOST: HOST} + ).add_to_hass(hass) + + result = await async_init_flow( + hass, context={CONF_SOURCE: SOURCE_IMPORT}, data={CONF_HOST: HOST} + ) + + assert result["type"] == RESULT_TYPE_ABORT + assert result["reason"] == "already_configured" + + result = await async_init_flow( + hass, context={CONF_SOURCE: SOURCE_USER}, data={CONF_HOST: HOST} + ) + + assert result["type"] == RESULT_TYPE_ABORT + assert result["reason"] == "already_configured" + + result = await async_init_flow( + hass, + context={CONF_SOURCE: SOURCE_SSDP}, + data={ATTR_SSDP_LOCATION: SSDP_LOCATION, ATTR_UPNP_SERIAL: UPNP_SERIAL}, + ) + + assert result["type"] == RESULT_TYPE_ABORT + assert result["reason"] == "already_configured" + + +async def test_form(hass: HomeAssistantType) -> None: + """Test we get the form.""" + await async_setup_component(hass, "persistent_notification", {}) + result = await hass.config_entries.flow.async_init( + DOMAIN, context={CONF_SOURCE: SOURCE_USER} + ) + assert result["type"] == RESULT_TYPE_FORM + assert result["errors"] == {} + + with patch( + "homeassistant.components.directv.async_setup", return_value=True + ) as mock_setup, patch( + "homeassistant.components.directv.async_setup_entry", return_value=True, + ) as mock_setup_entry: + result = await async_configure_flow(hass, result["flow_id"], {CONF_HOST: HOST}) + + assert result["type"] == RESULT_TYPE_CREATE_ENTRY + assert result["title"] == HOST + assert result["data"] == {CONF_HOST: HOST} + await hass.async_block_till_done() + assert len(mock_setup.mock_calls) == 1 + assert len(mock_setup_entry.mock_calls) == 1 + + +async def test_form_cannot_connect(hass: HomeAssistantType) -> None: + """Test we handle cannot connect error.""" + result = await hass.config_entries.flow.async_init( + DOMAIN, context={CONF_SOURCE: SOURCE_USER} + ) + + with patch( + "homeassistant.components.directv.config_flow.DIRECTV", new=MockDirectvClass, + ), patch( + "homeassistant.components.directv.config_flow.DIRECTV.get_version", + side_effect=RequestException, + ) as mock_validate_input: + result = await async_configure_flow(hass, result["flow_id"], {CONF_HOST: HOST},) + + assert result["type"] == RESULT_TYPE_FORM + assert result["errors"] == {"base": "cannot_connect"} + + await hass.async_block_till_done() + assert len(mock_validate_input.mock_calls) == 1 + + +async def test_form_unknown_error(hass: HomeAssistantType) -> None: + """Test we handle unknown error.""" + result = await hass.config_entries.flow.async_init( + DOMAIN, context={CONF_SOURCE: SOURCE_USER} + ) + + with patch( + "homeassistant.components.directv.config_flow.DIRECTV", new=MockDirectvClass, + ), patch( + "homeassistant.components.directv.config_flow.DIRECTV.get_version", + side_effect=Exception, + ) as mock_validate_input: + result = await async_configure_flow(hass, result["flow_id"], {CONF_HOST: HOST},) + + assert result["type"] == RESULT_TYPE_FORM + assert result["errors"] == {"base": "unknown"} + + await hass.async_block_till_done() + assert len(mock_validate_input.mock_calls) == 1 + + +async def test_import(hass: HomeAssistantType) -> None: + """Test the import step.""" + with patch( + "homeassistant.components.directv.async_setup", return_value=True + ) as mock_setup, patch( + "homeassistant.components.directv.async_setup_entry", return_value=True, + ) as mock_setup_entry: + result = await async_init_flow( + hass, context={CONF_SOURCE: SOURCE_IMPORT}, data={CONF_HOST: HOST}, + ) + + assert result["type"] == RESULT_TYPE_CREATE_ENTRY + assert result["title"] == HOST + assert result["data"] == {CONF_HOST: HOST} + + await hass.async_block_till_done() + assert len(mock_setup.mock_calls) == 1 + assert len(mock_setup_entry.mock_calls) == 1 + + +async def test_ssdp_discovery(hass: HomeAssistantType) -> None: + """Test the ssdp discovery step.""" + result = await hass.config_entries.flow.async_init( + DOMAIN, + context={CONF_SOURCE: SOURCE_SSDP}, + data={ATTR_SSDP_LOCATION: SSDP_LOCATION, ATTR_UPNP_SERIAL: UPNP_SERIAL}, + ) + + assert result["type"] == RESULT_TYPE_FORM + assert result["step_id"] == "ssdp_confirm" + assert result["description_placeholders"] == {CONF_NAME: HOST} + + with patch( + "homeassistant.components.directv.async_setup", return_value=True + ) as mock_setup, patch( + "homeassistant.components.directv.async_setup_entry", return_value=True, + ) as mock_setup_entry: + result = await async_configure_flow(hass, result["flow_id"], {}) + + assert result["type"] == RESULT_TYPE_CREATE_ENTRY + assert result["title"] == HOST + assert result["data"] == {CONF_HOST: HOST} + await hass.async_block_till_done() + assert len(mock_setup.mock_calls) == 1 + assert len(mock_setup_entry.mock_calls) == 1 + + +async def test_ssdp_discovery_confirm_abort(hass: HomeAssistantType) -> None: + """Test we handle SSDP confirm cannot connect error.""" + result = await hass.config_entries.flow.async_init( + DOMAIN, + context={CONF_SOURCE: SOURCE_SSDP}, + data={ATTR_SSDP_LOCATION: SSDP_LOCATION, ATTR_UPNP_SERIAL: UPNP_SERIAL}, + ) + + with patch( + "homeassistant.components.directv.config_flow.DIRECTV", new=MockDirectvClass, + ), patch( + "homeassistant.components.directv.config_flow.DIRECTV.get_version", + side_effect=RequestException, + ) as mock_validate_input: + result = await async_configure_flow(hass, result["flow_id"], {}) + + assert result["type"] == RESULT_TYPE_ABORT + + await hass.async_block_till_done() + assert len(mock_validate_input.mock_calls) == 1 + + +async def test_ssdp_discovery_confirm_unknown_error(hass: HomeAssistantType) -> None: + """Test we handle SSDP confirm unknown error.""" + result = await hass.config_entries.flow.async_init( + DOMAIN, + context={CONF_SOURCE: SOURCE_SSDP}, + data={ATTR_SSDP_LOCATION: SSDP_LOCATION, ATTR_UPNP_SERIAL: UPNP_SERIAL}, + ) + + with patch( + "homeassistant.components.directv.config_flow.DIRECTV", new=MockDirectvClass, + ), patch( + "homeassistant.components.directv.config_flow.DIRECTV.get_version", + side_effect=Exception, + ) as mock_validate_input: + result = await async_configure_flow(hass, result["flow_id"], {}) + + assert result["type"] == RESULT_TYPE_ABORT + + await hass.async_block_till_done() + assert len(mock_validate_input.mock_calls) == 1 diff --git a/tests/components/directv/test_init.py b/tests/components/directv/test_init.py new file mode 100644 index 00000000000..02e97b9b015 --- /dev/null +++ b/tests/components/directv/test_init.py @@ -0,0 +1,48 @@ +"""Tests for the Roku integration.""" +from asynctest import patch +from requests.exceptions import RequestException + +from homeassistant.components.directv.const import DOMAIN +from homeassistant.config_entries import ( + ENTRY_STATE_LOADED, + ENTRY_STATE_NOT_LOADED, + ENTRY_STATE_SETUP_RETRY, +) +from homeassistant.helpers.typing import HomeAssistantType + +from tests.components.directv import MockDirectvClass, setup_integration + +# pylint: disable=redefined-outer-name + + +async def test_config_entry_not_ready(hass: HomeAssistantType) -> None: + """Test the DirecTV configuration entry not ready.""" + with patch( + "homeassistant.components.directv.DIRECTV", new=MockDirectvClass, + ), patch( + "homeassistant.components.directv.DIRECTV.get_locations", + side_effect=RequestException, + ): + entry = await setup_integration(hass) + + assert entry.state == ENTRY_STATE_SETUP_RETRY + + +async def test_unload_config_entry(hass: HomeAssistantType) -> None: + """Test the DirecTV configuration entry unloading.""" + with patch( + "homeassistant.components.directv.DIRECTV", new=MockDirectvClass, + ), patch( + "homeassistant.components.directv.media_player.async_setup_entry", + return_value=True, + ): + entry = await setup_integration(hass) + + assert entry.entry_id in hass.data[DOMAIN] + assert entry.state == ENTRY_STATE_LOADED + + await hass.config_entries.async_unload(entry.entry_id) + await hass.async_block_till_done() + + assert entry.entry_id not in hass.data[DOMAIN] + assert entry.state == ENTRY_STATE_NOT_LOADED diff --git a/tests/components/directv/test_media_player.py b/tests/components/directv/test_media_player.py index be805d837f5..9c06164c309 100644 --- a/tests/components/directv/test_media_player.py +++ b/tests/components/directv/test_media_player.py @@ -1,17 +1,16 @@ """The tests for the DirecTV Media player platform.""" from datetime import datetime, timedelta -from unittest.mock import call, patch +from typing import Optional -import pytest -import requests +from asynctest import patch +from pytest import fixture +from requests import RequestException from homeassistant.components.directv.media_player import ( ATTR_MEDIA_CURRENTLY_RECORDING, ATTR_MEDIA_RATING, ATTR_MEDIA_RECORDED, ATTR_MEDIA_START_TIME, - DEFAULT_DEVICE, - DEFAULT_PORT, ) from homeassistant.components.media_player.const import ( ATTR_INPUT_SOURCE, @@ -24,7 +23,7 @@ from homeassistant.components.media_player.const import ( ATTR_MEDIA_POSITION_UPDATED_AT, ATTR_MEDIA_SERIES_TITLE, ATTR_MEDIA_TITLE, - DOMAIN, + DOMAIN as MP_DOMAIN, MEDIA_TYPE_TVSHOW, SERVICE_PLAY_MEDIA, SUPPORT_NEXT_TRACK, @@ -38,10 +37,6 @@ from homeassistant.components.media_player.const import ( ) from homeassistant.const import ( ATTR_ENTITY_ID, - CONF_DEVICE, - CONF_HOST, - CONF_NAME, - CONF_PORT, SERVICE_MEDIA_NEXT_TRACK, SERVICE_MEDIA_PAUSE, SERVICE_MEDIA_PLAY, @@ -54,184 +49,143 @@ from homeassistant.const import ( STATE_PLAYING, STATE_UNAVAILABLE, ) -from homeassistant.helpers.discovery import async_load_platform -from homeassistant.setup import async_setup_component -import homeassistant.util.dt as dt_util +from homeassistant.helpers.typing import HomeAssistantType +from homeassistant.util import dt as dt_util -from tests.common import async_fire_time_changed +from tests.common import MockConfigEntry, async_fire_time_changed +from tests.components.directv import ( + CLIENT_ADDRESS, + DOMAIN, + HOST, + MOCK_GET_LOCATIONS_MULTIPLE, + RECORDING, + MockDirectvClass, + setup_integration, +) ATTR_UNIQUE_ID = "unique_id" -CLIENT_ENTITY_ID = "media_player.client_dvr" -MAIN_ENTITY_ID = "media_player.main_dvr" -IP_ADDRESS = "127.0.0.1" +CLIENT_ENTITY_ID = f"{MP_DOMAIN}.bedroom_client" +MAIN_ENTITY_ID = f"{MP_DOMAIN}.main_dvr" -DISCOVERY_INFO = {"host": IP_ADDRESS, "serial": 1234} - -LIVE = { - "callsign": "HASSTV", - "date": "20181110", - "duration": 3600, - "isOffAir": False, - "isPclocked": 1, - "isPpv": False, - "isRecording": False, - "isVod": False, - "major": 202, - "minor": 65535, - "offset": 1, - "programId": "102454523", - "rating": "No Rating", - "startTime": 1541876400, - "stationId": 3900947, - "title": "Using Home Assistant to automate your home", -} - -LOCATIONS = [{"locationName": "Main DVR", "clientAddr": DEFAULT_DEVICE}] - -RECORDING = { - "callsign": "HASSTV", - "date": "20181110", - "duration": 3600, - "isOffAir": False, - "isPclocked": 1, - "isPpv": False, - "isRecording": True, - "isVod": False, - "major": 202, - "minor": 65535, - "offset": 1, - "programId": "102454523", - "rating": "No Rating", - "startTime": 1541876400, - "stationId": 3900947, - "title": "Using Home Assistant to automate your home", - "uniqueId": "12345", - "episodeTitle": "Configure DirecTV platform.", -} - -WORKING_CONFIG = { - "media_player": { - "platform": "directv", - CONF_HOST: IP_ADDRESS, - CONF_NAME: "Main DVR", - CONF_PORT: DEFAULT_PORT, - CONF_DEVICE: DEFAULT_DEVICE, - } -} +# pylint: disable=redefined-outer-name -@pytest.fixture -def client_dtv(): +@fixture +def client_dtv() -> MockDirectvClass: """Fixture for a client device.""" - mocked_dtv = MockDirectvClass("mock_ip") + mocked_dtv = MockDirectvClass(HOST, clientAddr=CLIENT_ADDRESS) mocked_dtv.attributes = RECORDING - mocked_dtv._standby = False + mocked_dtv._standby = False # pylint: disable=protected-access return mocked_dtv -@pytest.fixture -def main_dtv(): - """Fixture for main DVR.""" - return MockDirectvClass("mock_ip") - - -@pytest.fixture -def dtv_side_effect(client_dtv, main_dtv): - """Fixture to create DIRECTV instance for main and client.""" - - def mock_dtv(ip, port, client_addr="0"): - if client_addr != "0": - mocked_dtv = client_dtv - else: - mocked_dtv = main_dtv - mocked_dtv._host = ip - mocked_dtv._port = port - mocked_dtv._device = client_addr - return mocked_dtv - - return mock_dtv - - -@pytest.fixture -def mock_now(): +@fixture +def mock_now() -> datetime: """Fixture for dtutil.now.""" return dt_util.utcnow() -@pytest.fixture -def platforms(hass, dtv_side_effect, mock_now): - """Fixture for setting up test platforms.""" - config = { - "media_player": [ - { - "platform": "directv", - "name": "Main DVR", - "host": IP_ADDRESS, - "port": DEFAULT_PORT, - "device": DEFAULT_DEVICE, - }, - { - "platform": "directv", - "name": "Client DVR", - "host": IP_ADDRESS, - "port": DEFAULT_PORT, - "device": "2CA17D1CD30X", - }, - ] - } - +async def setup_directv(hass: HomeAssistantType) -> MockConfigEntry: + """Set up mock DirecTV integration.""" with patch( - "homeassistant.components.directv.media_player.DIRECTV", - side_effect=dtv_side_effect, - ), patch("homeassistant.util.dt.utcnow", return_value=mock_now): - hass.loop.run_until_complete(async_setup_component(hass, DOMAIN, config)) - hass.loop.run_until_complete(hass.async_block_till_done()) - yield + "homeassistant.components.directv.DIRECTV", new=MockDirectvClass, + ): + return await setup_integration(hass) -async def async_turn_on(hass, entity_id=None): +async def setup_directv_with_instance_error(hass: HomeAssistantType) -> MockConfigEntry: + """Set up mock DirecTV integration.""" + with patch( + "homeassistant.components.directv.DIRECTV", new=MockDirectvClass, + ), patch( + "homeassistant.components.directv.DIRECTV.get_locations", + return_value=MOCK_GET_LOCATIONS_MULTIPLE, + ), patch( + "homeassistant.components.directv.media_player.get_dtv_instance", + return_value=None, + ): + return await setup_integration(hass) + + +async def setup_directv_with_locations( + hass: HomeAssistantType, client_dtv: MockDirectvClass, +) -> MockConfigEntry: + """Set up mock DirecTV integration.""" + with patch( + "homeassistant.components.directv.DIRECTV", new=MockDirectvClass, + ), patch( + "homeassistant.components.directv.DIRECTV.get_locations", + return_value=MOCK_GET_LOCATIONS_MULTIPLE, + ), patch( + "homeassistant.components.directv.media_player.get_dtv_instance", + return_value=client_dtv, + ): + return await setup_integration(hass) + + +async def async_turn_on( + hass: HomeAssistantType, entity_id: Optional[str] = None +) -> None: """Turn on specified media player or all.""" data = {ATTR_ENTITY_ID: entity_id} if entity_id else {} - await hass.services.async_call(DOMAIN, SERVICE_TURN_ON, data) + await hass.services.async_call(MP_DOMAIN, SERVICE_TURN_ON, data) -async def async_turn_off(hass, entity_id=None): +async def async_turn_off( + hass: HomeAssistantType, entity_id: Optional[str] = None +) -> None: """Turn off specified media player or all.""" data = {ATTR_ENTITY_ID: entity_id} if entity_id else {} - await hass.services.async_call(DOMAIN, SERVICE_TURN_OFF, data) + await hass.services.async_call(MP_DOMAIN, SERVICE_TURN_OFF, data) -async def async_media_pause(hass, entity_id=None): +async def async_media_pause( + hass: HomeAssistantType, entity_id: Optional[str] = None +) -> None: """Send the media player the command for pause.""" data = {ATTR_ENTITY_ID: entity_id} if entity_id else {} - await hass.services.async_call(DOMAIN, SERVICE_MEDIA_PAUSE, data) + await hass.services.async_call(MP_DOMAIN, SERVICE_MEDIA_PAUSE, data) -async def async_media_play(hass, entity_id=None): +async def async_media_play( + hass: HomeAssistantType, entity_id: Optional[str] = None +) -> None: """Send the media player the command for play/pause.""" data = {ATTR_ENTITY_ID: entity_id} if entity_id else {} - await hass.services.async_call(DOMAIN, SERVICE_MEDIA_PLAY, data) + await hass.services.async_call(MP_DOMAIN, SERVICE_MEDIA_PLAY, data) -async def async_media_stop(hass, entity_id=None): +async def async_media_stop( + hass: HomeAssistantType, entity_id: Optional[str] = None +) -> None: """Send the media player the command for stop.""" data = {ATTR_ENTITY_ID: entity_id} if entity_id else {} - await hass.services.async_call(DOMAIN, SERVICE_MEDIA_STOP, data) + await hass.services.async_call(MP_DOMAIN, SERVICE_MEDIA_STOP, data) -async def async_media_next_track(hass, entity_id=None): +async def async_media_next_track( + hass: HomeAssistantType, entity_id: Optional[str] = None +) -> None: """Send the media player the command for next track.""" data = {ATTR_ENTITY_ID: entity_id} if entity_id else {} - await hass.services.async_call(DOMAIN, SERVICE_MEDIA_NEXT_TRACK, data) + await hass.services.async_call(MP_DOMAIN, SERVICE_MEDIA_NEXT_TRACK, data) -async def async_media_previous_track(hass, entity_id=None): +async def async_media_previous_track( + hass: HomeAssistantType, entity_id: Optional[str] = None +) -> None: """Send the media player the command for prev track.""" data = {ATTR_ENTITY_ID: entity_id} if entity_id else {} - await hass.services.async_call(DOMAIN, SERVICE_MEDIA_PREVIOUS_TRACK, data) + await hass.services.async_call(MP_DOMAIN, SERVICE_MEDIA_PREVIOUS_TRACK, data) -async def async_play_media(hass, media_type, media_id, entity_id=None, enqueue=None): +async def async_play_media( + hass: HomeAssistantType, + media_type: str, + media_id: str, + entity_id: Optional[str] = None, + enqueue: Optional[str] = None, +) -> None: """Send the media player the command for playing media.""" data = {ATTR_MEDIA_CONTENT_TYPE: media_type, ATTR_MEDIA_CONTENT_ID: media_id} @@ -241,191 +195,37 @@ async def async_play_media(hass, media_type, media_id, entity_id=None, enqueue=N if enqueue: data[ATTR_MEDIA_ENQUEUE] = enqueue - await hass.services.async_call(DOMAIN, SERVICE_PLAY_MEDIA, data) + await hass.services.async_call(MP_DOMAIN, SERVICE_PLAY_MEDIA, data) -class MockDirectvClass: - """A fake DirecTV DVR device.""" - - def __init__(self, ip, port=8080, clientAddr="0"): - """Initialize the fake DirecTV device.""" - self._host = ip - self._port = port - self._device = clientAddr - self._standby = True - self._play = False - - self._locations = LOCATIONS - - self.attributes = LIVE - - def get_locations(self): - """Mock for get_locations method.""" - test_locations = { - "locations": self._locations, - "status": { - "code": 200, - "commandResult": 0, - "msg": "OK.", - "query": "/info/getLocations", - }, - } - - return test_locations - - def get_serial_num(self): - """Mock for get_serial_num method.""" - test_serial_num = { - "serialNum": "9999999999", - "status": { - "code": 200, - "commandResult": 0, - "msg": "OK.", - "query": "/info/getSerialNum", - }, - } - - return test_serial_num - - def get_standby(self): - """Mock for get_standby method.""" - return self._standby - - def get_tuned(self): - """Mock for get_tuned method.""" - if self._play: - self.attributes["offset"] = self.attributes["offset"] + 1 - - test_attributes = self.attributes - test_attributes["status"] = { - "code": 200, - "commandResult": 0, - "msg": "OK.", - "query": "/tv/getTuned", - } - return test_attributes - - def get_version(self): - """Mock for get_version method.""" - test_version = { - "accessCardId": "0021-1495-6572", - "receiverId": "0288 7745 5858", - "status": { - "code": 200, - "commandResult": 0, - "msg": "OK.", - "query": "/info/getVersion", - }, - "stbSoftwareVersion": "0x4ed7", - "systemTime": 1281625203, - "version": "1.2", - } - - return test_version - - def key_press(self, keypress): - """Mock for key_press method.""" - if keypress == "poweron": - self._standby = False - self._play = True - elif keypress == "poweroff": - self._standby = True - self._play = False - elif keypress == "play": - self._play = True - elif keypress == "pause" or keypress == "stop": - self._play = False - - def tune_channel(self, source): - """Mock for tune_channel method.""" - self.attributes["major"] = int(source) +async def test_setup(hass: HomeAssistantType) -> None: + """Test setup with basic config.""" + await setup_directv(hass) + assert hass.states.get(MAIN_ENTITY_ID) -async def test_setup_platform_config(hass): - """Test setting up the platform from configuration.""" - with patch( - "homeassistant.components.directv.media_player.DIRECTV", new=MockDirectvClass - ): +async def test_setup_with_multiple_locations( + hass: HomeAssistantType, client_dtv: MockDirectvClass +) -> None: + """Test setup with basic config with client location.""" + await setup_directv_with_locations(hass, client_dtv) - await async_setup_component(hass, DOMAIN, WORKING_CONFIG) - await hass.async_block_till_done() - - state = hass.states.get(MAIN_ENTITY_ID) - assert state - assert len(hass.states.async_entity_ids("media_player")) == 1 + assert hass.states.get(MAIN_ENTITY_ID) + assert hass.states.get(CLIENT_ENTITY_ID) -async def test_setup_platform_discover(hass): - """Test setting up the platform from discovery.""" - with patch( - "homeassistant.components.directv.media_player.DIRECTV", new=MockDirectvClass - ): +async def test_setup_with_instance_error(hass: HomeAssistantType) -> None: + """Test setup with basic config with client location that results in instance error.""" + await setup_directv_with_instance_error(hass) - hass.async_create_task( - async_load_platform( - hass, DOMAIN, "directv", DISCOVERY_INFO, {"media_player": {}} - ) - ) - await hass.async_block_till_done() - - state = hass.states.get(MAIN_ENTITY_ID) - assert state - assert len(hass.states.async_entity_ids("media_player")) == 1 + assert hass.states.get(MAIN_ENTITY_ID) + assert hass.states.async_entity_ids(MP_DOMAIN) == [MAIN_ENTITY_ID] -async def test_setup_platform_discover_duplicate(hass): - """Test setting up the platform from discovery.""" - with patch( - "homeassistant.components.directv.media_player.DIRECTV", new=MockDirectvClass - ): - - await async_setup_component(hass, DOMAIN, WORKING_CONFIG) - await hass.async_block_till_done() - hass.async_create_task( - async_load_platform( - hass, DOMAIN, "directv", DISCOVERY_INFO, {"media_player": {}} - ) - ) - await hass.async_block_till_done() - - state = hass.states.get(MAIN_ENTITY_ID) - assert state - assert len(hass.states.async_entity_ids("media_player")) == 1 - - -async def test_setup_platform_discover_client(hass): - """Test setting up the platform from discovery.""" - LOCATIONS.append({"locationName": "Client 1", "clientAddr": "1"}) - LOCATIONS.append({"locationName": "Client 2", "clientAddr": "2"}) - - with patch( - "homeassistant.components.directv.media_player.DIRECTV", new=MockDirectvClass - ): - - await async_setup_component(hass, DOMAIN, WORKING_CONFIG) - await hass.async_block_till_done() - - hass.async_create_task( - async_load_platform( - hass, DOMAIN, "directv", DISCOVERY_INFO, {"media_player": {}} - ) - ) - await hass.async_block_till_done() - - del LOCATIONS[-1] - del LOCATIONS[-1] - state = hass.states.get(MAIN_ENTITY_ID) - assert state - state = hass.states.get("media_player.client_1") - assert state - state = hass.states.get("media_player.client_2") - assert state - - assert len(hass.states.async_entity_ids("media_player")) == 3 - - -async def test_unique_id(hass, platforms): +async def test_unique_id(hass: HomeAssistantType, client_dtv: MockDirectvClass) -> None: """Test unique id.""" + await setup_directv_with_locations(hass, client_dtv) + entity_registry = await hass.helpers.entity_registry.async_get_registry() main = entity_registry.async_get(MAIN_ENTITY_ID) @@ -435,8 +235,12 @@ async def test_unique_id(hass, platforms): assert client.unique_id == "2CA17D1CD30X" -async def test_supported_features(hass, platforms): +async def test_supported_features( + hass: HomeAssistantType, client_dtv: MockDirectvClass +) -> None: """Test supported features.""" + await setup_directv_with_locations(hass, client_dtv) + # Features supported for main DVR state = hass.states.get(MAIN_ENTITY_ID) assert ( @@ -464,8 +268,12 @@ async def test_supported_features(hass, platforms): ) -async def test_check_attributes(hass, platforms, mock_now): +async def test_check_attributes( + hass: HomeAssistantType, mock_now: dt_util.dt.datetime, client_dtv: MockDirectvClass +) -> None: """Test attributes.""" + await setup_directv_with_locations(hass, client_dtv) + next_update = mock_now + timedelta(minutes=5) with patch("homeassistant.util.dt.utcnow", return_value=next_update): async_fire_time_changed(hass, next_update) @@ -512,8 +320,12 @@ async def test_check_attributes(hass, platforms, mock_now): assert state.attributes.get(ATTR_MEDIA_POSITION_UPDATED_AT) == next_update -async def test_main_services(hass, platforms, main_dtv, mock_now): +async def test_main_services( + hass: HomeAssistantType, mock_now: dt_util.dt.datetime, client_dtv: MockDirectvClass +) -> None: """Test the different services.""" + await setup_directv_with_locations(hass, client_dtv) + next_update = mock_now + timedelta(minutes=5) with patch("homeassistant.util.dt.utcnow", return_value=next_update): async_fire_time_changed(hass, next_update) @@ -522,77 +334,50 @@ async def test_main_services(hass, platforms, main_dtv, mock_now): state = hass.states.get(MAIN_ENTITY_ID) assert state.state == STATE_OFF - # All these should call key_press in our class. - with patch.object( - main_dtv, "key_press", wraps=main_dtv.key_press - ) as mock_key_press, patch.object( - main_dtv, "tune_channel", wraps=main_dtv.tune_channel - ) as mock_tune_channel, patch.object( - main_dtv, "get_tuned", wraps=main_dtv.get_tuned - ) as mock_get_tuned, patch.object( - main_dtv, "get_standby", wraps=main_dtv.get_standby - ) as mock_get_standby: + # Turn main DVR on. When turning on DVR is playing. + await async_turn_on(hass, MAIN_ENTITY_ID) + await hass.async_block_till_done() + state = hass.states.get(MAIN_ENTITY_ID) + assert state.state == STATE_PLAYING - # Turn main DVR on. When turning on DVR is playing. - await async_turn_on(hass, MAIN_ENTITY_ID) - await hass.async_block_till_done() - assert mock_key_press.called - assert mock_key_press.call_args == call("poweron") - state = hass.states.get(MAIN_ENTITY_ID) - assert state.state == STATE_PLAYING + # Pause live TV. + await async_media_pause(hass, MAIN_ENTITY_ID) + await hass.async_block_till_done() + state = hass.states.get(MAIN_ENTITY_ID) + assert state.state == STATE_PAUSED - # Pause live TV. - await async_media_pause(hass, MAIN_ENTITY_ID) - await hass.async_block_till_done() - assert mock_key_press.called - assert mock_key_press.call_args == call("pause") - state = hass.states.get(MAIN_ENTITY_ID) - assert state.state == STATE_PAUSED + # Start play again for live TV. + await async_media_play(hass, MAIN_ENTITY_ID) + await hass.async_block_till_done() + state = hass.states.get(MAIN_ENTITY_ID) + assert state.state == STATE_PLAYING - # Start play again for live TV. - await async_media_play(hass, MAIN_ENTITY_ID) - await hass.async_block_till_done() - assert mock_key_press.called - assert mock_key_press.call_args == call("play") - state = hass.states.get(MAIN_ENTITY_ID) - assert state.state == STATE_PLAYING + # Change channel, currently it should be 202 + assert state.attributes.get("source") == 202 + await async_play_media(hass, "channel", 7, MAIN_ENTITY_ID) + await hass.async_block_till_done() + state = hass.states.get(MAIN_ENTITY_ID) + assert state.attributes.get("source") == 7 - # Change channel, currently it should be 202 - assert state.attributes.get("source") == 202 - await async_play_media(hass, "channel", 7, MAIN_ENTITY_ID) - await hass.async_block_till_done() - assert mock_tune_channel.called - assert mock_tune_channel.call_args == call("7") - state = hass.states.get(MAIN_ENTITY_ID) - assert state.attributes.get("source") == 7 + # Stop live TV. + await async_media_stop(hass, MAIN_ENTITY_ID) + await hass.async_block_till_done() + state = hass.states.get(MAIN_ENTITY_ID) + assert state.state == STATE_PAUSED - # Stop live TV. - await async_media_stop(hass, MAIN_ENTITY_ID) - await hass.async_block_till_done() - assert mock_key_press.called - assert mock_key_press.call_args == call("stop") - state = hass.states.get(MAIN_ENTITY_ID) - assert state.state == STATE_PAUSED - - # Turn main DVR off. - await async_turn_off(hass, MAIN_ENTITY_ID) - await hass.async_block_till_done() - assert mock_key_press.called - assert mock_key_press.call_args == call("poweroff") - state = hass.states.get(MAIN_ENTITY_ID) - assert state.state == STATE_OFF - - # There should have been 6 calls to check if DVR is in standby - assert main_dtv.get_standby.call_count == 6 - assert mock_get_standby.call_count == 6 - # There should be 5 calls to get current info (only 1 time it will - # not be called as DVR is in standby.) - assert main_dtv.get_tuned.call_count == 5 - assert mock_get_tuned.call_count == 5 + # Turn main DVR off. + await async_turn_off(hass, MAIN_ENTITY_ID) + await hass.async_block_till_done() + state = hass.states.get(MAIN_ENTITY_ID) + assert state.state == STATE_OFF -async def test_available(hass, platforms, main_dtv, mock_now): +async def test_available( + hass: HomeAssistantType, mock_now: dt_util.dt.datetime, client_dtv: MockDirectvClass +) -> None: """Test available status.""" + entry = await setup_directv_with_locations(hass, client_dtv) + next_update = mock_now + timedelta(minutes=5) with patch("homeassistant.util.dt.utcnow", return_value=next_update): async_fire_time_changed(hass, next_update) @@ -602,11 +387,17 @@ async def test_available(hass, platforms, main_dtv, mock_now): state = hass.states.get(MAIN_ENTITY_ID) assert state.state != STATE_UNAVAILABLE + assert hass.data[DOMAIN] + assert hass.data[DOMAIN][entry.entry_id] + assert hass.data[DOMAIN][entry.entry_id]["client"] + + main_dtv = hass.data[DOMAIN][entry.entry_id]["client"] + # Make update fail 1st time next_update = next_update + timedelta(minutes=5) - with patch.object( - main_dtv, "get_standby", side_effect=requests.RequestException - ), patch("homeassistant.util.dt.utcnow", return_value=next_update): + with patch.object(main_dtv, "get_standby", side_effect=RequestException), patch( + "homeassistant.util.dt.utcnow", return_value=next_update + ): async_fire_time_changed(hass, next_update) await hass.async_block_till_done() @@ -615,9 +406,9 @@ async def test_available(hass, platforms, main_dtv, mock_now): # Make update fail 2nd time within 1 minute next_update = next_update + timedelta(seconds=30) - with patch.object( - main_dtv, "get_standby", side_effect=requests.RequestException - ), patch("homeassistant.util.dt.utcnow", return_value=next_update): + with patch.object(main_dtv, "get_standby", side_effect=RequestException), patch( + "homeassistant.util.dt.utcnow", return_value=next_update + ): async_fire_time_changed(hass, next_update) await hass.async_block_till_done() @@ -626,9 +417,9 @@ async def test_available(hass, platforms, main_dtv, mock_now): # Make update fail 3rd time more then a minute after 1st failure next_update = next_update + timedelta(minutes=1) - with patch.object( - main_dtv, "get_standby", side_effect=requests.RequestException - ), patch("homeassistant.util.dt.utcnow", return_value=next_update): + with patch.object(main_dtv, "get_standby", side_effect=RequestException), patch( + "homeassistant.util.dt.utcnow", return_value=next_update + ): async_fire_time_changed(hass, next_update) await hass.async_block_till_done() @@ -640,5 +431,6 @@ async def test_available(hass, platforms, main_dtv, mock_now): with patch("homeassistant.util.dt.utcnow", return_value=next_update): async_fire_time_changed(hass, next_update) await hass.async_block_till_done() + state = hass.states.get(MAIN_ENTITY_ID) assert state.state != STATE_UNAVAILABLE