From ecca51b16bd806815bfe0ddec4a09b33a2bf5e30 Mon Sep 17 00:00:00 2001 From: ehendrix23 Date: Sat, 1 Dec 2018 02:28:27 -0700 Subject: [PATCH] Add tests for directv platform (#18590) * Create test for platform Created test for platform. Added media_stop to common.py test * Multiple improvements Fixed lint issue in common.py Fixed lint issues in test_directv.py Improved patching import using modile_patcher.start() and stop() Added asserts for service calls. * Updates based on Martin's review Updates based on Martin's review. * Updated test based on PR#18474 Updated test to use service play_media instead of select_source based on change from PR18474 * Lint issues Lint issues * Further updates based on feedback Updates based on feedback provided. * Using async_load_platform for discovery test Using async_load_platform for discovery tests. Added asserts to ensure entities are created with correct names. * Used HASS event_loop to setup component Use HASS event_loop to setup the component async. * Updated to use state machine for # entities Updated to use state machine to count # entities instead of entities. * Use hass.loop instead of getting current loop Small update to use hass.loop instead, thanks Martin! * Forgot to remove asyncio Removed asyncio import. * Added fixtures Added fixtures. * Remove not needed updates and assertions * Return mocked dtv instance from side_effect * Fix return correct fixture instance * Clean up assertions * Fix remaining patches * Mock time when setting up component in fixture * Patch time correctly * Attribute _last_update should return utcnow --- .../components/media_player/directv.py | 4 +- tests/components/media_player/common.py | 13 +- tests/components/media_player/test_directv.py | 535 ++++++++++++++++++ 3 files changed, 547 insertions(+), 5 deletions(-) create mode 100644 tests/components/media_player/test_directv.py diff --git a/homeassistant/components/media_player/directv.py b/homeassistant/components/media_player/directv.py index 7a1e240d82e..d8c67e372b2 100644 --- a/homeassistant/components/media_player/directv.py +++ b/homeassistant/components/media_player/directv.py @@ -162,8 +162,8 @@ class DirecTvDevice(MediaPlayerDevice): self._current['offset'] self._assumed_state = self._is_recorded self._last_position = self._current['offset'] - self._last_update = dt_util.now() if not self._paused or\ - self._last_update is None else self._last_update + self._last_update = dt_util.utcnow() if not self._paused \ + or self._last_update is None else self._last_update else: self._available = False except requests.RequestException as ex: diff --git a/tests/components/media_player/common.py b/tests/components/media_player/common.py index 3f4d4cb9f24..2174967eae5 100644 --- a/tests/components/media_player/common.py +++ b/tests/components/media_player/common.py @@ -11,9 +11,9 @@ from homeassistant.components.media_player import ( from homeassistant.const import ( ATTR_ENTITY_ID, SERVICE_MEDIA_NEXT_TRACK, SERVICE_MEDIA_PAUSE, SERVICE_MEDIA_PLAY, SERVICE_MEDIA_PLAY_PAUSE, SERVICE_MEDIA_PREVIOUS_TRACK, - SERVICE_MEDIA_SEEK, SERVICE_TOGGLE, SERVICE_TURN_OFF, SERVICE_TURN_ON, - SERVICE_VOLUME_DOWN, SERVICE_VOLUME_MUTE, SERVICE_VOLUME_SET, - SERVICE_VOLUME_UP) + SERVICE_MEDIA_SEEK, SERVICE_MEDIA_STOP, SERVICE_TOGGLE, SERVICE_TURN_OFF, + SERVICE_TURN_ON, SERVICE_VOLUME_DOWN, SERVICE_VOLUME_MUTE, + SERVICE_VOLUME_SET, SERVICE_VOLUME_UP) from homeassistant.loader import bind_hass @@ -95,6 +95,13 @@ def media_pause(hass, entity_id=None): hass.services.call(DOMAIN, SERVICE_MEDIA_PAUSE, data) +@bind_hass +def media_stop(hass, entity_id=None): + """Send the media player the command for stop.""" + data = {ATTR_ENTITY_ID: entity_id} if entity_id else {} + hass.services.call(DOMAIN, SERVICE_MEDIA_STOP, data) + + @bind_hass def media_next_track(hass, entity_id=None): """Send the media player the command for next track.""" diff --git a/tests/components/media_player/test_directv.py b/tests/components/media_player/test_directv.py new file mode 100644 index 00000000000..951f1319cc0 --- /dev/null +++ b/tests/components/media_player/test_directv.py @@ -0,0 +1,535 @@ +"""The tests for the DirecTV Media player platform.""" +from unittest.mock import call, patch + +from datetime import datetime, timedelta +import requests +import pytest + +import homeassistant.components.media_player as mp +from homeassistant.components.media_player import ( + ATTR_MEDIA_CONTENT_ID, ATTR_MEDIA_CONTENT_TYPE, ATTR_MEDIA_ENQUEUE, DOMAIN, + SERVICE_PLAY_MEDIA) +from homeassistant.components.media_player.directv import ( + ATTR_MEDIA_CURRENTLY_RECORDING, ATTR_MEDIA_RATING, ATTR_MEDIA_RECORDED, + ATTR_MEDIA_START_TIME, DEFAULT_DEVICE, DEFAULT_PORT) +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, + SERVICE_MEDIA_PREVIOUS_TRACK, SERVICE_MEDIA_STOP, SERVICE_TURN_OFF, + SERVICE_TURN_ON, STATE_OFF, STATE_PAUSED, 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 tests.common import MockDependency, async_fire_time_changed + +CLIENT_ENTITY_ID = 'media_player.client_dvr' +MAIN_ENTITY_ID = 'media_player.main_dvr' +IP_ADDRESS = '127.0.0.1' + +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 + } +} + + +@pytest.fixture +def client_dtv(): + """Fixture for a client device.""" + mocked_dtv = MockDirectvClass('mock_ip') + mocked_dtv.attributes = RECORDING + mocked_dtv._standby = False + 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): + 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 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': '1' + }] + } + + with MockDependency('DirectPy'), \ + patch('DirectPy.DIRECTV', side_effect=dtv_side_effect), \ + patch('homeassistant.util.dt.utcnow', return_value=mock_now): + hass.loop.run_until_complete(async_setup_component( + hass, mp.DOMAIN, config)) + hass.loop.run_until_complete(hass.async_block_till_done()) + yield + + +async def async_turn_on(hass, entity_id=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) + + +async def async_turn_off(hass, entity_id=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) + + +async def async_media_pause(hass, entity_id=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) + + +async def async_media_play(hass, entity_id=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) + + +async def async_media_stop(hass, entity_id=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) + + +async def async_media_next_track(hass, entity_id=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) + + +async def async_media_previous_track(hass, entity_id=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) + + +async def async_play_media(hass, media_type, media_id, entity_id=None, + enqueue=None): + """Send the media player the command for playing media.""" + data = {ATTR_MEDIA_CONTENT_TYPE: media_type, + ATTR_MEDIA_CONTENT_ID: media_id} + + if entity_id: + data[ATTR_ENTITY_ID] = entity_id + + if enqueue: + data[ATTR_MEDIA_ENQUEUE] = enqueue + + await hass.services.async_call(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_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 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_platform_config(hass): + """Test setting up the platform from configuration.""" + with MockDependency('DirectPy'), \ + patch('DirectPy.DIRECTV', new=MockDirectvClass): + + await async_setup_component(hass, mp.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 + + +async def test_setup_platform_discover(hass): + """Test setting up the platform from discovery.""" + with MockDependency('DirectPy'), \ + patch('DirectPy.DIRECTV', new=MockDirectvClass): + + hass.async_create_task( + async_load_platform(hass, mp.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_duplicate(hass): + """Test setting up the platform from discovery.""" + with MockDependency('DirectPy'), \ + patch('DirectPy.DIRECTV', new=MockDirectvClass): + + await async_setup_component(hass, mp.DOMAIN, WORKING_CONFIG) + await hass.async_block_till_done() + hass.async_create_task( + async_load_platform(hass, mp.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 MockDependency('DirectPy'), \ + patch('DirectPy.DIRECTV', new=MockDirectvClass): + + await async_setup_component(hass, mp.DOMAIN, WORKING_CONFIG) + await hass.async_block_till_done() + + hass.async_create_task( + async_load_platform(hass, mp.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_supported_features(hass, platforms): + """Test supported features.""" + # Features supported for main DVR + state = hass.states.get(MAIN_ENTITY_ID) + assert mp.SUPPORT_PAUSE | mp.SUPPORT_TURN_ON | mp.SUPPORT_TURN_OFF |\ + mp.SUPPORT_PLAY_MEDIA | mp.SUPPORT_STOP | mp.SUPPORT_NEXT_TRACK |\ + mp.SUPPORT_PREVIOUS_TRACK | mp.SUPPORT_PLAY ==\ + state.attributes.get('supported_features') + + # Feature supported for clients. + state = hass.states.get(CLIENT_ENTITY_ID) + assert mp.SUPPORT_PAUSE |\ + mp.SUPPORT_PLAY_MEDIA | mp.SUPPORT_STOP | mp.SUPPORT_NEXT_TRACK |\ + mp.SUPPORT_PREVIOUS_TRACK | mp.SUPPORT_PLAY ==\ + state.attributes.get('supported_features') + + +async def test_check_attributes(hass, platforms, mock_now): + """Test attributes.""" + 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() + + # 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() + + state = hass.states.get(CLIENT_ENTITY_ID) + assert state.state == STATE_PLAYING + + assert state.attributes.get(mp.ATTR_MEDIA_CONTENT_ID) == \ + RECORDING['programId'] + assert state.attributes.get(mp.ATTR_MEDIA_CONTENT_TYPE) == \ + mp.MEDIA_TYPE_TVSHOW + assert state.attributes.get(mp.ATTR_MEDIA_DURATION) == \ + RECORDING['duration'] + assert state.attributes.get(mp.ATTR_MEDIA_POSITION) == 2 + assert state.attributes.get( + mp.ATTR_MEDIA_POSITION_UPDATED_AT) == next_update + assert state.attributes.get(mp.ATTR_MEDIA_TITLE) == RECORDING['title'] + assert state.attributes.get(mp.ATTR_MEDIA_SERIES_TITLE) == \ + RECORDING['episodeTitle'] + assert state.attributes.get(mp.ATTR_MEDIA_CHANNEL) == \ + "{} ({})".format(RECORDING['callsign'], RECORDING['major']) + assert state.attributes.get(mp.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_RECORDED) + assert state.attributes.get(ATTR_MEDIA_START_TIME) == \ + datetime(2018, 11, 10, 19, 0, tzinfo=dt_util.UTC) + + # 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)): + 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( + mp.ATTR_MEDIA_POSITION_UPDATED_AT) == next_update + + +async def test_main_services(hass, platforms, main_dtv, mock_now): + """Test the different services.""" + 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() + # DVR starts in off state. + 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() + 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() + 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() + 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() + 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() + 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 + + +async def test_available(hass, platforms, main_dtv, mock_now): + """Test available status.""" + 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() + + # Confirm service is currently set to available. + state = hass.states.get(MAIN_ENTITY_ID) + assert state.state != STATE_UNAVAILABLE + + # Make update fail (i.e. DVR offline) + 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): + 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 + + # Recheck state, update should work again. + next_update = next_update + timedelta(minutes=5) + with patch('homeassistant.util.dt.utcnow', return_value=next_update): + async_fire_time_changed(hass, next_update) + await hass.async_block_till_done() + state = hass.states.get(MAIN_ENTITY_ID) + assert state.state != STATE_UNAVAILABLE