added +5v sensor to vertex2

tests/test_vertex2.py

@@ -19,6 +19,8 @@ class Harness(Vertex2):
         self._interrupt_stop = threading.Event()
         self._status_requested = threading.Event()
         self._last_logged_status = None
+        self._input_5v_states = {}
+        self._schedule_input_5v_refresh = Mock()
         self._connection_lock = threading.Lock()
         self._interrupt_connection = None
         self.startup_delay = 30
@@ -348,6 +350,21 @@ class StatusChangeTests(unittest.TestCase):
         self.status = dict(rx0='no signal', rx1='no signal', tx0='', tx1='',
                            audout='', insel='0 4')
 
+    def test_input_5v_refreshes_on_interrupt_even_when_video_status_is_unchanged(self):
+        app = self.app
+        app.input_5v_sensors = {}
+        app._schedule_input_5v_refresh = lambda: Vertex2._schedule_input_5v_refresh(app)
+        app._refresh_input_5v_sensors = Mock()
+        for metadata in ({'reconfigure': True}, {'interrupt_at': 100}):
+            app.run_in.reset_mock()
+            app._refresh_input_5v_sensors.reset_mock()
+            app._apply_vertex2_status(dict(self.status, **metadata))
+            app._refresh_input_5v_sensors.assert_not_called()
+            app.run_in.assert_called_once_with(app._refresh_input_5v_sensors, 0.1)
+            callback = app.run_in.call_args.args[0]
+            callback({})
+            app._refresh_input_5v_sensors.assert_called_once_with({})
+
     def test_identical_status_is_silent_but_still_applied(self):
         app = self.app
         app._apply_vertex2_status(self.status)
@@ -454,6 +471,32 @@ class SpamSensorTests(unittest.TestCase):
         with patch('vertex2.time.monotonic', return_value=at):
             self.app._observe_no_signal_spam(status)
 
+    def test_stop_cancels_timers_without_publishing_spam_state(self):
+        app = self.app
+        self.observe(100)
+        self.observe(103)
+        self.observe(106)
+        self.assertEqual(self.state, 'on')
+        app._spam_timer = 'spam_timer'
+        app.no_signal_spam_sensor.reset_mock()
+        app.cancel_timer.reset_mock()
+
+        app._stop_interrupt_listener()
+
+        self.assertTrue(app._interrupt_stop.is_set())
+        self.assertIsNone(app._spam_timer)
+        self.assertIsNone(app._spam_deadline)
+        app.cancel_timer.assert_any_call('spam_timer')
+        app.cancel_timer.assert_any_call('timer')
+        app.no_signal_spam_sensor.set_state.assert_not_called()
+        # An already queued expiry callback must also leave the state alone.
+        app._expire_no_signal_spam({'deadline': 110})
+        app.no_signal_spam_sensor.set_state.assert_not_called()
+
+    def test_connection_loss_still_marks_spam_sensor_unavailable(self):
+        self.app._reset_no_signal_spam('unavailable')
+        self.assertEqual(self.state, 'unavailable')
+
     def test_three_identical_interrupts_detect_and_silence_clears(self):
         self.observe(100)
         self.observe(103)
@@ -622,3 +665,136 @@ class InputSelectTests(unittest.TestCase):
         config = json.loads(discovery.kwargs['payload'])
         self.assertEqual(config['availability'][1], {
             'topic': 'transport/connection', 'payload_available': 'ready', 'payload_not_available': 'gone'})
