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()