437 lines
18 KiB
Python
437 lines
18 KiB
Python
import asyncio
|
|
import json
|
|
from pathlib import Path
|
|
import sys
|
|
import tempfile
|
|
from types import SimpleNamespace
|
|
import unittest
|
|
from unittest.mock import AsyncMock, Mock, patch
|
|
|
|
import aiohttp
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parents[1] / 'apps'))
|
|
sys.path.insert(0, str(Path(__file__).resolve().parents[1] / 'apps' / 'ad_toolbox'))
|
|
from samsungtheframeasync import SamsungTheFrame, TVError, _TVConnection
|
|
|
|
|
|
def message(data):
|
|
return SimpleNamespace(type=aiohttp.WSMsgType.TEXT, data=json.dumps(data))
|
|
|
|
|
|
class FakeSocket:
|
|
def __init__(self, ready=True):
|
|
self.messages = asyncio.Queue()
|
|
self.closed = False
|
|
self.sent = []
|
|
self.respond = True
|
|
self.messages.put_nowait(message({'event': 'ms.channel.connect', 'data': {}}))
|
|
if ready:
|
|
self.messages.put_nowait(message({'event': 'ms.channel.ready'}))
|
|
|
|
async def receive(self):
|
|
return await self.messages.get()
|
|
|
|
def __aiter__(self):
|
|
return self
|
|
|
|
async def __anext__(self):
|
|
msg = await self.receive()
|
|
if msg.type == aiohttp.WSMsgType.CLOSED:
|
|
raise StopAsyncIteration
|
|
return msg
|
|
|
|
async def send_json(self, payload):
|
|
self.sent.append(payload)
|
|
if payload['method'] == 'ms.channel.emit' and self.respond:
|
|
request = json.loads(payload['params']['data'])
|
|
self.push({'event': 'artmode_status', 'value': 'on',
|
|
'request_id': request['request_id']})
|
|
|
|
def push(self, data):
|
|
self.messages.put_nowait(message({'event': 'd2d_service_message',
|
|
'data': json.dumps(data)}))
|
|
|
|
async def close(self):
|
|
self.closed = True
|
|
self.messages.put_nowait(SimpleNamespace(type=aiohttp.WSMsgType.CLOSED))
|
|
|
|
|
|
class ConnectionTests(unittest.IsolatedAsyncioTestCase):
|
|
async def asyncSetUp(self):
|
|
self.temp = tempfile.TemporaryDirectory()
|
|
self.ws = FakeSocket()
|
|
self.session = Mock()
|
|
self.session.ws_connect = AsyncMock(return_value=self.ws)
|
|
self.events = Mock()
|
|
self.client = _TVConnection(self.session, '192.0.2.1',
|
|
Path(self.temp.name) / 'token', self.events, .05)
|
|
|
|
async def asyncTearDown(self):
|
|
await self.client.close()
|
|
self.temp.cleanup()
|
|
|
|
async def test_art_requests_reuse_socket_and_reader(self):
|
|
first = await self.client.art_request('get_artmode_status')
|
|
reader = self.client.readers['art']
|
|
second = await self.client.art_request('get_artmode_status')
|
|
self.assertEqual(first['value'], 'on')
|
|
self.assertEqual(second['value'], 'on')
|
|
self.session.ws_connect.assert_awaited_once()
|
|
self.assertIs(self.client.readers['art'], reader)
|
|
self.assertFalse(reader.done())
|
|
self.assertEqual(self.client.pending, {})
|
|
self.events.assert_not_called() # Query responses applied by refresh.
|
|
|
|
async def test_unsolicited_notifications_delivered_while_idle(self):
|
|
await self.client.connect('art')
|
|
self.ws.push({'event': 'art_mode_changed', 'status': 'off'})
|
|
await asyncio.sleep(.01)
|
|
self.events.assert_called_once_with({'event': 'art_mode_changed', 'status': 'off'})
|
|
|
|
async def test_wrong_request_id_cannot_satisfy_request(self):
|
|
self.ws.respond = False
|
|
task = asyncio.create_task(self.client.art_request('get_artmode_status'))
|
|
await asyncio.sleep(.01)
|
|
self.ws.push({'event': 'artmode_status', 'value': 'off', 'id': 'wrong'})
|
|
with self.assertRaises(TimeoutError):
|
|
await task
|
|
self.assertEqual(self.client.pending, {})
|
|
self.assertTrue(self.ws.closed)
|
|
|
|
async def test_disconnect_fails_pending_request_without_waiting_for_timeout(self):
|
|
self.ws.respond = False
|
|
task = asyncio.create_task(self.client.art_request('get_artmode_status'))
|
|
await asyncio.sleep(.01)
|
|
await self.ws.close()
|
|
with self.assertRaises(TVError):
|
|
await task
|
|
self.assertEqual(self.client.pending, {})
|
|
|
|
async def test_request_rejection_is_not_mistaken_for_state(self):
|
|
self.ws.respond = False
|
|
task = asyncio.create_task(self.client.art_request('get_artmode_status'))
|
|
await asyncio.sleep(.01)
|
|
request = json.loads(self.ws.sent[0]['params']['data'])
|
|
self.ws.push({'event': 'error', 'request_id': request['request_id'], 'error_code': -1})
|
|
with self.assertRaises(TVError):
|
|
await task
|
|
|
|
async def test_cancel_request_removes_pending_future(self):
|
|
self.ws.respond = False
|
|
task = asyncio.create_task(self.client.art_request('get_artmode_status'))
|
|
await asyncio.sleep(.01)
|
|
task.cancel()
|
|
with self.assertRaises(asyncio.CancelledError):
|
|
await task
|
|
self.assertEqual(self.client.pending, {})
|
|
|
|
async def test_timeout_covers_handshake_and_does_not_block_loop(self):
|
|
self.ws = FakeSocket(ready=False)
|
|
self.session.ws_connect.return_value = self.ws
|
|
task = asyncio.create_task(self.client.connect('art'))
|
|
await asyncio.sleep(.01)
|
|
self.assertFalse(task.done())
|
|
with self.assertRaises(TimeoutError):
|
|
await task
|
|
self.assertTrue(self.ws.closed)
|
|
|
|
async def test_unauthorized_handshake_closes_socket(self):
|
|
self.ws.messages = asyncio.Queue()
|
|
self.ws.messages.put_nowait(message({'event': 'ms.channel.unauthorized'}))
|
|
with self.assertRaises(TVError):
|
|
await self.client.connect('art')
|
|
self.assertTrue(self.ws.closed)
|
|
|
|
async def test_remote_reuse_and_cancellation_releases_held_power_key(self):
|
|
task = asyncio.create_task(self.client.send_key('KEY_POWER', hold=4))
|
|
await asyncio.sleep(.01)
|
|
task.cancel()
|
|
with self.assertRaises(asyncio.CancelledError):
|
|
await task
|
|
await self.client.send_key('KEY_VOLUP')
|
|
self.assertEqual([v['params']['Cmd'] for v in self.ws.sent],
|
|
['Press', 'Release', 'Click'])
|
|
self.session.ws_connect.assert_awaited_once()
|
|
|
|
async def test_close_cancels_reader(self):
|
|
await self.client.connect('art')
|
|
reader = self.client.readers['art']
|
|
await self.client.close()
|
|
self.assertTrue(reader.done())
|
|
self.assertEqual(self.client.readers, {})
|
|
self.assertEqual(self.client.sockets, {})
|
|
|
|
async def test_corrupt_frame_is_contained_and_requests_fail(self):
|
|
self.ws.respond = False
|
|
task = asyncio.create_task(self.client.art_request('get_artmode_status'))
|
|
await asyncio.sleep(.01)
|
|
self.ws.messages.put_nowait(SimpleNamespace(type=aiohttp.WSMsgType.TEXT, data='{bad'))
|
|
with self.assertRaises(TVError):
|
|
await task
|
|
self.assertTrue(self.ws.closed)
|
|
|
|
async def test_reconnect_creates_one_fresh_socket(self):
|
|
await self.client.art_request('get_artmode_status')
|
|
await self.ws.close()
|
|
await asyncio.sleep(.01)
|
|
fresh = FakeSocket()
|
|
self.session.ws_connect.return_value = fresh
|
|
await self.client.art_request('get_artmode_status')
|
|
self.assertEqual(self.session.ws_connect.await_count, 2)
|
|
self.assertTrue(self.client.alive('art'))
|
|
|
|
|
|
def fake_entity(value):
|
|
entity = Mock()
|
|
state = {'value': value}
|
|
entity.get_state = AsyncMock(side_effect=lambda: state['value'])
|
|
async def set_state(*, state):
|
|
state_holder['value'] = state
|
|
state_holder = state
|
|
entity.set_state = AsyncMock(side_effect=set_state)
|
|
return entity
|
|
|
|
|
|
class ApplicationTests(unittest.IsolatedAsyncioTestCase):
|
|
def app(self):
|
|
app = object.__new__(SamsungTheFrame)
|
|
app.AD = SimpleNamespace(app_dir='/tmp')
|
|
app.args = {'ip_address': '192.0.2.1', 'reconnect_delay': .05}
|
|
app._loop = asyncio.get_running_loop()
|
|
app._stopped = False
|
|
app._wake = asyncio.Event()
|
|
app._ready = asyncio.Event()
|
|
app._ready.set()
|
|
app._publish_lock = asyncio.Lock()
|
|
app._volume = asyncio.Queue(maxsize=8)
|
|
app._tasks = set()
|
|
app._target_task = None
|
|
app._generation = 0
|
|
app._observed = (False, True)
|
|
app._observation_revision = 0
|
|
app._refresh_at = 0
|
|
app._retry_at = 0
|
|
app._backoff = .05
|
|
app.requested_state = None
|
|
app.power_plug, app.power_plug_power = 'switch.tv', 'sensor.power'
|
|
app.get_state = AsyncMock(side_effect=lambda e: 'on' if e == app.power_plug else '60')
|
|
app.is_on_entity = fake_entity('off')
|
|
app.is_art_mode_entity = fake_entity('on')
|
|
app.tv_auto_shutdown_entity = fake_entity('off')
|
|
app.turn_on = AsyncMock()
|
|
app.turn_off = AsyncMock()
|
|
app.call_service = AsyncMock()
|
|
app.log_info = Mock()
|
|
app.log_warning = Mock()
|
|
app._client = Mock()
|
|
for method in ('art_request', 'send_key', 'app_status', 'close', 'close_channel', 'load_token'):
|
|
setattr(app._client, method, AsyncMock())
|
|
app._client.art_request.return_value = {'event': 'artmode_status', 'value': 'on'}
|
|
app._client.remote_lock = asyncio.Lock()
|
|
app.app_list = []
|
|
return app
|
|
|
|
async def test_fresh_target_shortens_idle_reconnect_delay(self):
|
|
app = self.app()
|
|
app._retry_at = app._loop.time() + 60
|
|
app._transition = AsyncMock(return_value=None)
|
|
app._set_target('hdmi')
|
|
self.assertLessEqual(app._retry_at - app._loop.time(), 2)
|
|
await app._target_task
|
|
|
|
async def test_wake_rechecks_after_ten_seconds(self):
|
|
app = self.app()
|
|
app.args['wake_up_service'] = 'esphome/wake_tv'
|
|
app.get_state = AsyncMock(side_effect=lambda e: 'on' if e == app.power_plug else '0')
|
|
self.assertEqual(await app._transition('hdmi'), 10)
|
|
app.call_service.assert_awaited_once_with('esphome/wake_tv')
|
|
app._client.send_key.assert_not_awaited()
|
|
|
|
async def asyncTearDown(self):
|
|
tasks = [t for t in asyncio.all_tasks() if t is not asyncio.current_task()]
|
|
for task in tasks:
|
|
task.cancel()
|
|
await asyncio.gather(*tasks, return_exceptions=True)
|
|
|
|
async def test_existing_yaml_contract_initializes_without_tv_io(self):
|
|
app = self.app()
|
|
app.args.update({
|
|
'input_sensors': {'power_plug': 'switch.tv', 'power_plug_power': 'sensor.power'},
|
|
'output_sensors': {'is_on': 'binary_sensor.tv_on',
|
|
'is_art_mode': 'binary_sensor.art',
|
|
'tv_auto_shutdown': 'binary_sensor.shutdown'},
|
|
'tv_state': 'sensor.target', 'wake_up_service': 'esphome/wake_tv',
|
|
'volume_up_events': ['volume_up'], 'volume_down_events': ['volume_down']})
|
|
app.AD.loop = app._loop
|
|
app.get_state = Mock(return_value='off')
|
|
app.create_entity = Mock(side_effect=lambda *a, **kw: Mock())
|
|
app.listen_state = Mock()
|
|
app._post = Mock()
|
|
app.create_task = Mock(side_effect=lambda coro: coro.close())
|
|
with patch('samsungtheframeasync.EventHandler') as handler:
|
|
app.on_initialize_smart_object()
|
|
self.assertEqual(app.listen_state.call_count, 3)
|
|
self.assertEqual(handler.call_count, 2)
|
|
self.assertEqual(app.create_entity.call_count, 3)
|
|
app.call_service.assert_not_called()
|
|
self.assertEqual(type(app).__name__, 'SamsungTheFrame')
|
|
|
|
async def test_callbacks_only_post_work_to_loop(self):
|
|
app = self.app()
|
|
app._loop = Mock()
|
|
app.update_sensors_cb()
|
|
app.request_tv_state('hdmi')
|
|
app.on_volume_event('volume', {}, 'up')
|
|
self.assertEqual(app._loop.call_soon_threadsafe.call_count, 3)
|
|
app._client.send_key.assert_not_called()
|
|
app._client.art_request.assert_not_called()
|
|
|
|
async def test_notification_updates_state_and_disconnect_preserves_it(self):
|
|
app = self.app()
|
|
app._on_art_event({'event': 'art_mode_changed', 'status': 'off'})
|
|
self.assertEqual(app._observed, (True, False))
|
|
self.assertTrue(app._wake.is_set())
|
|
app._on_art_event({'event': 'connection_lost', 'channel': 'art'})
|
|
self.assertEqual(app._observed, (True, False))
|
|
app._on_art_event({'event': 'go_to_standby'})
|
|
self.assertEqual(app._observed, (False, False))
|
|
|
|
async def test_late_query_cannot_overwrite_newer_standby_notification(self):
|
|
app = self.app()
|
|
async def request(*args):
|
|
app._on_art_event({'event': 'go_to_standby'})
|
|
return {'value': 'on'}
|
|
app._client.art_request.side_effect = request
|
|
await app._refresh()
|
|
self.assertEqual(app._observed, (False, False))
|
|
|
|
async def test_concurrent_queries_apply_latest_response(self):
|
|
app = self.app()
|
|
calls = 0
|
|
async def request(*args):
|
|
nonlocal calls
|
|
calls += 1
|
|
value = 'on' if calls == 1 else 'off'
|
|
await asyncio.sleep(.01 if value == 'on' else .02)
|
|
return {'value': value}
|
|
app._client.art_request.side_effect = request
|
|
await asyncio.gather(app._refresh(), app._refresh())
|
|
self.assertEqual(app._observed, (True, False))
|
|
|
|
async def test_unknown_consumption_is_not_off(self):
|
|
app = self.app()
|
|
for invalid in ('unavailable', None, 'nan', 'inf'):
|
|
app.get_state.side_effect = lambda e: 'on' if e == app.power_plug else invalid
|
|
self.assertEqual(await app._power(), (True, None))
|
|
|
|
async def test_hdmi_shutdown_flag_survives_initial_art_mode_read(self):
|
|
app = self.app()
|
|
await app.tv_auto_shutdown_entity.set_state(state='on')
|
|
await app._publish(False, True)
|
|
self.assertEqual(await app.tv_auto_shutdown_entity.get_state(), 'on')
|
|
app.is_on_entity = fake_entity('on')
|
|
await app._publish(False, True)
|
|
self.assertEqual(await app.tv_auto_shutdown_entity.get_state(), 'off')
|
|
|
|
async def test_power_modes_issue_expected_commands(self):
|
|
for target, mode, hold in [('art_mode', 'off', 0), ('tv_off', 'on', 4),
|
|
('tv_off', 'off', 0), ('hdmi', 'on', 0)]:
|
|
with self.subTest(target=target, mode=mode):
|
|
app = self.app()
|
|
app._client.art_request.return_value = {'value': mode}
|
|
self.assertIn(await app._transition(target), (5, 10))
|
|
if hold:
|
|
app._client.send_key.assert_awaited_once_with('KEY_POWER', hold=4)
|
|
else:
|
|
app._client.send_key.assert_awaited_once_with('KEY_POWER')
|
|
|
|
async def test_hdmi_preserves_app_visibility_check(self):
|
|
app = self.app()
|
|
app.app_list = ['netflix', 'prime']
|
|
app._client.art_request.return_value = {'value': 'off'}
|
|
app._client.app_status.side_effect = [{'visible': False}, {'visible': True}]
|
|
self.assertEqual(await app._transition('hdmi'), 5)
|
|
app._client.send_key.assert_awaited_once_with('KEY_HDMI')
|
|
|
|
async def test_off_plug_wakes_for_art_but_satisfies_power_off(self):
|
|
app = self.app()
|
|
app.get_state.return_value = 'off'
|
|
app.get_state.side_effect = None
|
|
self.assertIsNone(await app._transition('power_off'))
|
|
app.turn_on.assert_not_called()
|
|
self.assertEqual(await app._transition('art_mode'), 5)
|
|
app.turn_on.assert_awaited_once_with(app.power_plug)
|
|
app._client.art_request.assert_not_called()
|
|
|
|
async def test_manual_cancels_pending_power_cut(self):
|
|
app = self.app()
|
|
app._select_random_art = AsyncMock()
|
|
app._set_target('power_off')
|
|
await asyncio.sleep(.02)
|
|
app._select_random_art.assert_awaited_once()
|
|
app._set_target('manual')
|
|
await asyncio.sleep(.02)
|
|
app.turn_off.assert_not_called()
|
|
self.assertIsNone(app.requested_state)
|
|
self.assertFalse(app._tasks)
|
|
|
|
async def test_volume_queue_bounded_and_old_events_dropped(self):
|
|
app = self.app()
|
|
app._observed = (True, False)
|
|
for _ in range(30):
|
|
app._queue_volume('up')
|
|
self.assertEqual(app._volume.qsize(), 8)
|
|
for _ in range(8):
|
|
app._volume.get_nowait()
|
|
app._volume.put_nowait(('up', app._loop.time() - 5))
|
|
app._volume.put_nowait(('down', app._loop.time()))
|
|
task = asyncio.create_task(app._volume_worker())
|
|
await asyncio.sleep(.01)
|
|
task.cancel()
|
|
with self.assertRaises(asyncio.CancelledError):
|
|
await task
|
|
app._client.send_key.assert_awaited_once_with('KEY_VOLDOWN')
|
|
|
|
async def test_monitor_coalesces_sensor_bursts_and_shutdown_closes_client(self):
|
|
app = self.app()
|
|
client = app._client
|
|
with patch('samsungtheframeasync._TVConnection', return_value=client):
|
|
task = asyncio.create_task(app._run())
|
|
await asyncio.sleep(.02)
|
|
for _ in range(30):
|
|
app.update_sensors_cb()
|
|
await asyncio.sleep(.02)
|
|
client.art_request.assert_awaited_once()
|
|
app._on_art_event({'event': 'art_mode_changed', 'status': 'off'})
|
|
await asyncio.sleep(.02)
|
|
self.assertEqual(await app.is_on_entity.get_state(), 'on')
|
|
task.cancel()
|
|
with self.assertRaises(asyncio.CancelledError):
|
|
await task
|
|
client.close.assert_awaited_once()
|
|
self.assertFalse(app._tasks)
|
|
|
|
async def test_power_events_cannot_bypass_reconnect_backoff(self):
|
|
app = self.app()
|
|
client = app._client
|
|
app._backoff = .5
|
|
app.args['reconnect_delay'] = .5
|
|
client.alive.return_value = False
|
|
client.art_request.side_effect = TVError('offline')
|
|
with patch('samsungtheframeasync._TVConnection', return_value=client):
|
|
task = asyncio.create_task(app._run())
|
|
await asyncio.sleep(.01)
|
|
for _ in range(10):
|
|
app._wake.set()
|
|
await asyncio.sleep(.001)
|
|
self.assertEqual(client.art_request.await_count, 1)
|
|
await asyncio.sleep(.6)
|
|
self.assertGreaterEqual(client.art_request.await_count, 2)
|
|
task.cancel()
|
|
with self.assertRaises(asyncio.CancelledError):
|
|
await task
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|