diff --git a/mfd_switchmanagement/connections/ssh.py b/mfd_switchmanagement/connections/ssh.py index 2b3afbf..29ef557 100644 --- a/mfd_switchmanagement/connections/ssh.py +++ b/mfd_switchmanagement/connections/ssh.py @@ -33,6 +33,7 @@ def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) self._use_ssh_key: bool = kwargs.get("use_ssh_key", False) self._ssh_key_file: Union[str, "Path"] = kwargs.get("ssh_key_file", "") + self._resolved_device_type: Optional[str] = self._device_type self._connection = self.connect() def connect(self) -> Netmiko: @@ -66,6 +67,8 @@ def connect(self) -> Netmiko: raise SwitchException("Detected not supported OS Switch, contact with developers of module") else: logger.log(level=log_levels.MODULE_DEBUG, msg=f'Detected "{best_match}" switch type.') + # Store the final resolved device type (either explicit or auto-detected) + self._resolved_device_type = str(switch["device_type"]) # to properly identify prompt on slower switches we're temporally increasing delay connection = Netmiko(**switch, global_delay_factor=delay) # if user did not provide delay, let's restore Netmiko default after setup @@ -91,14 +94,26 @@ def _reconnect(self) -> None: def _check_connection(self) -> Optional[bool]: """Check connection to switch.""" - connection_status = ( - self._connection.remote_conn.transport.is_alive() - ) # check if transport layer of channel of connection is active + connection_status = bool(self._connection.is_alive()) + + # Fallback to Paramiko transport for Arista when Netmiko reports False + if not connection_status and self._is_arista_device_type(): + remote_conn = getattr(self._connection, "remote_conn", None) + if remote_conn: + transport = getattr(remote_conn, "transport", None) + if transport and callable(getattr(transport, "is_alive", None)): + connection_status = bool(transport.is_alive()) + if connection_status: logger.log(level=log_levels.MODULE_DEBUG, msg="Connection established.") - return connection_status + return True logger.log(level=log_levels.MODULE_DEBUG, msg="Connection not established.") + def _is_arista_device_type(self) -> bool: + """Determine whether connected switch uses Arista Netmiko driver.""" + device_type = (self._resolved_device_type or self._device_type or "").lower() + return "arista" in device_type + @property def _remote(self) -> Netmiko: """If connection is dropped reconnect.""" diff --git a/tests/unit/test_mfd_switchmanagement/test_connections/test_ssh.py b/tests/unit/test_mfd_switchmanagement/test_connections/test_ssh.py index 401c69f..f1cf8d6 100644 --- a/tests/unit/test_mfd_switchmanagement/test_connections/test_ssh.py +++ b/tests/unit/test_mfd_switchmanagement/test_connections/test_ssh.py @@ -21,7 +21,6 @@ def ssh_connection(self, mocker): } ssh_connection = SSHSwitchConnection(**params) ssh_connection._connection = mocker.create_autospec(Netmiko) - ssh_connection._connection.remote_conn = mocker.Mock() mocker.stopall() return ssh_connection @@ -49,7 +48,7 @@ def test__reconnect_connection_failed(self, ssh_connection, mocker): def test__check_connection_connection_established(self, ssh_connection, mocker): log_debug = mocker.patch("mfd_switchmanagement.connections.ssh.logger.log") - ssh_connection._connection.remote_conn.transport.is_alive = mocker.Mock(return_value=True) + ssh_connection._connection.is_alive = mocker.Mock(return_value=True) return_value = ssh_connection._check_connection() expected_value = True log_debug.assert_called_once_with(level=log_levels.MODULE_DEBUG, msg="Connection established.") @@ -57,12 +56,70 @@ def test__check_connection_connection_established(self, ssh_connection, mocker): def test__check_connection_connection_not_established(self, ssh_connection, mocker): log_debug = mocker.patch("mfd_switchmanagement.connections.ssh.logger.log") - ssh_connection._connection.remote_conn.transport.is_alive = mocker.Mock(return_value=False) + ssh_connection._connection.is_alive = mocker.Mock(return_value=False) return_value = ssh_connection._check_connection() expected_value = None log_debug.assert_called_once_with(level=log_levels.MODULE_DEBUG, msg="Connection not established.") assert return_value == expected_value + def test__check_connection_arista_transport_fallback(self, ssh_connection, mocker): + log_debug = mocker.patch("mfd_switchmanagement.connections.ssh.logger.log") + ssh_connection._resolved_device_type = "arista_eos" + ssh_connection._connection.is_alive = mocker.Mock(return_value=False) + ssh_connection._connection.remote_conn = mocker.Mock() + ssh_connection._connection.remote_conn.transport = mocker.Mock() + ssh_connection._connection.remote_conn.transport.is_alive = mocker.Mock(return_value=True) + + return_value = ssh_connection._check_connection() + + log_debug.assert_called_once_with(level=log_levels.MODULE_DEBUG, msg="Connection established.") + assert return_value is True + + def test__check_connection_non_arista_does_not_use_transport_fallback(self, ssh_connection, mocker): + log_debug = mocker.patch("mfd_switchmanagement.connections.ssh.logger.log") + ssh_connection._resolved_device_type = "extreme_exos" + ssh_connection._connection.is_alive = mocker.Mock(return_value=False) + ssh_connection._connection.remote_conn = mocker.Mock() + ssh_connection._connection.remote_conn.transport = mocker.Mock() + ssh_connection._connection.remote_conn.transport.is_alive = mocker.Mock(return_value=True) + + return_value = ssh_connection._check_connection() + + log_debug.assert_called_once_with(level=log_levels.MODULE_DEBUG, msg="Connection not established.") + assert return_value is None + + def test__check_connection_arista_autodetect_fallback(self, mocker): + """Test that Arista fallback activates after autodetect.""" + mocker.patch("mfd_switchmanagement.connections.ssh.SSHSwitchConnection.connect") + + params = { + "ip": "10.10.10.10", + "username": "root", + "password": "***", + "secret": "***", + "auth_timeout": 60, + # device_type NOT provided - will trigger autodetect + } + ssh_connection = SSHSwitchConnection(**params) + ssh_connection._connection = mocker.create_autospec(Netmiko) + + # Simulate that autodetect returned "arista_eos" + ssh_connection._resolved_device_type = "arista_eos" + + log_debug = mocker.patch("mfd_switchmanagement.connections.ssh.logger.log") + # Netmiko.is_alive() returns False (Arista quirk) + ssh_connection._connection.is_alive = mocker.Mock(return_value=False) + # But Paramiko transport is alive + ssh_connection._connection.remote_conn = mocker.Mock() + ssh_connection._connection.remote_conn.transport = mocker.Mock() + ssh_connection._connection.remote_conn.transport.is_alive = mocker.Mock(return_value=True) + + return_value = ssh_connection._check_connection() + + # Should detect "Connection established" via transport fallback + log_debug.assert_called_once_with(level=log_levels.MODULE_DEBUG, msg="Connection established.") + assert return_value is True + def test__remote_reconnect(self, ssh_connection, mocker): mocker.patch("mfd_switchmanagement.connections.ssh.logging") ssh_connection._check_connection = mocker.Mock(return_value=False)