Added support for power plug and spam sensor
This commit is contained in:
Pierre committed
1 parent
0de8352417
commit
1fd4dab916
2 files changed
+793
-25
No files matched your search
@@ -0,0 +1,506 @@
|
||||
from pathlib import Path
|
||||
import socket
|
||||
import sys
|
||||
import threading
|
||||
import unittest
|
||||
from unittest.mock import Mock, patch
|
||||
|
||||
sys.path[:0] = [str(Path(__file__).resolve().parents[1] / 'apps'),
|
||||
str(Path(__file__).resolve().parents[1] / 'apps/ad_toolbox')]
|
||||
from vertex2 import Vertex2
|
||||
import ad_toolbox.smartcondition as SmartCondition
|
||||
|
||||
|
||||
class Harness(Vertex2):
|
||||
def __init__(self, plugged=True):
|
||||
self.plugged = plugged
|
||||
self._power_allowed = True
|
||||
self.power_plug_entity = 'switch.vertex2'
|
||||
self._interrupt_stop = threading.Event()
|
||||
self._status_requested = threading.Event()
|
||||
self._last_logged_status = None
|
||||
self._connection_lock = threading.Lock()
|
||||
self._interrupt_connection = None
|
||||
self.startup_delay = 30
|
||||
self._startup_deadline = 0
|
||||
self._startup_commands = []
|
||||
self._command_processing_lock = threading.Lock()
|
||||
self._command_processing_deadline = 123
|
||||
self._command_processing_handle = 'timer'
|
||||
self.command_processing_timeout = 30
|
||||
self.command_processing_sensor = Mock()
|
||||
self.plus5_sensors = {}
|
||||
self.audio_out_signal_sensor = Mock()
|
||||
self.unreachable_sensor = Mock()
|
||||
self._update_output_video_sensors = Mock()
|
||||
self.turn_on = Mock()
|
||||
self.turn_off = Mock()
|
||||
self.cancel_timer = Mock()
|
||||
self.log_warning = Mock()
|
||||
self.log_info = Mock()
|
||||
self.run_in = Mock()
|
||||
self.vertex2_tcp_port = 2220
|
||||
|
||||
@property
|
||||
def args(self):
|
||||
return {'vertex2_address': 'http://192.0.2.1'}
|
||||
|
||||
@property
|
||||
def name(self):
|
||||
return 'test_vertex'
|
||||
|
||||
def get_state(self, entity):
|
||||
return 'on' if self.plugged else 'off'
|
||||
|
||||
|
||||
class PowerTests(unittest.TestCase):
|
||||
def test_condition_closes_socket_before_cutting_power(self):
|
||||
app = Harness()
|
||||
connection = Mock()
|
||||
app._interrupt_connection = connection
|
||||
app.turn_off.side_effect = lambda entity: connection.close.assert_called_once()
|
||||
app.on_change_power_plug_conditions(None, SmartCondition.Result.Failed)
|
||||
connection.shutdown.assert_called_once_with(socket.SHUT_RDWR)
|
||||
app.turn_off.assert_called_once_with('switch.vertex2')
|
||||
self.assertFalse(app._vertex2_powered())
|
||||
app.cancel_timer.assert_called_once_with('timer')
|
||||
self.assertIsNone(app._command_processing_deadline)
|
||||
app.command_processing_sensor.set_state.assert_called_once_with(state='off')
|
||||
app.unreachable_sensor.set_state.assert_called_once_with(state='off')
|
||||
|
||||
def test_off_blocks_http_commands(self):
|
||||
for plugged, allowed in [(False, True), (True, False)]:
|
||||
app = Harness(plugged)
|
||||
app._power_allowed = allowed
|
||||
with patch('vertex2.get') as get:
|
||||
app.send_cmd('reboot')
|
||||
get.assert_not_called()
|
||||
|
||||
def test_no_plug_preserves_connectivity(self):
|
||||
app = Harness(False)
|
||||
app.power_plug_entity = None
|
||||
self.assertTrue(app._vertex2_powered())
|
||||
|
||||
def test_delayed_status_is_ignored_after_power_off(self):
|
||||
app = Harness(False)
|
||||
app._apply_vertex2_status({})
|
||||
app.log_info.assert_not_called()
|
||||
app.run_in.assert_not_called()
|
||||
|
||||
def test_external_plug_off_suspends_connection(self):
|
||||
app = Harness(False)
|
||||
connection = Mock()
|
||||
app._interrupt_connection = connection
|
||||
app.on_power_plug_state_change('switch.vertex2', 'state', 'on', 'off', {})
|
||||
connection.close.assert_called_once()
|
||||
app._update_output_video_sensors.assert_called_once_with('', '')
|
||||
|
||||
def test_plug_on_requests_fresh_status(self):
|
||||
app = Harness()
|
||||
app.on_power_plug_state_change('switch.vertex2', 'state', 'off', 'on', {})
|
||||
self.assertTrue(app._status_requested.is_set())
|
||||
self.assertFalse(app._vertex2_ready())
|
||||
|
||||
def test_boot_commands_wait_and_are_replayed(self):
|
||||
app = Harness()
|
||||
with patch('vertex2.time.monotonic', return_value=100):
|
||||
app._begin_startup()
|
||||
with patch('vertex2.get') as get:
|
||||
app.send_cmd('tx1plus5', 'on')
|
||||
app.send_cmd('edid', 'audiotx1 on')
|
||||
get.assert_not_called()
|
||||
self.assertFalse(app._vertex2_ready())
|
||||
with patch('vertex2.time.monotonic', return_value=130):
|
||||
self.assertTrue(app._vertex2_ready())
|
||||
with patch('vertex2.get') as get:
|
||||
app._flush_startup_commands()
|
||||
self.assertEqual(get.call_count, 2)
|
||||
self.assertEqual(app._startup_commands, [])
|
||||
|
||||
def test_http_failure_after_boot_still_warns(self):
|
||||
from requests.exceptions import ConnectionError
|
||||
app = Harness()
|
||||
app._startup_deadline = 130
|
||||
with patch('vertex2.time.monotonic', return_value=130):
|
||||
with patch('vertex2.get', side_effect=ConnectionError):
|
||||
app.send_cmd('tx1plus5', 'on')
|
||||
app.log_warning.assert_called_once()
|
||||
|
||||
def test_power_cut_discards_queued_commands(self):
|
||||
app = Harness()
|
||||
app._begin_startup()
|
||||
app.send_cmd('tx1plus5', 'on')
|
||||
app._suspend_vertex2()
|
||||
self.assertEqual(app._startup_commands, [])
|
||||
|
||||
def test_listener_does_not_connect_during_boot(self):
|
||||
app = Harness()
|
||||
app._begin_startup()
|
||||
worker = threading.Thread(target=app._interrupt_listener, args=(app._interrupt_stop,))
|
||||
with patch('vertex2.socket.create_connection') as connect:
|
||||
worker.start()
|
||||
try:
|
||||
self.assertFalse(app._interrupt_stop.wait(.1))
|
||||
connect.assert_not_called()
|
||||
finally:
|
||||
app._interrupt_stop.set()
|
||||
worker.join(2)
|
||||
self.assertFalse(worker.is_alive())
|
||||
|
||||
def test_cut_power_unblocks_recv_without_warning(self):
|
||||
app = Harness()
|
||||
receiving = threading.Event()
|
||||
closed = threading.Event()
|
||||
connection = Mock()
|
||||
connection.shutdown.side_effect = lambda how: closed.set()
|
||||
def recv(size):
|
||||
receiving.set()
|
||||
closed.wait(2)
|
||||
return b''
|
||||
connection.recv.side_effect = recv
|
||||
context = Mock(__enter__=Mock(return_value=connection), __exit__=Mock(return_value=False))
|
||||
worker = threading.Thread(target=app._interrupt_listener, args=(app._interrupt_stop,))
|
||||
with patch('vertex2.socket.create_connection', return_value=context):
|
||||
worker.start()
|
||||
try:
|
||||
self.assertTrue(receiving.wait(2))
|
||||
app.on_change_power_plug_conditions(None, SmartCondition.Result.Failed)
|
||||
self.assertTrue(closed.is_set())
|
||||
finally:
|
||||
app._interrupt_stop.set()
|
||||
closed.set()
|
||||
worker.join(2)
|
||||
self.assertFalse(worker.is_alive())
|
||||
app.log_warning.assert_not_called()
|
||||
|
||||
def test_listener_waits_off_then_reconnects_when_on(self):
|
||||
app = Harness(False)
|
||||
attempted = threading.Event()
|
||||
def connect(*args, **kwargs):
|
||||
attempted.set()
|
||||
app._interrupt_stop.set()
|
||||
raise OSError('test stop')
|
||||
worker = threading.Thread(target=app._interrupt_listener, args=(app._interrupt_stop,))
|
||||
with patch('vertex2.socket.create_connection', side_effect=connect) as create:
|
||||
worker.start()
|
||||
try:
|
||||
self.assertFalse(attempted.wait(.1))
|
||||
create.assert_not_called()
|
||||
app.plugged = True
|
||||
self.assertTrue(attempted.wait(2))
|
||||
finally:
|
||||
app._interrupt_stop.set()
|
||||
worker.join(2)
|
||||
self.assertFalse(worker.is_alive())
|
||||
app.log_warning.assert_not_called()
|
||||
|
||||
|
||||
class ReconfigurationTests(unittest.TestCase):
|
||||
def test_reconnect_reapplies_current_plus5_and_selections(self):
|
||||
app = Harness()
|
||||
app.power_plug_conditions_evaluator = Mock()
|
||||
app.power_plug_conditions_evaluator.evaluate.return_value = SmartCondition.Result.Succeeded
|
||||
app.input_selection_evaluators = [Mock()]
|
||||
app.output_selection_evaluators = [Mock()]
|
||||
app.on_update_input_selection = Mock()
|
||||
app.on_update_output_selection = Mock()
|
||||
tx0, tx1 = Mock(), Mock()
|
||||
tx0.evaluate.return_value = SmartCondition.Result.Failed
|
||||
tx1.evaluate.return_value = SmartCondition.Result.Succeeded
|
||||
app.output_config_conditions = {0: {'force_plus5': tx0}, 1: {'force_plus5': tx1}}
|
||||
app.force_tx0_plus5 = Mock()
|
||||
app.force_tx1_plus5 = Mock()
|
||||
app._reconfigure_vertex2()
|
||||
app.power_plug_conditions_evaluator.evaluate.assert_called_once_with(False)
|
||||
app.on_update_input_selection.assert_called_once_with()
|
||||
app.on_update_output_selection.assert_called_once_with()
|
||||
app.force_tx0_plus5.assert_called_once_with(False)
|
||||
app.force_tx1_plus5.assert_called_once_with(True)
|
||||
for evaluator in (tx0, tx1):
|
||||
evaluator.evaluate.assert_called_once_with(False)
|
||||
|
||||
def test_reconnect_stops_if_power_condition_no_longer_allows_device(self):
|
||||
app = Harness()
|
||||
app.power_plug_conditions_evaluator = Mock()
|
||||
app.power_plug_conditions_evaluator.evaluate.return_value = SmartCondition.Result.Failed
|
||||
app.on_update_output_selection = Mock()
|
||||
app._reconfigure_vertex2()
|
||||
app.turn_off.assert_called_once_with('switch.vertex2')
|
||||
app.on_update_output_selection.assert_not_called()
|
||||
|
||||
def test_input_selection_without_matching_condition_does_not_send_command(self):
|
||||
app = Harness()
|
||||
evaluator = Mock()
|
||||
evaluator.evaluate.return_value = SmartCondition.Result.Failed
|
||||
app.input_selection_evaluators = [(evaluator, 'AppleTV')]
|
||||
app.select_input = Mock()
|
||||
app.on_update_input_selection()
|
||||
app.select_input.assert_not_called()
|
||||
|
||||
def test_only_first_complete_status_requests_reconfiguration(self):
|
||||
app = Harness()
|
||||
fields = ('rx0', 'rx1', 'tx0', 'tx1', 'tx0sink', 'tx1sink',
|
||||
'aud0', 'aud1', 'audout', 'spd0', 'spd1')
|
||||
response = ('\n'.join(f'{field.upper()}: no signal' for field in fields)
|
||||
+ '\ninsel 0 0\n').encode('ascii')
|
||||
connection = Mock()
|
||||
responses = iter([response, response])
|
||||
def recv(size):
|
||||
try:
|
||||
return next(responses)
|
||||
except StopIteration:
|
||||
app._interrupt_stop.set()
|
||||
return b''
|
||||
connection.recv.side_effect = recv
|
||||
app.run_in.side_effect = lambda *args, **kwargs: app._status_requested.set()
|
||||
context = Mock(__enter__=Mock(return_value=connection), __exit__=Mock(return_value=False))
|
||||
with patch('vertex2.socket.create_connection', return_value=context):
|
||||
app._interrupt_listener(app._interrupt_stop)
|
||||
self.assertEqual([call.kwargs['reconfigure'] for call in app.run_in.call_args_list],
|
||||
[True, False])
|
||||
|
||||
def test_status_reconfigures_before_readback(self):
|
||||
app = Harness()
|
||||
app.input_selected_sensors = {i: Mock() for i in range(2)}
|
||||
app.generate_input_selected_sensor_name = lambda i: i
|
||||
app.get_input_config = Mock(return_value={'name': 'AppleTV'})
|
||||
app.output_names = {}
|
||||
app.output_selected_sensor = Mock()
|
||||
app._refresh_selected_input_sensor = Mock()
|
||||
sequence = Mock()
|
||||
app._flush_startup_commands = sequence.flush
|
||||
app._reconfigure_vertex2 = sequence.reconfigure
|
||||
app._refresh_plus5_sensors = sequence.readback
|
||||
app._apply_vertex2_status(dict(rx0='', rx1='', tx0='', tx1='', audout='',
|
||||
insel='0 0', reconfigure=True))
|
||||
self.assertEqual([call[0] for call in sequence.mock_calls],
|
||||
['flush', 'reconfigure', 'readback'])
|
||||
sequence.reset_mock()
|
||||
app._apply_vertex2_status(dict(rx0='', rx1='', tx0='', tx1='', audout='', insel='0 0'))
|
||||
sequence.reconfigure.assert_not_called()
|
||||
|
||||
|
||||
class Plus5Tests(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.app = Harness()
|
||||
self.app.plus5_sensors = {i: Mock() for i in range(2)}
|
||||
|
||||
def test_reads_actual_settings_independently(self):
|
||||
with patch('vertex2.get') as get:
|
||||
get.return_value.json.return_value = {'tx0plus5': '0', 'tx1plus5': '1'}
|
||||
self.app._refresh_plus5_sensors()
|
||||
self.app.plus5_sensors[0].set_state.assert_called_once_with(state='off')
|
||||
self.app.plus5_sensors[1].set_state.assert_called_once_with(state='on')
|
||||
|
||||
def test_unknown_value_does_not_confirm_off(self):
|
||||
with patch('vertex2.get') as get:
|
||||
get.return_value.json.return_value = {'tx0plus5': 'unexpected'}
|
||||
self.app._refresh_plus5_sensors()
|
||||
for sensor in self.app.plus5_sensors.values():
|
||||
sensor.set_state.assert_called_once_with(state='unavailable')
|
||||
|
||||
def test_failed_or_malformed_read_invalidates_old_settings(self):
|
||||
from requests.exceptions import HTTPError, Timeout
|
||||
for error in (HTTPError('503'), Timeout(), ValueError('invalid JSON')):
|
||||
with self.subTest(error=error):
|
||||
for sensor in self.app.plus5_sensors.values():
|
||||
sensor.reset_mock()
|
||||
with patch('vertex2.get', side_effect=error):
|
||||
self.app._refresh_plus5_sensors()
|
||||
for sensor in self.app.plus5_sensors.values():
|
||||
sensor.set_state.assert_called_once_with(state='unavailable')
|
||||
|
||||
def test_power_off_during_read_discards_response(self):
|
||||
with patch('vertex2.get') as get:
|
||||
def result():
|
||||
self.app.plugged = False
|
||||
return {'tx0plus5': '1', 'tx1plus5': '1'}
|
||||
get.return_value.json.side_effect = result
|
||||
self.app._refresh_plus5_sensors()
|
||||
for sensor in self.app.plus5_sensors.values():
|
||||
sensor.set_state.assert_called_once_with(state='unavailable')
|
||||
|
||||
def test_command_requests_readback_without_optimistic_confirmation(self):
|
||||
with patch('vertex2.get'):
|
||||
self.app.send_cmd('tx0plus5', 'on')
|
||||
self.assertTrue(self.app._status_requested.is_set())
|
||||
for sensor in self.app.plus5_sensors.values():
|
||||
sensor.set_state.assert_called_once_with(state='unavailable')
|
||||
|
||||
def test_power_cut_invalidates_settings(self):
|
||||
self.app._suspend_vertex2()
|
||||
for sensor in self.app.plus5_sensors.values():
|
||||
sensor.set_state.assert_called_once_with(state='unavailable')
|
||||
|
||||
|
||||
class StatusChangeTests(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.app = app = Harness()
|
||||
app.input_selected_sensors = {i: Mock() for i in range(2)}
|
||||
app.generate_input_selected_sensor_name = lambda i: i
|
||||
app.get_input_config = Mock(return_value={'name': 'AppleTV'})
|
||||
app.output_names = {}
|
||||
app.output_selected_sensor = Mock()
|
||||
app._refresh_selected_input_sensor = Mock()
|
||||
app._flush_startup_commands = Mock()
|
||||
app._reconfigure_vertex2 = Mock()
|
||||
app._refresh_plus5_sensors = Mock()
|
||||
self.status = dict(rx0='no signal', rx1='no signal', tx0='', tx1='',
|
||||
audout='', insel='0 4')
|
||||
|
||||
def test_identical_status_is_silent_but_still_applied(self):
|
||||
app = self.app
|
||||
app._apply_vertex2_status(self.status)
|
||||
app._apply_vertex2_status(dict(self.status, reconfigure=True))
|
||||
app.log_info.assert_called_once()
|
||||
app.run_in.assert_not_called()
|
||||
self.assertEqual(app._update_output_video_sensors.call_count, 2)
|
||||
app._reconfigure_vertex2.assert_called_once()
|
||||
|
||||
def test_changed_status_extends_active_window_only_once(self):
|
||||
app = self.app
|
||||
app._apply_vertex2_status(self.status)
|
||||
changed = dict(self.status, tx0='4K50')
|
||||
with patch('vertex2.time.monotonic', return_value=200):
|
||||
app._apply_vertex2_status(changed)
|
||||
self.assertEqual(app._command_processing_deadline, 230)
|
||||
app._apply_vertex2_status(changed)
|
||||
self.assertEqual(app.log_info.call_count, 2)
|
||||
app.run_in.assert_called_once()
|
||||
app._on_command_processing_timeout({'deadline': 123})
|
||||
self.assertEqual(app._command_processing_deadline, 230)
|
||||
app._on_command_processing_timeout({'deadline': 230})
|
||||
app.command_processing_sensor.set_state.assert_called_with(state='off')
|
||||
|
||||
def test_change_without_active_command_logs_without_starting_window(self):
|
||||
app = self.app
|
||||
app._command_processing_deadline = None
|
||||
app._apply_vertex2_status(self.status)
|
||||
app._apply_vertex2_status(dict(self.status, audout='L-PCM'))
|
||||
self.assertEqual(app.log_info.call_count, 2)
|
||||
app.run_in.assert_not_called()
|
||||
app.command_processing_sensor.set_state.assert_not_called()
|
||||
|
||||
|
||||
class CommandProcessingTests(unittest.TestCase):
|
||||
def test_interrupt_spam_does_not_extend_command_deadline(self):
|
||||
app = Harness()
|
||||
app.command_processing_sensor.get_state.return_value = 'off'
|
||||
with patch('vertex2.time.monotonic', return_value=100), patch('vertex2.get'):
|
||||
app.send_cmd('mutetx0audio', 'on')
|
||||
self.assertEqual(app._command_processing_deadline, 130)
|
||||
app.run_in.reset_mock()
|
||||
app.cancel_timer.reset_mock()
|
||||
connection = Mock()
|
||||
chunks = iter([b'int\n'] * 20)
|
||||
def recv(size):
|
||||
try:
|
||||
return next(chunks)
|
||||
except StopIteration:
|
||||
app._interrupt_stop.set()
|
||||
return b''
|
||||
connection.recv.side_effect = recv
|
||||
context = Mock(__enter__=Mock(return_value=connection), __exit__=Mock(return_value=False))
|
||||
with patch('vertex2.socket.create_connection', return_value=context):
|
||||
app._interrupt_listener(app._interrupt_stop)
|
||||
self.assertEqual(app._command_processing_deadline, 130)
|
||||
app.cancel_timer.assert_not_called()
|
||||
app.run_in.assert_not_called()
|
||||
app._on_command_processing_timeout({'deadline': 130})
|
||||
app.command_processing_sensor.set_state.assert_called_with(state='off')
|
||||
|
||||
def test_new_command_extends_window_and_old_callback_is_ignored(self):
|
||||
app = Harness()
|
||||
with patch('vertex2.time.monotonic', return_value=100), patch('vertex2.get'):
|
||||
app.send_cmd('mutetx0audio', 'on')
|
||||
with patch('vertex2.time.monotonic', return_value=110), patch('vertex2.get'):
|
||||
app.send_cmd('mutetx0audio', 'off')
|
||||
self.assertEqual(app._command_processing_deadline, 140)
|
||||
app.command_processing_sensor.set_state.reset_mock()
|
||||
app._on_command_processing_timeout({'deadline': 130})
|
||||
app.command_processing_sensor.set_state.assert_not_called()
|
||||
app._on_command_processing_timeout({'deadline': 140})
|
||||
app.command_processing_sensor.set_state.assert_called_once_with(state='off')
|
||||
|
||||
def test_queued_command_does_not_start_window_before_boot(self):
|
||||
app = Harness()
|
||||
app._begin_startup()
|
||||
with patch('vertex2.get') as get:
|
||||
app.send_cmd('mutetx0audio', 'on')
|
||||
get.assert_not_called()
|
||||
app.command_processing_sensor.set_state.assert_not_called()
|
||||
app.run_in.assert_not_called()
|
||||
|
||||
|
||||
class SpamSensorTests(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.app = app = Harness()
|
||||
app.no_signal_spam_sensor = Mock()
|
||||
self.state = 'off'
|
||||
app.no_signal_spam_sensor.get_state.side_effect = lambda: self.state
|
||||
app.no_signal_spam_sensor.set_state.side_effect = lambda **kw: setattr(self, 'state', kw['state'])
|
||||
app._spam_lock = threading.RLock()
|
||||
app._spam_last_interrupt = None
|
||||
app._spam_snapshot = None
|
||||
app._spam_count = 0
|
||||
app._spam_timer = None
|
||||
app._spam_deadline = None
|
||||
self.status = dict(rx0='no signal', rx1='no signal', tx0='', tx1='', audout='', insel='0 4')
|
||||
|
||||
def observe(self, at, interrupt=True, **changes):
|
||||
status = dict(self.status, **changes)
|
||||
if interrupt:
|
||||
status['interrupt_at'] = at
|
||||
with patch('vertex2.time.monotonic', return_value=at):
|
||||
self.app._observe_no_signal_spam(status)
|
||||
|
||||
def test_three_identical_interrupts_detect_and_silence_clears(self):
|
||||
self.observe(100)
|
||||
self.observe(103)
|
||||
self.assertEqual(self.state, 'off')
|
||||
self.observe(106)
|
||||
self.assertEqual(self.state, 'on')
|
||||
self.app._expire_no_signal_spam({'deadline': 107})
|
||||
self.assertEqual(self.state, 'on')
|
||||
self.app._expire_no_signal_spam({'deadline': 110})
|
||||
self.assertEqual(self.state, 'off')
|
||||
|
||||
def test_polls_do_not_count_and_changed_poll_breaks_streak(self):
|
||||
for at in range(100, 110):
|
||||
self.observe(at, interrupt=False)
|
||||
self.assertEqual(self.state, 'off')
|
||||
self.observe(112)
|
||||
self.observe(115)
|
||||
self.observe(116, interrupt=False, insel='1 4')
|
||||
self.observe(118)
|
||||
self.assertEqual(self.state, 'off')
|
||||
|
||||
def test_normal_transitions_and_nonperiodic_events_do_not_trigger(self):
|
||||
self.observe(100)
|
||||
self.observe(103, audout='L-PCM')
|
||||
self.observe(106)
|
||||
self.assertEqual(self.state, 'off')
|
||||
self.observe(107)
|
||||
self.observe(113)
|
||||
self.assertEqual(self.state, 'off')
|
||||
|
||||
def test_signal_return_or_power_loss_clears(self):
|
||||
for at in (100, 103, 106):
|
||||
self.observe(at)
|
||||
self.observe(107, tx0='4K50')
|
||||
self.assertEqual(self.state, 'off')
|
||||
for at in (110, 113, 116):
|
||||
self.observe(at)
|
||||
self.app._suspend_vertex2()
|
||||
self.assertEqual(self.state, 'off')
|
||||
self.app._expire_no_signal_spam({'deadline': 120})
|
||||
self.assertEqual(self.state, 'off')
|
||||
|
||||
def test_unknown_input_is_not_no_signal(self):
|
||||
for at in (100, 103, 106):
|
||||
self.observe(at, rx0='unavailable')
|
||||
self.assertEqual(self.state, 'off')
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
unittest.main()
|
||||
Reference in new issue
Block a user