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._input_5v_states = {}
        self._schedule_input_5v_refresh = Mock()
        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', silent=True)
        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_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)
        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_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', silent=True)
        app.cancel_timer.assert_any_call('timer', silent=True)
        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)
        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()


class SelectHarness(Harness):
    @property
    def args(self):
        return dict(super().args,
                    input_selected_sensor='sensor.input', unreachable_sensor='binary_sensor.unreachable',
                    input_select={'entity': 'select.vertex2_input', 'unique_id': 'vertex2_input',
                                  'connection_topic': 'appdaemon/connection'},
                    input_config={0: {'name': 'AppleTV', 'icon': 'mdi:apple'},
                                  1: {'name': 'Xbox', 'icon': 'mdi:microsoft-xbox'},
                                  'default': {'name': 'None', 'icon': 'mdi:television-off'}})


class InputSelectTests(unittest.TestCase):
    def setUp(self):
        self.app = app = SelectHarness()
        app.input_names = {'AppleTV': 0, 'Xbox': 1}
        app.input_selected_sensor = Mock()
        app.input_selected_sensor.get_state.return_value = 'AppleTV'
        app.unreachable_sensor.get_state.return_value = 'off'
        app.call_service = Mock()
        app.get_plugin_config = Mock(return_value={'will_topic': 'transport/connection'})
        app.listen_state = Mock()
        app.listen_event = Mock()
        app.send_cmd = Mock()
        app._initialize_input_select()

    def published(self, suffix):
        return [c.kwargs['payload'] for c in self.app.call_service.call_args_list
                if c.args[0] == 'mqtt/publish'
                and c.kwargs['topic'] == self.app._select_topic + suffix][-1]

    def test_discovery_options_and_initial_icon(self):
        import json
        discovery = next(c for c in self.app.call_service.call_args_list
                         if c.kwargs.get('topic') == 'homeassistant/select/vertex2_input/config')
        config = json.loads(discovery.kwargs['payload'])
        self.assertEqual(config['options'], ['AppleTV', 'Xbox'])
        self.assertFalse(config['optimistic'])
        self.assertEqual(config['availability'][1]['topic'], 'appdaemon/connection')
        self.assertEqual(json.loads(self.published('/attributes'))['icon'], 'mdi:apple')

    def test_user_command_sets_both_tx_without_changing_confirmed_state(self):
        app = self.app
        app.call_service.reset_mock()
        app._on_input_select_mqtt('MQTT_MESSAGE',
                                 {'topic': app._select_topic + '/set', 'payload': 'Xbox'}, {})
        app.send_cmd.assert_called_once_with('insel', '1 1')
        self.assertTrue(app._status_requested.is_set())
        app.input_selected_sensor.set_state.assert_not_called()
        app.call_service.assert_not_called()

    def test_unknown_or_powered_off_command_is_rejected(self):
        app = self.app
        for plugged, value in ((True, 'invalid'), (False, 'Xbox')):
            app.plugged = plugged
            app._on_input_select_mqtt('MQTT_MESSAGE',
                                     {'topic': app._select_topic + '/set', 'payload': value}, {})
        app.send_cmd.assert_not_called()

    def test_state_and_icon_follow_input_on_active_output(self):
        import json
        app = self.app
        state = {'name': 'AppleTV'}
        app.input_selected_sensor.get_state.side_effect = lambda: state['name']
        app.input_selected_sensor.set_state.side_effect = lambda **kw: state.update(name=kw.get('state', state['name']))
        app.input_selected_sensors = {
            app.generate_input_selected_sensor_name(0): Mock(),
            app.generate_input_selected_sensor_name(1): Mock(),
        }
        for index, name in enumerate(('AppleTV', 'Xbox')):
            app.input_selected_sensors[app.generate_input_selected_sensor_name(index)].get_state.return_value = name
        for index, expected, icon in ((1, 'Xbox', 'mdi:microsoft-xbox'), (0, 'AppleTV', 'mdi:apple')):
            app._refresh_selected_input_sensor(index)
            app._on_input_select_state(None, None, None, None, {})
            self.assertEqual(self.published('/state'), expected)
            self.assertEqual(json.loads(self.published('/attributes'))['icon'], icon)
            self.assertEqual(app._select_discovery_payload['icon'], icon)
        app._refresh_selected_input_sensor(None)
        app._sync_input_select()
        self.assertEqual(self.published('/availability'), 'offline')
        self.assertEqual(self.published('/state'), 'None')

    def test_unreachable_and_power_off_mark_select_unavailable(self):
        app = self.app
        app.unreachable_sensor.get_state.return_value = 'on'
        app._sync_input_select()
        self.assertEqual(self.published('/availability'), 'offline')
        app.unreachable_sensor.get_state.return_value = 'off'
        app.plugged = False
        app._sync_input_select()
        self.assertEqual(self.published('/availability'), 'offline')

    def test_reconnect_resubscribes_and_republishes_discovery(self):
        app = self.app
        for data in ({'state': 'Connected'}, {'topic': 'homeassistant/status', 'payload': 'online'}):
            app.call_service.reset_mock()
            app._on_input_select_mqtt('MQTT_MESSAGE', data, {})
            subscriptions = [c.kwargs['topic'] for c in app.call_service.call_args_list
                             if c.args[0] == 'mqtt/subscribe']
            self.assertIn(app._select_topic + '/set', subscriptions)
            self.assertTrue(any(c.kwargs.get('topic', '').endswith('/config')
                                for c in app.call_service.call_args_list))

    def test_connection_topic_is_derived_from_shared_plugin(self):
        import json
        app = self.app
        app._input_select_config.pop('connection_topic')
        app.get_plugin_config.return_value = {
            'will_topic': 'transport/connection', 'birth_payload': 'ready', 'will_payload': 'gone'}
        app.call_service.reset_mock()
        app._discover_input_select()
        discovery = next(c for c in app.call_service.call_args_list
                         if c.kwargs.get('topic') == 'homeassistant/select/vertex2_input/config')
        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()