New async version of samsungtheframe

tests/test_samsungtheframe.py

@@ -0,0 +1,149 @@
+import sys
+from pathlib import Path
+from types import SimpleNamespace
+import unittest
+from unittest.mock import Mock, patch
+
+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 samsungtheframe import SamsungTheFrame
+import requests
+import websocket
+
+
+class SamsungTheFrameTests(unittest.TestCase):
+    def app(self):
+        app = object.__new__(SamsungTheFrame)
+        app.args = {'ip_address': '192.0.2.1'}
+        app.AD = SimpleNamespace(app_dir='/tmp')
+        app.update_sensors_handle = None
+        app.update_sensors_is_retry = False
+        app.requested_state_confirm_cb_handle = None
+        app.volume_connection = None
+        app.power_plug = 'switch.tv'
+        app.power_plug_power = 'sensor.tv_power'
+        app.get_state = Mock(side_effect=lambda e: 'on' if e == app.power_plug else '60')
+        for name in ('is_on_entity', 'is_art_mode_entity', 'tv_auto_shutdown_entity'):
+            entity = Mock()
+            entity.get_state.return_value = 'on'
+            setattr(app, name, entity)
+        app.log_info = Mock()
+        app.log = Mock()
+        app.run_in = Mock(side_effect=lambda *a, **kw: object())
+        app.cancel_timer = Mock()
+        app.turn_off = Mock()
+        return app
+
+    def test_timeout_propagates_to_art_and_rest(self):
+        a = self.app()
+        tv = a.connect()
+        self.assertEqual(tv.timeout, 2)
+        self.assertEqual(tv.art().timeout, 2)
+        self.assertEqual(tv._get_rest_api().timeout, 2)
+
+    def test_sensor_burst_only_schedules_one_read(self):
+        a = self.app()
+        a.connect = Mock()
+        for _ in range(10):
+            a.update_sensors_cb()
+        a.run_in.assert_called_once_with(a.refresh_sensors_cb, 1)
+        a.connect.assert_not_called()
+
+    def test_failed_refresh_has_one_retry_replaced_by_new_reading(self):
+        a = self.app()
+        a.update_sensors = Mock(return_value=(False, 30))
+        a.refresh_sensors_cb({})
+        retry = a.update_sensors_handle
+        a.schedule_sensor_update(30, retry=True)
+        self.assertEqual(a.run_in.call_count, 1)
+        a.update_sensors_cb()
+        a.cancel_timer.assert_called_once_with(retry)
+        self.assertEqual(a.run_in.call_count, 2)
+        self.assertFalse(a.update_sensors_is_retry)
+
+    def test_read_errors_preserve_states_and_close_both_connections(self):
+        errors = [websocket.WebSocketTimeoutException('timeout'),
+                  ConnectionRefusedError(), requests.Timeout(), KeyError('value')]
+        for error in errors:
+            with self.subTest(error=type(error).__name__):
+                a = self.app()
+                tv, art = Mock(), Mock()
+                tv.art.return_value = art
+                art.get_artmode.side_effect = error
+                a.connect = Mock(return_value=tv)
+                self.assertEqual(a.update_sensors(), (False, 30))
+                art.close.assert_called_once()
+                tv.close.assert_called_once()
+                a.is_on_entity.set_state.assert_not_called()
+                a.is_art_mode_entity.set_state.assert_not_called()
+
+    def test_invalid_readings_do_not_mark_tv_off(self):
+        for power in ('unavailable', None, 'nan', 'inf'):
+            with self.subTest(power=power):
+                a = self.app()
+                a.get_state.side_effect = lambda e: 'on' if e == a.power_plug else power
+                a.connect = Mock()
+                self.assertEqual(a.update_sensors(), (False, 30))
+                a.connect.assert_not_called()
+                a.is_on_entity.set_state.assert_not_called()
+
+    def test_successful_read_consumes_queued_refresh(self):
+        a = self.app()
+        a.schedule_sensor_update()
+        pending = a.update_sensors_handle
+        tv = Mock()
+        tv.art.return_value.get_artmode.return_value = 'off'
+        a.connect = Mock(return_value=tv)
+        self.assertEqual(a.update_sensors(), (True, 0))
+        a.cancel_timer.assert_called_once_with(pending)
+        self.assertIsNone(a.update_sensors_handle)
+        a.is_art_mode_entity.set_state.assert_called_once_with(state='off')
+        tv.art.return_value.close.assert_called_once()
+
+    def test_art_selection_reuses_one_connection(self):
+        a = self.app()
+        tv, art = Mock(), Mock()
+        tv.art.return_value = art
+        art.available.return_value = [{'content_id': 'image'}]
+        a.connect = Mock(return_value=tv)
+        self.assertTrue(a.select_random_art())
+        tv.art.assert_called_once()
+        art.select_image.assert_called_once_with('image', show=True)
+        art.close.assert_called_once()
+
+    def test_command_timeout_closes_connection_and_schedules_confirmation(self):
+        a = self.app()
+        a.requested_state = 'tv_off'
+        a.update_sensors = Mock(return_value=(True, 0))
+        tv = Mock()
+        tv.shortcuts.return_value.power.side_effect = websocket.WebSocketTimeoutException()
+        a.connect = Mock(return_value=tv)
+        self.assertFalse(a._process_state_request())
+        tv.close.assert_called_once()
+        a.run_in.assert_called_once_with(a.confirm_change_state_cb, 30)
+
+    def test_power_off_wait_is_scheduled_and_cancelled_by_new_request(self):
+        a = self.app()
+        a.requested_state = 'power_off'
+        a.is_on_entity.get_state.return_value = 'off'
+        a.update_sensors = Mock(return_value=(True, 0))
+        a.select_random_art = Mock(return_value=True)
+        self.assertFalse(a._process_state_request())
+        a.turn_off.assert_not_called()
+        a.run_in.assert_called_once_with(a.power_off_cb, 5)
+        pending = a.requested_state_confirm_cb_handle
+        a.request_tv_state('manual')
+        a.cancel_timer.assert_called_once_with(pending)
+        a.power_off_cb({})
+        a.turn_off.assert_not_called()
+
+    def test_power_off_timer_cuts_power_and_schedules_confirmation(self):
+        a = self.app()
+        a.requested_state = 'power_off'
+        a.power_off_cb({})
+        a.turn_off.assert_called_once_with(a.power_plug)
+        a.run_in.assert_called_once_with(a.confirm_change_state_cb, 5)
+
+
+if __name__ == '__main__':
+    unittest.main()