fix: Restore Netmiko check to handle closed channel correctly

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."""

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,20 +48,78 @@ 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.")
         assert return_value == expected_value
 
     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)