+
+
+class Input5VTests(unittest.TestCase):
+    def setUp(self):
+        self.app = app = SelectHarness()
+        app.input_names = {'AppleTV': 0, 'Xbox': 1}
+        app.create_entity = Mock(side_effect=lambda *a, **kw: Mock())
+        app.run_every = Mock()
+        app._initialize_input_5v_sensors()
+
+    def test_names_and_icons_come_from_input_configuration(self):
+        calls = self.app.create_entity.call_args_list
+        self.assertEqual(calls[0].args[0], 'binary_sensor.input_appletv_5v')
+        self.assertEqual(calls[0].kwargs['friendly_name'], 'AppleTV +5 V')
+        self.assertEqual(calls[0].kwargs['icon'], 'mdi:apple')
+        self.assertEqual(calls[1].kwargs['icon'], 'mdi:microsoft-xbox')
+        self.app.run_every.assert_not_called()
+
+    def test_reads_each_input_even_without_video_signal(self):
+        with patch('vertex2.get') as get:
+            get.return_value.json.return_value = {'RX0': 'no signal', 'rx0in5v': '2', 'rx1in5v': '1'}
+            self.app._refresh_input_5v_sensors()
+        self.app.input_5v_sensors[0].set_state.assert_called_once_with(state='on')
+        self.app.input_5v_sensors[1].set_state.assert_called_once_with(state='off')
+
+    def test_invalid_or_missing_values_are_unavailable(self):
+        with patch('vertex2.get') as get:
+            get.return_value.json.return_value = {'rx0in5v': 'unexpected'}
+            self.app._refresh_input_5v_sensors()
+        for sensor in self.app.input_5v_sensors.values():
+            sensor.set_state.assert_called_once_with(state='unavailable')
+
+    def test_http_failure_invalidates_old_values(self):
+        from requests.exceptions import Timeout
+        with patch('vertex2.get', side_effect=Timeout):
+            self.app._refresh_input_5v_sensors()
+        for sensor in self.app.input_5v_sensors.values():
+            sensor.set_state.assert_called_once_with(state='unavailable')
+
+    def test_no_read_while_off_or_booting(self):
+        for plugged, deadline in ((False, 0), (True, float('inf'))):
+            self.app.plugged = plugged
+            self.app._startup_deadline = deadline
+            with patch('vertex2.get') as get:
+                self.app._refresh_input_5v_sensors()
+                get.assert_not_called()
+
+    def test_power_cut_during_request_discards_result(self):
+        def result():
+            self.app.plugged = False
+            return {'rx0in5v': '2', 'rx1in5v': '2'}
+        with patch('vertex2.get') as get:
+            get.return_value.json.side_effect = result
+            self.app._refresh_input_5v_sensors()
+        for sensor in self.app.input_5v_sensors.values():
+            sensor.set_state.assert_called_once_with(state='unavailable')
+
+    def test_5v_edges_start_or_extend_command_processing_once_per_read(self):
+        app = self.app
+        for previous, raw in (('off', '2'), ('on', '1')):
+            app._input_5v_states = {index: previous for index in range(4)}
+            for sensor in app.input_5v_sensors.values():
+                sensor.get_state.return_value = previous
+            app.command_processing_sensor.get_state.return_value = 'off'
+            app.command_processing_sensor.set_state.reset_mock()
+            app.run_in.reset_mock()
+            with patch('vertex2.time.monotonic', return_value=200), patch('vertex2.get') as get:
+                get.return_value.json.return_value = {'rx0in5v': raw, 'rx1in5v': raw}
+                app._refresh_input_5v_sensors()
+            app.command_processing_sensor.set_state.assert_called_once_with(state='on')
+            self.assertEqual(app._command_processing_deadline, 230)
+            app.run_in.assert_called_once()
+            app._on_command_processing_timeout({'deadline': 230})
+            app.command_processing_sensor.set_state.assert_called_with(state='off')
+
+    def test_initial_unavailable_and_unchanged_5v_do_not_start_processing(self):
+        app = self.app
+        app._start_command_processing = Mock()
+        for previous, raw in (('unavailable', '2'), ('unknown', '1'), ('on', '2'),
+                              ('off', '1'), ('on', None)):
+            app._input_5v_states = {index: previous for index in range(4)}
+            for sensor in app.input_5v_sensors.values():
+                sensor.get_state.return_value = previous
+            with patch('vertex2.get') as get:
+                get.return_value.json.return_value = {'rx0in5v': raw, 'rx1in5v': raw}
+                app._refresh_input_5v_sensors()
+        app._start_command_processing.assert_not_called()
+
+    def test_processing_is_identical_with_all_some_or_no_input_sensors(self):
+        app = self.app
+        sensors = app.input_5v_sensors.copy()
+        for exposed in (sensors, {0: sensors[0]}, {}):
+            with self.subTest(exposed=list(exposed)):
+                app.input_5v_sensors = exposed
+                app._input_5v_states = {}
+                app._start_command_processing = Mock()
+                with patch('vertex2.get') as get:
+                    # All physical ports are tracked even without input_config entries.
+                    for values in (('1', '1', '1', '1'), ('1', '1', '1', '2'),
+                                   ('1', '1', '1', '2'), ('1', '1', '1', '1')):
+                        get.return_value.json.return_value = {
+                            f'rx{index}in5v': value for index, value in enumerate(values)}
+                        app._refresh_input_5v_sensors()
+                self.assertEqual(app._start_command_processing.call_count, 2)
+                self.assertEqual(app._input_5v_states[3], 'off')
+
+    def test_ha_sensor_state_cannot_create_or_hide_a_transition(self):
+        app = self.app
+        app._start_command_processing = Mock()
+        app._input_5v_states = {0: 'off', 1: 'off'}
+        for sensor in app.input_5v_sensors.values():
+            sensor.get_state.return_value = 'on'
+        with patch('vertex2.get') as get:
+            get.return_value.json.return_value = {'rx0in5v': '1', 'rx1in5v': '1'}
+            app._refresh_input_5v_sensors()
+            app._start_command_processing.assert_not_called()
+            get.return_value.json.return_value = {'rx0in5v': '2', 'rx1in5v': '2'}
+            app._refresh_input_5v_sensors()
+            app._start_command_processing.assert_called_once()
+
+    def test_read_failure_resets_internal_baseline_without_processing(self):
+        from requests.exceptions import Timeout
+        app = self.app
+        app.input_5v_sensors = {}
+        app._input_5v_states = {0: 'on'}
+        app._start_command_processing = Mock()
+        with patch('vertex2.get', side_effect=Timeout):
+            app._refresh_input_5v_sensors()
+        self.assertEqual(app._input_5v_states, {})
+        with patch('vertex2.get') as get:
+            get.return_value.json.return_value = {'rx0in5v': '1'}
+            app._refresh_input_5v_sensors()
+        app._start_command_processing.assert_not_called()