christianmanivong
db6b2e4ea7
STANDARDS.md binds every repository in the NAPALM org: one default branch, Conventional Commits, the device-types contract, test-driven development, clean-code limits, a quality ratchet for existing code, packaging rules and releases by tag to PyPI. napalm-standard holds what a tool can check: the conformance check (with release rules on a tag), the quality ratchet (ported from netOrk's quality_ratchet.py), and the Gitea release from the CHANGELOG section. .gitea/workflows/driver.yml is the CI every driver calls; template/ is a new driver's starting point and passes the check as it is.
94 lines
2.5 KiB
Python
94 lines
2.5 KiB
Python
"""NAPALM driver for Example switches over SSH.
|
|
|
|
Optional args
|
|
-------------
|
|
port : int
|
|
SSH port (default 22).
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
from typing import Any, Protocol
|
|
|
|
from napalm.base.exceptions import ConnectionException
|
|
from napalm.base.models import FactsDict
|
|
from napalm_device_types import SwitchDriver
|
|
|
|
from napalm_example.parsers import parse_version
|
|
|
|
LOG = logging.getLogger(__name__)
|
|
|
|
VENDOR = "Example"
|
|
DEFAULT_PORT = 22
|
|
|
|
|
|
class Transport(Protocol):
|
|
def command(self, line: str) -> str: ...
|
|
|
|
def close(self) -> None: ...
|
|
|
|
|
|
class ExampleDriver(SwitchDriver):
|
|
"""A switch that answers line commands over SSH."""
|
|
|
|
def __init__(
|
|
self,
|
|
hostname: str,
|
|
username: str,
|
|
password: str,
|
|
timeout: int = 60,
|
|
optional_args: dict[str, Any] | None = None,
|
|
) -> None:
|
|
self.hostname = hostname
|
|
self.username = username
|
|
self.password = password
|
|
self.timeout = timeout
|
|
self.port = int((optional_args or {}).get("port", DEFAULT_PORT))
|
|
self.transport: Transport | None = None
|
|
|
|
def open(self) -> None: # pragma: no cover - live I/O
|
|
from netmiko import ConnectHandler
|
|
|
|
session = ConnectHandler(
|
|
device_type="generic",
|
|
host=self.hostname,
|
|
port=self.port,
|
|
username=self.username,
|
|
password=self.password,
|
|
timeout=self.timeout,
|
|
)
|
|
|
|
class _Netmiko:
|
|
def command(self, line: str) -> str:
|
|
return str(session.send_command(line))
|
|
|
|
def close(self) -> None:
|
|
session.disconnect()
|
|
|
|
self.transport = _Netmiko()
|
|
|
|
def close(self) -> None:
|
|
if self.transport is not None:
|
|
self.transport.close()
|
|
self.transport = None
|
|
|
|
def _command(self, line: str) -> str:
|
|
if self.transport is None:
|
|
raise ConnectionException(f"{self.hostname}: not connected; call open() first")
|
|
LOG.debug("%s: %s", self.hostname, line)
|
|
return self.transport.command(line)
|
|
|
|
def get_facts(self) -> FactsDict:
|
|
version = parse_version(self._command("show version"))
|
|
return {
|
|
"vendor": VENDOR,
|
|
"model": version["model"],
|
|
"hostname": version["hostname"],
|
|
"fqdn": version["hostname"],
|
|
"os_version": version["os_version"],
|
|
"serial_number": version["serial_number"],
|
|
"uptime": version["uptime"],
|
|
"interface_list": [],
|
|
}
|