Module scrapli.driver.core.cisco_nxos.async_driver¶
scrapli.driver.core.cisco_nxos.async_driver
Expand source code
"""scrapli.driver.core.cisco_nxos.async_driver"""
from copy import deepcopy
from io import BytesIO
from typing import Any, Callable, Dict, List, Optional, Union
from scrapli.driver import AsyncNetworkDriver
from scrapli.driver.core.cisco_nxos.base_driver import FAILED_WHEN_CONTAINS, PRIVS, NXOSDriverBase
from scrapli.driver.network.base_driver import PrivilegeLevel
async def nxos_on_open(conn: AsyncNetworkDriver) -> None:
"""
AsyncNXOSDriver default on_open callable
Args:
conn: NetworkDriver object
Returns:
None
Raises:
N/A
"""
await conn.acquire_priv(desired_priv=conn.default_desired_privilege_level)
await conn.send_command(command="terminal length 0")
await conn.send_command(command="terminal width 511")
async def nxos_on_close(conn: AsyncNetworkDriver) -> None:
"""
AsyncNXOSDriver default on_close callable
Args:
conn: NetworkDriver object
Returns:
None
Raises:
N/A
"""
await conn.acquire_priv(desired_priv=conn.default_desired_privilege_level)
conn.channel.write(channel_input="exit")
conn.channel.send_return()
class AsyncNXOSDriver(AsyncNetworkDriver, NXOSDriverBase):
def __init__(
self,
host: str,
privilege_levels: Optional[Dict[str, PrivilegeLevel]] = None,
default_desired_privilege_level: str = "privilege_exec",
port: Optional[int] = None,
auth_username: str = "",
auth_password: str = "",
auth_private_key: str = "",
auth_private_key_passphrase: str = "",
auth_strict_key: bool = True,
auth_bypass: bool = False,
timeout_socket: float = 15.0,
timeout_transport: float = 30.0,
timeout_ops: float = 30.0,
comms_return_char: str = "\n",
comms_ansi: Optional[bool] = None,
ssh_config_file: Union[str, bool] = False,
ssh_known_hosts_file: Union[str, bool] = False,
on_init: Optional[Callable[..., Any]] = None,
on_open: Optional[Callable[..., Any]] = None,
on_close: Optional[Callable[..., Any]] = None,
transport: str = "system",
transport_options: Optional[Dict[str, Any]] = None,
channel_log: Union[str, bool, BytesIO] = False,
channel_lock: bool = False,
logging_uid: str = "",
auth_secondary: str = "",
failed_when_contains: Optional[List[str]] = None,
textfsm_platform: str = "cisco_nxos",
genie_platform: str = "nxos",
):
"""
NXOSDriver Object
Please see `scrapli.driver.base.base_driver.Driver` for all "base driver" arguments!
# noqa: DAR101
Args:
privilege_levels: optional user provided privilege levels, if left None will default to
scrapli standard privilege levels
default_desired_privilege_level: string of name of default desired priv, this is the
priv level that is generally used to disable paging/set terminal width and things
like that upon first login, and is also the priv level scrapli will try to acquire
for normal "command" operations (`send_command`, `send_commands`)
auth_secondary: password to use for secondary authentication (enable)
on_open: callable that accepts the class instance as its only argument. this callable,
if provided, is executed immediately after authentication is completed. Common use
cases for this callable would be to disable paging or accept any kind of banner
message that prompts a user upon connection
on_close: callable that accepts the class instance as its only argument. this callable,
if provided, is executed immediately prior to closing the underlying transport.
Common use cases for this callable would be to save configurations prior to exiting,
or to logout properly to free up vtys or similar.
textfsm_platform: string name of textfsm parser platform
genie_platform: string name of cisco genie parser platform
failed_when_contains: List of strings that indicate a command/config has failed
Returns:
None
Raises:
N/A
"""
# somewhere/somehow the mixin is causing mypy to be upset about comms_prompt_pattern...
self.comms_prompt_pattern: str
if privilege_levels is None:
privilege_levels = deepcopy(PRIVS)
if on_open is None:
on_open = nxos_on_open
if on_close is None:
on_close = nxos_on_close
if failed_when_contains is None:
failed_when_contains = FAILED_WHEN_CONTAINS.copy()
super().__init__(
host=host,
port=port,
auth_username=auth_username,
auth_password=auth_password,
auth_private_key=auth_private_key,
auth_private_key_passphrase=auth_private_key_passphrase,
auth_strict_key=auth_strict_key,
auth_bypass=auth_bypass,
timeout_socket=timeout_socket,
timeout_transport=timeout_transport,
timeout_ops=timeout_ops,
comms_return_char=comms_return_char,
comms_ansi=comms_ansi,
ssh_config_file=ssh_config_file,
ssh_known_hosts_file=ssh_known_hosts_file,
on_init=on_init,
on_open=on_open,
on_close=on_close,
transport=transport,
transport_options=transport_options,
channel_log=channel_log,
channel_lock=channel_lock,
logging_uid=logging_uid,
privilege_levels=privilege_levels,
default_desired_privilege_level=default_desired_privilege_level,
auth_secondary=auth_secondary,
failed_when_contains=failed_when_contains,
textfsm_platform=textfsm_platform,
genie_platform=genie_platform,
)
async def _abort_config(self) -> None:
"""
Abort NXOS configuration session (if using a config session!)
Args:
N/A
Returns:
None
Raises:
N/A
"""
# nxos pattern for config sessions should *always* have `config-s`
if "config\\-s" in self._current_priv_level.pattern:
await self.channel.send_input(channel_input="abort")
self._current_priv_level = self.privilege_levels["privilege_exec"]
def register_configuration_session(self, session_name: str) -> None:
"""
NXOS specific implementation of register_configuration_session
Args:
session_name: name of session to register
Returns:
None
Raises:
N/A
"""
self._create_configuration_session(session_name=session_name)
self.update_privilege_levels()
Functions¶
nxos_on_close¶
nxos_on_close(conn: scrapli.driver.network.async_driver.AsyncNetworkDriver) ‑> NoneType
1 2 3 4 5 6 7 8 9 10 | |
nxos_on_open¶
nxos_on_open(conn: scrapli.driver.network.async_driver.AsyncNetworkDriver) ‑> NoneType
1 2 3 4 5 6 7 8 9 10 | |
Classes¶
AsyncNXOSDriver¶
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 | |
Expand source code
class AsyncNXOSDriver(AsyncNetworkDriver, NXOSDriverBase):
def __init__(
self,
host: str,
privilege_levels: Optional[Dict[str, PrivilegeLevel]] = None,
default_desired_privilege_level: str = "privilege_exec",
port: Optional[int] = None,
auth_username: str = "",
auth_password: str = "",
auth_private_key: str = "",
auth_private_key_passphrase: str = "",
auth_strict_key: bool = True,
auth_bypass: bool = False,
timeout_socket: float = 15.0,
timeout_transport: float = 30.0,
timeout_ops: float = 30.0,
comms_return_char: str = "\n",
comms_ansi: Optional[bool] = None,
ssh_config_file: Union[str, bool] = False,
ssh_known_hosts_file: Union[str, bool] = False,
on_init: Optional[Callable[..., Any]] = None,
on_open: Optional[Callable[..., Any]] = None,
on_close: Optional[Callable[..., Any]] = None,
transport: str = "system",
transport_options: Optional[Dict[str, Any]] = None,
channel_log: Union[str, bool, BytesIO] = False,
channel_lock: bool = False,
logging_uid: str = "",
auth_secondary: str = "",
failed_when_contains: Optional[List[str]] = None,
textfsm_platform: str = "cisco_nxos",
genie_platform: str = "nxos",
):
"""
NXOSDriver Object
Please see `scrapli.driver.base.base_driver.Driver` for all "base driver" arguments!
# noqa: DAR101
Args:
privilege_levels: optional user provided privilege levels, if left None will default to
scrapli standard privilege levels
default_desired_privilege_level: string of name of default desired priv, this is the
priv level that is generally used to disable paging/set terminal width and things
like that upon first login, and is also the priv level scrapli will try to acquire
for normal "command" operations (`send_command`, `send_commands`)
auth_secondary: password to use for secondary authentication (enable)
on_open: callable that accepts the class instance as its only argument. this callable,
if provided, is executed immediately after authentication is completed. Common use
cases for this callable would be to disable paging or accept any kind of banner
message that prompts a user upon connection
on_close: callable that accepts the class instance as its only argument. this callable,
if provided, is executed immediately prior to closing the underlying transport.
Common use cases for this callable would be to save configurations prior to exiting,
or to logout properly to free up vtys or similar.
textfsm_platform: string name of textfsm parser platform
genie_platform: string name of cisco genie parser platform
failed_when_contains: List of strings that indicate a command/config has failed
Returns:
None
Raises:
N/A
"""
# somewhere/somehow the mixin is causing mypy to be upset about comms_prompt_pattern...
self.comms_prompt_pattern: str
if privilege_levels is None:
privilege_levels = deepcopy(PRIVS)
if on_open is None:
on_open = nxos_on_open
if on_close is None:
on_close = nxos_on_close
if failed_when_contains is None:
failed_when_contains = FAILED_WHEN_CONTAINS.copy()
super().__init__(
host=host,
port=port,
auth_username=auth_username,
auth_password=auth_password,
auth_private_key=auth_private_key,
auth_private_key_passphrase=auth_private_key_passphrase,
auth_strict_key=auth_strict_key,
auth_bypass=auth_bypass,
timeout_socket=timeout_socket,
timeout_transport=timeout_transport,
timeout_ops=timeout_ops,
comms_return_char=comms_return_char,
comms_ansi=comms_ansi,
ssh_config_file=ssh_config_file,
ssh_known_hosts_file=ssh_known_hosts_file,
on_init=on_init,
on_open=on_open,
on_close=on_close,
transport=transport,
transport_options=transport_options,
channel_log=channel_log,
channel_lock=channel_lock,
logging_uid=logging_uid,
privilege_levels=privilege_levels,
default_desired_privilege_level=default_desired_privilege_level,
auth_secondary=auth_secondary,
failed_when_contains=failed_when_contains,
textfsm_platform=textfsm_platform,
genie_platform=genie_platform,
)
async def _abort_config(self) -> None:
"""
Abort NXOS configuration session (if using a config session!)
Args:
N/A
Returns:
None
Raises:
N/A
"""
# nxos pattern for config sessions should *always* have `config-s`
if "config\\-s" in self._current_priv_level.pattern:
await self.channel.send_input(channel_input="abort")
self._current_priv_level = self.privilege_levels["privilege_exec"]
def register_configuration_session(self, session_name: str) -> None:
"""
NXOS specific implementation of register_configuration_session
Args:
session_name: name of session to register
Returns:
None
Raises:
N/A
"""
self._create_configuration_session(session_name=session_name)
self.update_privilege_levels()
Ancestors (in MRO)¶
- scrapli.driver.network.async_driver.AsyncNetworkDriver
- scrapli.driver.generic.async_driver.AsyncGenericDriver
- scrapli.driver.base.async_driver.AsyncDriver
- scrapli.driver.base.base_driver.BaseDriver
- scrapli.driver.generic.base_driver.BaseGenericDriver
- scrapli.driver.network.base_driver.BaseNetworkDriver
- scrapli.driver.core.cisco_nxos.base_driver.NXOSDriverBase
Class variables¶
auth_secondary: str
failed_when_contains: List[str]
genie_platform: str
logger: logging.LoggerAdapter
privilege_levels: Dict[str, scrapli.driver.network.base_driver.PrivilegeLevel]
textfsm_platform: str
Methods¶
register_configuration_session¶
register_configuration_session(self, session_name: str) ‑> NoneType
1 2 3 4 5 6 7 8 9 10 | |