fix: update asyncssh arg handling

scrapli_netconf/transport/plugins/asyncssh/transport.py

@@ -1,5 +1,6 @@
 """scrapli_netconf.transport.plugins.asyncssh.transport"""
 import asyncio
+from typing import Any, Dict
 
 from asyncssh.connection import SSHClientConnection, connect
 from asyncssh.misc import ChannelOpenError, PermissionDenied
@@ -38,8 +39,13 @@ async def open_netconf(self) -> None:
 
         # we already fetched host/port/user from the user input and/or the ssh config file, so we
         # want to use those explicitly. likewise we pass config file we already found. set known
-        # hosts and agent to None so we can not have an agent and deal w/ known hosts ourselves
-        common_args = {
+        # hosts and agent to None so we can not have an agent and deal w/ known hosts ourselves.
+        # to use ssh-agent either pass an empty tuple (to pick up ssh-agent socket from
+        # SSH_AUTH_SOCK), or pass an explicit path to ssh-agent socket should be provided as part
+        # of transport_options -- in either case these get merged into the dict *after* we set the
+        # default value of `None`, so users options override our defaults.
+
+        common_args: Dict[str, Any] = {
             "host": self._base_transport_args.host,
             "port": self._base_transport_args.port,
             "username": self.plugin_transport_args.auth_username,
@@ -48,17 +54,29 @@ async def open_netconf(self) -> None:
             "config": self.plugin_transport_args.ssh_config_file,
         }
 
+        # Allow passing `transport_options` to asyncssh
+        common_args.update(self._base_transport_args.transport_options.get("asyncssh", {}))
+
+        # Common authentication args
+        auth_args: Dict[str, Any] = {
+            "client_keys": self.plugin_transport_args.auth_private_key,
+            "password": self.plugin_transport_args.auth_password,
+            "preferred_auth": (
+                "publickey",
+                "keyboard-interactive",
+                "password",
+            ),
+        }
+
+        # The session args to use in connect() - to merge the dicts in
+        # the order to have transport options preference over predefined auth args
+        conn_args = {**auth_args, **common_args}
+
         try:
             self.session: SSHClientConnection = await asyncio.wait_for(
                 connect(
-                    client_keys=self.plugin_transport_args.auth_private_key,
-                    password=self.plugin_transport_args.auth_password,
-                    preferred_auth=(
-                        "publickey",
-                        "keyboard-interactive",
-                        "password",
-                    ),
-                    **common_args,
+                    connect(**conn_args),
+                    timeout=self._base_transport_args.timeout_socket,
                 ),
                 timeout=self._base_transport_args.timeout_socket,
             )

tests/unit/channel/test_sync_channel.py

@@ -1,7 +1,5 @@
 import pytest
 
-from scrapli_netconf.constants import NetconfClientCapabilities
-
 
 def test_open_netconf():
     pass