feat: standards, shared CI and release tooling for NAPALM drivers
CI / test (3.11) (push) Successful in 34s
CI / test (3.12) (push) Successful in 30s
CI / test (3.13) (push) Successful in 37s

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.
This commit is contained in:
christianmanivong committed 2026-10-22 19:01:14 +02:00
commit db6b2e4ea7
37 files changed
+2894

No files matched your search

+5
View File
@@ -0,0 +1,5 @@
"""NAPALM driver for Example switches."""
from napalm_example.driver import ExampleDriver
__all__ = ["ExampleDriver"]
+93
View File
@@ -0,0 +1,93 @@
"""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": [],
}
@@ -0,0 +1,45 @@
"""Pure functions: device output in, typed values out. No I/O here."""
from __future__ import annotations
import re
from typing import TypedDict
#: "Key : value" lines of `show version`.
_FIELD = re.compile(r"^(?P<key>[^:\n]+?)\s*:\s*(?P<value>.*)$", re.MULTILINE)
#: "3 days, 4:05:06" or "4:05:06".
_UPTIME = re.compile(r"^(?:(?P<days>\d+) days?, )?(?P<h>\d+):(?P<m>\d\d):(?P<s>\d\d)$")
_SECONDS_PER_DAY = 86400
class VersionDict(TypedDict):
model: str
hostname: str
os_version: str
serial_number: str
uptime: float
def parse_uptime(text: str) -> float:
"""Seconds from an uptime like ``3 days, 4:05:06``."""
match = _UPTIME.match(text.strip())
if not match:
raise ValueError(f"unreadable uptime: {text!r}")
days = int(match.group("days") or 0)
hours, minutes, seconds = (int(match.group(k)) for k in ("h", "m", "s"))
return float(days * _SECONDS_PER_DAY + hours * 3600 + minutes * 60 + seconds)
def parse_version(text: str) -> VersionDict:
"""Facts from ``show version``; a field the device did not print is an error."""
fields = {m.group("key").strip(): m.group("value").strip() for m in _FIELD.finditer(text)}
try:
return {
"model": text.splitlines()[0].removesuffix(" Switch").strip(),
"hostname": fields["Hostname"],
"os_version": fields["Firmware"],
"serial_number": fields["Serial number"],
"uptime": parse_uptime(fields["Uptime"]),
}
except KeyError as exc:
raise ValueError(f"show version lacks {exc.args[0]}") from exc
View File
Whitespace-only changes.