150 lines
6.1 KiB
Python
150 lines
6.1 KiB
Python
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()
|