How to use the netmiko.base_connection.BaseConnection function in netmiko

To help you get started, we’ve selected a few netmiko examples, based on popular ways it is used in public projects.

Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.

github ktbyers / netmiko / tests / unit / test_base_connection.py View on Github external
#!/usr/bin/env python

import time
from os.path import dirname, join
from threading import Lock

from netmiko import NetMikoTimeoutException
from netmiko.base_connection import BaseConnection

RESOURCE_FOLDER = join(dirname(dirname(__file__)), "etc")


class FakeBaseConnection(BaseConnection):
    def __init__(self, **kwargs):
        for key, value in kwargs.items():
            setattr(self, key, value)
        self._session_locker = Lock()


def test_timeout_exceeded():
    """Raise NetMikoTimeoutException if waiting too much"""
    connection = FakeBaseConnection(session_timeout=10)
    start = time.time() - 11
    try:
        connection._timeout_exceeded(start)
    except NetMikoTimeoutException as exc:
        assert isinstance(exc, NetMikoTimeoutException)
        return
github ktbyers / netmiko / netmiko / terminal_server / terminal_server.py View on Github external
"""Generic Terminal Server driver."""
from __future__ import unicode_literals
from netmiko.base_connection import BaseConnection


class TerminalServer(BaseConnection):
    """Generic Terminal Server driver.

    Allow direct write_channel / read_channel operations without session_preparation causing
    an exception.
    """

    def session_preparation(self):
        """Do nothing here; base_prompt is not set; paging is not disabled."""
        pass


class TerminalServerSSH(TerminalServer):
    """Generic Terminal Server driver SSH."""

    pass
github ktbyers / netmiko / netmiko / f5 / f5_tmsh_ssh.py View on Github external
from __future__ import unicode_literals
import time
from netmiko.base_connection import BaseConnection


class F5TmshSSH(BaseConnection):
    def session_preparation(self):
        """Prepare the session after the connection has been established."""
        self._test_channel_read()
        self.set_base_prompt()
        self.tmsh_mode()
        self.set_base_prompt()
        self.disable_paging(
            command="modify cli preference pager disabled display-threshold 0"
        )
        self.clear_buffer()
        cmd = 'run /util bash -c "stty cols 255"'
        self.set_terminal_width(command=cmd)

    def tmsh_mode(self, delay_factor=1):
        """tmsh command is equivalent to config command on F5."""
        delay_factor = self.select_delay_factor(delay_factor)
github ktbyers / netmiko / netmiko / netapp / netapp_cdot_ssh.py View on Github external
from __future__ import unicode_literals

from netmiko.base_connection import BaseConnection


class NetAppcDotSSH(BaseConnection):
    def session_preparation(self):
        """Prepare the session after the connection has been established."""
        self.set_base_prompt()
        cmd = self.RETURN + "rows 0" + self.RETURN
        self.disable_paging(command=cmd)

    def send_command_with_y(self, *args, **kwargs):
        output = self.send_command_timing(*args, **kwargs)
        if "{y|n}" in output:
            output += self.send_command_timing(
                "y", strip_prompt=False, strip_command=False
            )
        return output

    def check_config_mode(self, check_string="*>"):
        return super(NetAppcDotSSH, self).check_config_mode(check_string=check_string)
github ktbyers / netmiko / netmiko / juniper / juniper_screenos.py View on Github external
import time
from netmiko.base_connection import BaseConnection


class JuniperScreenOsSSH(BaseConnection):
    """
    Implement methods for interacting with Juniper ScreenOS devices.
    """

    def session_preparation(self):
        """
        Prepare the session after the connection has been established.

        Disable paging (the '--more--' prompts).
        Set the base prompt for interaction ('>').
        """
        self._test_channel_read()
        self.set_base_prompt()
        self.disable_paging(command="set console page 0")
        # Clear the read buffer
        time.sleep(0.3 * self.global_delay_factor)
github ktbyers / netmiko / netmiko / pluribus / pluribus_ssh.py View on Github external
from __future__ import unicode_literals
import time
from netmiko.base_connection import BaseConnection


class PluribusSSH(BaseConnection):
    """Common methods for Pluribus."""

    def __init__(self, *args, **kwargs):
        super(PluribusSSH, self).__init__(*args, **kwargs)
        self._config_mode = False

    def disable_paging(self, command="pager off", delay_factor=1):
        """Make sure paging is disabled."""
        return super(PluribusSSH, self).disable_paging(
            command=command, delay_factor=delay_factor
        )

    def session_preparation(self):
        """Prepare the netmiko session."""
        self._test_channel_read()
        self.set_base_prompt()
github ktbyers / netmiko / netmiko / paloalto / paloalto_panos.py View on Github external
from __future__ import unicode_literals
import time
import re
from netmiko.base_connection import BaseConnection


class PaloAltoPanosBase(BaseConnection):
    """
    Implement methods for interacting with PaloAlto devices.

    Disables `enable()` and `check_enable_mode()`
    methods.  Overrides several methods for PaloAlto-specific compatibility.
    """

    def session_preparation(self):
        """
        Prepare the session after the connection has been established.

        Disable paging (the '--more--' prompts).
        Set the base prompt for interaction ('>').
        """
        self._test_channel_read()
        self.set_base_prompt(delay_factor=20)
github ktbyers / netmiko / netmiko / rad / rad_etx.py View on Github external
import time
from netmiko.base_connection import BaseConnection


class RadETXBase(BaseConnection):
    """RAD ETX Support, Tested on RAD 203AX, 205A and 220A."""

    def session_preparation(self):
        self._test_channel_read()
        self.set_base_prompt()
        self.disable_paging(command="config term length 0")
        # Clear the read buffer
        time.sleep(0.3 * self.global_delay_factor)
        self.clear_buffer()

    def save_config(self, cmd="admin save", confirm=False, confirm_response=""):
        """Saves Config Using admin save."""
        if confirm:
            output = self.send_command_timing(command_string=cmd)
            if confirm_response:
                output += self.send_command_timing(confirm_response)
github ktbyers / netmiko / netmiko / dell / dell_isilon_ssh.py View on Github external
import time
import re

from netmiko.base_connection import BaseConnection


class DellIsilonSSH(BaseConnection):
    def set_base_prompt(
        self, pri_prompt_terminator="$", alt_prompt_terminator="#", delay_factor=1
    ):
        """Determine base prompt."""
        return super(DellIsilonSSH, self).set_base_prompt(
            pri_prompt_terminator=pri_prompt_terminator,
            alt_prompt_terminator=alt_prompt_terminator,
            delay_factor=delay_factor,
        )

    def strip_ansi_escape_codes(self, string_buffer):
        """Remove Null code"""
        output = re.sub(r"\x00", "", string_buffer)
        return super(DellIsilonSSH, self).strip_ansi_escape_codes(output)

    def session_preparation(self):
github ktbyers / netmiko / netmiko / checkpoint / checkpoint_gaia_ssh.py View on Github external
from __future__ import unicode_literals
import time
from netmiko.base_connection import BaseConnection


class CheckPointGaiaSSH(BaseConnection):
    """
    Implements methods for communicating with Check Point Gaia
    firewalls.
    """

    def session_preparation(self):
        """
        Prepare the session after the connection has been established.

        Set the base prompt for interaction ('>').
        """
        self._test_channel_read()
        self.set_base_prompt()
        self.disable_paging(command="set clienv rows 0")
        # Clear the read buffer
        time.sleep(0.3 * self.global_delay_factor)