NAPALM/napalm-netgear
1
0 Fork 0
Code Issues 4 Pull Requests Actions 1 Packages Projects Releases Wiki Activity

chore: adopt the NAPALM driver standard #15

Merged
christianmanivong merged 5 commits from chore/driver-standard into main 2026-10-23 10:03:35 +00:00
Conversation 0 Commits 5 Files Changed 13
+495 -248
13 changed files with 495 additions and 248 deletions
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

.gitea/workflows/ci.yml
+15
View File
@@ -0,0 +1,15 @@
name: CI
on:
push:
branches: [main]
tags: ["v*"]
pull_request:
jobs:
driver:
uses: NAPALM/driver-standard/.gitea/workflows/driver.yml@main
with:
standard-ref: main
secrets:
PYPI_TOKEN: ${{ secrets.PYPI_TOKEN }}
.github/workflows/ci.yml
-48
View File
@@ -1,48 +0,0 @@
name: CI
on:
push:
branches: ["**"]
pull_request:
branches: ["**"]
jobs:
test:
runs-on: ubuntu-latest
strategy:
fail-fast: false
matrix:
python-version: ["3.10", "3.11", "3.12"]
steps:
- name: Checkout
uses: actions/checkout@v4
- name: Setup Python
uses: actions/setup-python@v5
with:
python-version: ${{ matrix.python-version }}
cache: pip
- name: Install package with dev extras
run: |
python -m pip install --upgrade pip
# napalm-device-types lives in git.netork.io/NAPALM, not on PyPI: without this
# pip looks there, finds an unrelated 0.1.0 and the job dies before any test.
python -m pip install "napalm-device-types @ git+https://git.netork.io/NAPALM/napalm-device-types.git"
python -m pip install -e ".[dev]"
- name: Run unit tests
run: |
python -m pytest -q --tb=short
- name: Build wheel and sdist
run: |
python -m pip install build
python -m build
- name: Upload dist artifacts
# v4 refuses to run on Gitea ("not currently supported on GHES").
uses: actions/upload-artifact@v3
with:
name: dist-${{ matrix.python-version }}
path: dist/*
CHANGELOG.md
+24
View File
@@ -0,0 +1,24 @@
# Changelog
All notable changes to this driver. The format follows
[Keep a Changelog](https://keepachangelog.com/), versions follow
[Semantic Versioning](https://semver.org/).
## [Unreleased]
## [0.6.0] - 2026-10-23
The first release on PyPI, under the new name `napalm-netgear-switches`.
### Changed
- **The package is now called `napalm-netgear-switches`** on PyPI
(`pip install napalm-netgear-switches`): the name `napalm-netgear` belongs to another
project. The import package `napalm_netgear` and the driver names `netgear_smart` and
`netgear_plus` stay the same.
- The driver follows the NAPALM driver standard
(https://git.netork.io/NAPALM/driver-standard): CI on Gitea runs ruff, mypy, the
tests on Python 3.10–3.13 and at the lowest declared versions, bandit, pip-audit and
a build check, and a release tag publishes to PyPI. The package now ships `py.typed`
and a LICENSE file. **Python 3.8 and 3.9 are no longer supported**; napalm-device-types
never ran on them. It needs napalm-device-types 4.21.1 or later.
- A failed logout when a Netgear Plus session is closed is now logged at debug level.
LICENSE
+191
View File
@@ -0,0 +1,191 @@
Apache License
Version 2.0, January 2004
http://www.apache.org/licenses/
TERMS AND CONDITIONS FOR USE, REPRODUCTION, AND DISTRIBUTION
1. Definitions.
"License" shall mean the terms and conditions for use, reproduction,
and distribution as defined by Sections 1 through 9 of this document.
"Licensor" shall mean the copyright owner or entity authorized by
the copyright owner that is granting the License.
"Legal Entity" shall mean the union of the acting entity and all
other entities that control, are controlled by, or are under common
control with that entity. For the purposes of this definition,
"control" means (i) the power, direct or indirect, to cause the
direction or management of such entity, whether by contract or
otherwise, or (ii) ownership of fifty percent (50%) or more of the
outstanding shares, or (iii) beneficial ownership of such entity.
"You" (or "Your") shall mean an individual or Legal Entity
exercising permissions granted by this License.
"Source" form shall mean the preferred form for making modifications,
including but not limited to software source code, documentation
source, and configuration files.
"Object" form shall mean any form resulting from mechanical
transformation or translation of a Source form, including but
not limited to compiled object code, generated documentation,
and conversions to other media types.
"Work" shall mean the work of authorship made available under
the License, as indicated by a copyright notice that is included in
or attached to the work (an example is provided in the Appendix below).
"Derivative Works" shall mean any work, whether in Source or Object
form, that is based on (or derived from) the Work and for which the
editorial revisions, annotations, elaborations, or other transformations
represent, as a whole, an original work of authorship. For the purposes
of this License, Derivative Works shall not include works that remain
separable from, or merely link (or bind by name) to the interfaces of,
the Work and Derivative Works thereof.
"Contribution" shall mean, as submitted to the Licensor for inclusion
in the Work by the copyright owner or by an individual or Legal Entity
authorized to submit on behalf of the copyright owner. For the purposes
of this definition, "submitted" means any form of electronic, verbal,
or written communication sent to the Licensor or its representatives,
including but not limited to communication on electronic mailing lists,
source code control systems, and issue tracking systems that are managed
by, or on behalf of, the Licensor for the purpose of discussing and
improving the Work, but excluding communication that is conspicuously
marked or designated in writing by the copyright owner as "Not a
Contribution."
"Contributor" shall mean Licensor and any Legal Entity on behalf of
whom a Contribution has been received by the Licensor and incorporated
within the Work.
2. Grant of Copyright License. Subject to the terms and conditions of
this License, each Contributor hereby grants to You a perpetual,
worldwide, non-exclusive, no-charge, royalty-free, irrevocable
copyright license to reproduce, prepare Derivative Works of,
publicly display, publicly perform, sublicense, and distribute the
Work and such Derivative Works in Source or Object form.
3. Grant of Patent License. Subject to the terms and conditions of
this License, each Contributor hereby grants to You a perpetual,
worldwide, non-exclusive, no-charge, royalty-free, irrevocable
(except as stated in this section) patent license to make, have made,
use, offer to sell, sell, import, and otherwise transfer the Work,
where such license applies only to those patent claims licensable
by such Contributor that are necessarily infringed by their
Contribution(s) alone or by the combined Contribution(s) and the Work
to which such Contribution(s) was submitted. If You institute patent
litigation against any entity (including a cross-claim or counterclaim
in a lawsuit) alleging that the Work or any Contribution embodied
within the Work constitutes direct or contributory patent infringement,
then any patent licenses granted to You under this License for that
Work shall terminate as of the date such litigation is filed.
4. Redistribution. You may reproduce and distribute copies of the
Work or Derivative Works thereof in any medium, with or without
modifications, and in Source or Object form, provided that You
meet the following conditions:
(a) You must give any other recipients of the Work or Derivative
Works a copy of this License; and
(b) You must cause any modified files to carry prominent notices
stating that You changed the files; and
(c) You must retain, in the Source form of any Derivative Works
that You distribute, all copyright, patent, trademark, and
attribution notices from the Source form of the Work,
excluding those notices that do not pertain to any part of
the Derivative Works; and
(d) If the Work includes a "NOTICE" text file as part of its
distribution, You must include a readable copy of the
attribution notices contained within such NOTICE file, in
at least one of the following places: within a NOTICE text file
distributed as part of the Derivative Works; within the Source
form or documentation, if provided along with the Derivative
Works; or, within a display generated by the Derivative Works,
if and wherever such third-party notices normally appear. The
contents of the NOTICE file are for informational purposes only
and do not modify the License. You may add Your own attribution
notices within Derivative Works that You distribute, alongside
or as an addendum to the NOTICE text from the Work, provided
that such additional attribution notices cannot be construed
as modifying the License.
You may add Your own license statement for Your modifications and
may provide additional grant of rights to use, copy, modify, merge,
publish, distribute, sublicense, and/or sell copies of the Work,
and to permit persons to whom the Work is furnished to do so,
subject to the following conditions:
5. Submission of Contributions. Unless You explicitly state otherwise,
any Contribution intentionally submitted for inclusion in the Work
by You to the Licensor shall be under the terms and conditions of
this License, without any additional terms or conditions.
Notwithstanding the above, nothing herein shall supersede or modify
the terms of any separate license agreement you may have executed
with Licensor regarding such Contributions.
6. Trademarks. This License does not grant permission to use the trade
names, trademarks, service marks, or product names of the Licensor,
except as required for reasonable and customary use in describing the
origin of the Work and reproducing the content of the NOTICE file.
7. Disclaimer of Warranty. Unless required by applicable law or
agreed to in writing, Licensor provides the Work (and each
Contributor provides its Contributions) on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
implied, including, without limitation, any conditions of TITLE,
MERCHANTIBILITY, or FITNESS FOR A PARTICULAR PURPOSE.
See the License for the specific language governing permissions and
limitations under the License.
8. Limitation of Liability. In no event and under no legal theory,
whether in tort (including negligence), contract, or otherwise,
unless required by applicable law (such as deliberate and grossly
negligent acts) or agreed to in writing, shall any Contributor be
liable to You for damages, including any direct, indirect, special,
incidental, or exemplary damages of any character arising as a
result of this License or out of the use or inability to use the
Work (even if such Contributor has been advised of the possibility
of such damages).
9. Accepting Warranty or Additional Liability. While redistributing
the Work or Derivative Works thereof, You may choose to offer,
and charge a fee for, acceptance of support, warranty, indemnity,
or other liability obligations and/or rights consistent with this
License. However, in accepting such obligations, You may offer only
conditions consistent with this License, subject to the following
additional conditions:
(a) You may offer additional warranty, indemnity, or other liability
terms and conditions; and
(b) You may offer support for the Work in return for a fee.
END OF TERMS AND CONDITIONS
APPENDIX: How to apply the Apache License to your work.
To apply the Apache License to your work, attach the following
boilerplate notice, with the fields enclosed by brackets "[]"
replaced with your own identifying information. (Don't include
the brackets!) The text should be enclosed in the appropriate
comment syntax for the file format in question. It may also be
possible to make it available with a creative commons license.
Copyright 2026 Christian Manivong
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
README.md
+12 -6
View File
@@ -1,4 +1,4 @@
# napalm-netgear
# napalm-netgear-switches
NAPALM community driver for **Netgear Smart Managed** and **Netgear Plus Managed** switches
(GS108EPP, GS110TP, GS316EPP and compatible Smart Managed Pro series).
@@ -18,20 +18,24 @@ NAPALM community driver for **Netgear Smart Managed** and **Netgear Plus Managed
| Dependency | Minimum version |
|---|---|
| Python | 3.8 |
| Python | 3.10 |
| NAPALM | 4.0 |
| napalm-device-types | 4.21.1 |
| Netmiko | 4.0 |
## Installation
```bash
pip install napalm-netgear
pip install napalm-netgear-switches
```
The name `napalm-netgear` on PyPI belongs to another project. The import package is
still `napalm_netgear`, and the drivers are still `netgear_smart` and `netgear_plus`.
Or from source:
```bash
git clone https://github.com/napalm-automation-community/napalm-netgear
git clone https://git.netork.io/NAPALM/napalm-netgear.git
cd napalm-netgear
pip install -e .
```
@@ -102,9 +106,11 @@ Any additional keyword arguments are forwarded to Netmiko.
## Development
The rules are in [STANDARDS.md](https://git.netork.io/NAPALM/driver-standard/src/branch/main/STANDARDS.md).
```bash
pip install -e ".[dev]"
pytest tests/
python -m venv .venv && .venv/bin/pip install -e ".[dev]"
.venv/bin/pytest --cov
```
## License
napalm_netgear/__init__.py
+1 -1
View File
@@ -1,6 +1,6 @@
"""NAPALM driver for Netgear Smart Managed switches."""
from napalm_netgear.netgear_smart import NetgearSmartDriver
from napalm_netgear.netgear_plus import NetgearPlusDriver
from napalm_netgear.netgear_smart import NetgearSmartDriver
__all__ = ["NetgearSmartDriver", "NetgearPlusDriver"]
napalm_netgear/netgear_plus.py
+49 -64
View File
@@ -1,4 +1,3 @@
# -*- coding: utf-8 -*-
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
@@ -22,12 +21,15 @@ The ``username`` parameter is accepted but ignored.
from __future__ import annotations
import logging
import threading
import time
from typing import Any
from napalm_device_types import FingerprintRule, FirmwareOfferMixin, SwitchDriver
from napalm.base.exceptions import ConnectionException
from napalm_device_types import FingerprintRule, FirmwareOfferMixin, SwitchDriver
logger = logging.getLogger(__name__)
# ---------------------------------------------------------------------------
# Firmware update check — module-level constants and cache
@@ -41,9 +43,9 @@ _FW_CDN_BASE = "https://www.downloads.netgear.com/files/GDC"
# Verified via HEAD requests against downloads.netgear.com on 2026-05-24.
# Update this table when Netgear releases new firmware.
_FW_KNOWN_LATEST: dict[str, str] = {
"GS305EP": "V2.0.0.10",
"GS305EP": "V2.0.0.10",
"GS305EPP": "V2.0.0.10",
"GS308EP": "V2.0.0.11",
"GS308EP": "V2.0.0.11",
"GS308EPP": "V2.0.0.11",
}
@@ -63,15 +65,15 @@ def _fw_version_tuple(version: str) -> tuple[int, ...]:
def _fw_cdn_exists(model: str, version: str) -> bool:
"""Return True if a firmware zip exists on the Netgear CDN (HEAD request)."""
import urllib.request
import ssl
import urllib.request
url = f"{_FW_CDN_BASE}/{model}/{model}_{version}.zip"
ctx = ssl.create_default_context()
req = urllib.request.Request(url, method="HEAD",
headers={"User-Agent": "Mozilla/5.0"})
req = urllib.request.Request(url, method="HEAD", headers={"User-Agent": "Mozilla/5.0"})
try:
with urllib.request.urlopen(req, timeout=3, context=ctx):
# The URL is always https on _FW_CDN_BASE; model and version only fill the path.
with urllib.request.urlopen(req, timeout=3, context=ctx): # nosec B310
return True
except Exception:
return False
@@ -150,7 +152,15 @@ class NetgearPlusDriver(FirmwareOfferMixin, SwitchDriver):
VENDOR = "Netgear"
DRIVER_NAME = "netgear_plus"
SNMP_OBJECT_ID_PREFIX = "1.3.6.1.4.1.4526"
OUI_PREFIXES = ["00:09:5B", "20:4E:7F", "E0:91:F5", "A0:40:A0", "C4:04:15", "00:1B:2F", "00:1F:33"]
OUI_PREFIXES = [
"00:09:5B",
"20:4E:7F",
"E0:91:F5",
"A0:40:A0",
"C4:04:15",
"00:1B:2F",
"00:1F:33",
]
HTTP_FINGERPRINT = [
FingerprintRule("netgear gs1", weight=8.0),
FingerprintRule("netgear gs2", weight=8.0),
@@ -191,8 +201,7 @@ class NetgearPlusDriver(FirmwareOfferMixin, SwitchDriver):
from py_netgear_plus import NetgearSwitchConnector
except ImportError as exc:
raise ConnectionException(
"py-netgear-plus is not installed. "
"Add 'py-netgear-plus' to your dependencies."
"py-netgear-plus is not installed. Add 'py-netgear-plus' to your dependencies."
) from exc
try:
@@ -207,9 +216,7 @@ class NetgearPlusDriver(FirmwareOfferMixin, SwitchDriver):
except Exception as exc:
if isinstance(exc, ConnectionException):
raise
raise ConnectionException(
f"Cannot connect to {self.hostname}: {exc}"
) from exc
raise ConnectionException(f"Cannot connect to {self.hostname}: {exc}") from exc
def close(self) -> None:
"""Log out from the switch."""
@@ -217,7 +224,7 @@ class NetgearPlusDriver(FirmwareOfferMixin, SwitchDriver):
try:
self._connector.delete_login_cookie()
except Exception:
pass
logger.debug("Logging out of the switch failed: ignored", exc_info=True)
self._connector = None
self._switch_infos_cache = None
@@ -285,9 +292,7 @@ class NetgearPlusDriver(FirmwareOfferMixin, SwitchDriver):
hostname = infos.get("switch_name", self.hostname)
os_version = infos.get("switch_firmware", "")
serial_number = infos.get("switch_serial_number", "")
interface_list = [
self._port_name(i) for i in range(1, self._connector.ports + 1)
]
interface_list = [self._port_name(i) for i in range(1, self._connector.ports + 1)]
return {
"vendor": self.VENDOR,
"model": str(model),
@@ -313,7 +318,9 @@ class NetgearPlusDriver(FirmwareOfferMixin, SwitchDriver):
name = self._port_name(port)
status = infos.get(f"port_{port}_status", "off") == "on"
speed = float(infos.get(f"port_{port}_connection_speed", 0) or 0)
description = port_names.get(port, "") or infos.get(f"port_{port}_description", "") or ""
description = (
port_names.get(port, "") or infos.get(f"port_{port}_description", "") or ""
)
interfaces[name] = {
"is_up": status,
"is_enabled": True,
@@ -372,9 +379,7 @@ class NetgearPlusDriver(FirmwareOfferMixin, SwitchDriver):
"""
infos = self._get_switch_infos()
power: dict[str, Any] = {}
poe_max_single = getattr(
self._connector.switch_model, "POE_MAX_POWER_SINGLE_PORT", None
)
poe_max_single = getattr(self._connector.switch_model, "POE_MAX_POWER_SINGLE_PORT", None)
for port in self._connector.poe_ports:
poe_active = infos.get(f"port_{port}_poe_power_active")
poe_status = bool(poe_active) if poe_active is not None else False
@@ -450,8 +455,8 @@ class NetgearPlusDriver(FirmwareOfferMixin, SwitchDriver):
# Example: <span vid-4>...<i ...></i>24</span><span vnm-4 ...>Guest</span>
vlan_names: dict[str, str] = {}
vlan_entry_re = re.compile(
r'<span\s+vid-4[^>]*>.*?(\d+)\s*</span>\s*'
r'<span\s+vnm-4[^>]*>([^<]*)</span>',
r"<span\s+vid-4[^>]*>.*?(\d+)\s*</span>\s*"
r"<span\s+vnm-4[^>]*>([^<]*)</span>",
re.DOTALL,
)
for m in vlan_entry_re.finditer(text):
@@ -463,15 +468,14 @@ class NetgearPlusDriver(FirmwareOfferMixin, SwitchDriver):
return {}
vlans: dict[str, Any] = {
vid: {"name": name, "tagged": [], "untagged": []}
for vid, name in vlan_names.items()
vid: {"name": name, "tagged": [], "untagged": []} for vid, name in vlan_names.items()
}
# Parse per-port PVID membership.
# Each <span pvid-str ...>1*, 24, 25,</span> corresponds to one port
# in order (port 1, port 2, ...). A VID followed by "*" is the native
# (untagged) VLAN for that port; all others are tagged.
pvid_span_re = re.compile(r'<span\s[^>]*pvid-str[^>]*>([^<]+)</span>')
pvid_span_re = re.compile(r"<span\s[^>]*pvid-str[^>]*>([^<]+)</span>")
for port_idx, pvid_str in enumerate(pvid_span_re.finditer(text)):
port_num = port_idx + 1
if port_num > self._connector.ports:
@@ -540,34 +544,24 @@ class NetgearPlusDriver(FirmwareOfferMixin, SwitchDriver):
return {"running": "", "startup": "", "candidate": ""}
def load_merge_candidate(self, filename: str | None = None, config: str | None = None) -> None:
raise NotImplementedError(
"Config management is not supported for Netgear Plus switches."
)
raise NotImplementedError("Config management is not supported for Netgear Plus switches.")
def load_replace_candidate(self, filename: str | None = None, config: str | None = None) -> None:
raise NotImplementedError(
"Config management is not supported for Netgear Plus switches."
)
def load_replace_candidate(
self, filename: str | None = None, config: str | None = None
) -> None:
raise NotImplementedError("Config management is not supported for Netgear Plus switches.")
def compare_config(self) -> str:
raise NotImplementedError(
"Config management is not supported for Netgear Plus switches."
)
raise NotImplementedError("Config management is not supported for Netgear Plus switches.")
def commit_config(self, message: str = "") -> None:
raise NotImplementedError(
"Config management is not supported for Netgear Plus switches."
)
raise NotImplementedError("Config management is not supported for Netgear Plus switches.")
def discard_config(self) -> None:
raise NotImplementedError(
"Config management is not supported for Netgear Plus switches."
)
raise NotImplementedError("Config management is not supported for Netgear Plus switches.")
def rollback(self) -> None:
raise NotImplementedError(
"Config management is not supported for Netgear Plus switches."
)
raise NotImplementedError("Config management is not supported for Netgear Plus switches.")
def has_pending_commit(self) -> bool:
return False
@@ -587,9 +581,7 @@ class NetgearPlusDriver(FirmwareOfferMixin, SwitchDriver):
Not supported on Netgear Plus switches via HTTP.
"""
raise NotImplementedError(
"ping() is not supported for Netgear Plus switches."
)
raise NotImplementedError("ping() is not supported for Netgear Plus switches.")
def traceroute(
self,
@@ -603,9 +595,7 @@ class NetgearPlusDriver(FirmwareOfferMixin, SwitchDriver):
Not supported on Netgear Plus switches via HTTP.
"""
raise NotImplementedError(
"traceroute() is not supported for Netgear Plus switches."
)
raise NotImplementedError("traceroute() is not supported for Netgear Plus switches.")
def cli(self, commands: list[str], encoding: str = "text") -> dict[str, Any]:
"""Send CLI commands.
@@ -689,12 +679,8 @@ class NetgearPlusDriver(FirmwareOfferMixin, SwitchDriver):
Reports PoE power status from switch infos.
"""
infos = self._get_switch_infos()
poe_max_all = getattr(
self._connector.switch_model, "POE_MAX_POWER_ALL_PORTS", None
)
poe_max_single = getattr(
self._connector.switch_model, "POE_MAX_POWER_SINGLE_PORT", None
)
poe_max_all = getattr(self._connector.switch_model, "POE_MAX_POWER_ALL_PORTS", None)
poe_max_single = getattr(self._connector.switch_model, "POE_MAX_POWER_SINGLE_PORT", None)
ports: dict[str, Any] = {}
for p in self._connector.poe_ports:
poe_active = infos.get(f"port_{p}_poe_power_active")
@@ -742,9 +728,7 @@ class NetgearPlusDriver(FirmwareOfferMixin, SwitchDriver):
Not supported via HTTP API.
"""
raise NotImplementedError(
"set_interface() is not supported for Netgear Plus switches."
)
raise NotImplementedError("set_interface() is not supported for Netgear Plus switches.")
# ── SNMP / Health ──────────────────────────────────────────────────────────
@@ -763,9 +747,10 @@ class NetgearPlusDriver(FirmwareOfferMixin, SwitchDriver):
decide (napalm-device-types ``FirmwareOfferMixin``).
"""
try:
model_name: str = getattr(
self._connector.switch_model, "MODEL_NAME", ""
) or self._connector.switch_model.__class__.__name__
model_name: str = (
getattr(self._connector.switch_model, "MODEL_NAME", "")
or self._connector.switch_model.__class__.__name__
)
current_fw: str = self._get_switch_infos().get("switch_firmware", "")
latest_fw = _fw_get_latest(model_name) if model_name and current_fw else None
if latest_fw is None:
napalm_netgear/netgear_smart.py
+104 -106
View File
@@ -1,4 +1,3 @@
# -*- coding: utf-8 -*-
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
@@ -28,15 +27,11 @@ CLI notes
"""
import re
import socket
import time
from typing import Any, ClassVar, Dict, List, Optional, Union
from typing import Any, ClassVar
import napalm.base.constants as C
import netaddr
from netmiko import ConnectHandler
from netmiko.exceptions import NetmikoAuthenticationException, NetmikoTimeoutException
from napalm_device_types import ConfigLifecycleMixin, FingerprintRule, SnmpAgentMixin, SwitchDriver
from napalm.base import helpers as napalm_helpers
from napalm.base.exceptions import (
CommandErrorException,
@@ -46,8 +41,10 @@ from napalm.base.exceptions import (
ReplaceConfigException,
)
from napalm.base.netmiko_helpers import netmiko_args
from napalm_device_types import ConfigLifecycleMixin, FingerprintRule, SnmpAgentMixin, SwitchDriver
from napalm_device_types.ssh_host_keys import HostKeyCheck, SshHostKeyMismatch
import napalm.base.constants as C
from netmiko import ConnectHandler
from netmiko.exceptions import NetmikoAuthenticationException, NetmikoTimeoutException
class NetgearSmartDriver(ConfigLifecycleMixin, SnmpAgentMixin, SwitchDriver):
@@ -73,7 +70,7 @@ class NetgearSmartDriver(ConfigLifecycleMixin, SnmpAgentMixin, SwitchDriver):
#: Where the switch publishes its health values over SNMP
#: (napalm-device-types ``SnmpHealthProfile``); the application polls.
#: CPU is text ("5 Secs ( 3.2%) 60 Secs ( …"); the five-second figure counts.
SNMP_HEALTH_PROFILE: ClassVar[Dict[str, Any]] = {
SNMP_HEALTH_PROFILE: ClassVar[dict[str, Any]] = {
"uptime": {"oids": ["1.3.6.1.2.1.1.3.0"]},
"cpu": {
"oids": ["1.3.6.1.4.1.4526.100.1.1.4.9.0"],
@@ -97,13 +94,13 @@ class NetgearSmartDriver(ConfigLifecycleMixin, SnmpAgentMixin, SwitchDriver):
username: str,
password: str,
timeout: int = 60,
optional_args: Optional[Dict] = None,
optional_args: dict | None = None,
) -> None:
self.hostname = hostname
self.username = username
self.password = password
self.timeout = timeout
self.device: Optional[ConnectHandler] = None
self.device: ConnectHandler | None = None
if optional_args is None:
optional_args = {}
@@ -131,16 +128,16 @@ class NetgearSmartDriver(ConfigLifecycleMixin, SnmpAgentMixin, SwitchDriver):
self._host_key = HostKeyCheck(optional_args, hostname, self.port)
# Config management state
self._candidate_config: Optional[str] = None
self._candidate_mode: Optional[str] = None # 'merge' or 'replace'
self._backup_config: Optional[str] = None
self._candidate_config: str | None = None
self._candidate_mode: str | None = None # 'merge' or 'replace'
self._backup_config: str | None = None
# CLI style detection cache: True = "v7" (Cisco-like, GigabitEthernet/lag
# interfaces, "show info"/"show interfaces ... status"), False = legacy
# FASTPATH-style (0/1, ch1, "show sysinfo"/"show port all"/"show vlan").
self._cli_v7: Optional[bool] = None
self._port_count_v7: Optional[int] = None
self._version_cache: Optional[str] = None
self._cli_v7: bool | None = None
self._port_count_v7: int | None = None
self._version_cache: str | None = None
# ------------------------------------------------------------------
# Connection management
@@ -179,20 +176,20 @@ class NetgearSmartDriver(ConfigLifecycleMixin, SnmpAgentMixin, SwitchDriver):
self.device.disconnect()
self.device = None
def is_alive(self) -> Dict[str, bool]:
def is_alive(self) -> dict[str, bool]:
"""Return connection liveness without writing to the channel."""
if self.device is None:
return {"is_alive": False}
try:
return {"is_alive": self.device.remote_conn.transport.is_active()}
except (socket.error, EOFError, AttributeError):
except (OSError, EOFError, AttributeError):
return {"is_alive": False}
# ------------------------------------------------------------------
# Internal helpers
# ------------------------------------------------------------------
def _send_command(self, command: Union[str, List[str]]) -> str:
def _send_command(self, command: str | list[str]) -> str:
"""Send a command (or ordered list of fallback commands) to the device.
Uses an explicit ``expect_string`` so that Netmiko never confuses a
@@ -216,7 +213,7 @@ class NetgearSmartDriver(ConfigLifecycleMixin, SnmpAgentMixin, SwitchDriver):
return output
return output
return _do_send(command)
except (socket.error, EOFError) as exc:
except (OSError, EOFError) as exc:
raise ConnectionClosedException(str(exc)) from exc
# ------------------------------------------------------------------
@@ -260,10 +257,10 @@ class NetgearSmartDriver(ConfigLifecycleMixin, SnmpAgentMixin, SwitchDriver):
read_timeout=self.timeout,
)
def _apply_config_lines(self, config_text: str) -> List[str]:
def _apply_config_lines(self, config_text: str) -> list[str]:
"""Send config lines while in config mode; return list of rejected lines."""
ep_any = self._any_prompt()
errors: List[str] = []
errors: list[str] = []
for line in config_text.splitlines():
stripped = line.strip()
if not stripped or stripped.startswith(("!", "#")):
@@ -363,7 +360,7 @@ class NetgearSmartDriver(ConfigLifecycleMixin, SnmpAgentMixin, SwitchDriver):
return full
@staticmethod
def _expand_port_list_v7(raw: str) -> List[str]:
def _expand_port_list_v7(raw: str) -> list[str]:
"""Expand a "v7" interface-list token into individual interface names.
Input examples::
@@ -374,7 +371,7 @@ class NetgearSmartDriver(ConfigLifecycleMixin, SnmpAgentMixin, SwitchDriver):
raw = raw.strip()
if not raw or raw == "---":
return []
ports: List[str] = []
ports: list[str] = []
for token in raw.split(","):
token = token.strip()
if not token:
@@ -402,7 +399,7 @@ class NetgearSmartDriver(ConfigLifecycleMixin, SnmpAgentMixin, SwitchDriver):
# NAPALM getters
# ------------------------------------------------------------------
def get_facts(self) -> Dict:
def get_facts(self) -> dict:
"""Return general device facts.
Combines ``show sysinfo`` and ``show version``.
@@ -467,7 +464,7 @@ class NetgearSmartDriver(ConfigLifecycleMixin, SnmpAgentMixin, SwitchDriver):
"interface_list": interface_list,
}
def _get_facts_v7(self) -> Dict:
def _get_facts_v7(self) -> dict:
"""``get_facts()`` implementation for the "v7" Cisco-like CLI.
Example ``show info`` output::
@@ -526,7 +523,7 @@ class NetgearSmartDriver(ConfigLifecycleMixin, SnmpAgentMixin, SwitchDriver):
self._port_count_v7 = int(m.group(1)) if m else 0
return self._port_count_v7
def _get_interface_list_v7(self) -> List[str]:
def _get_interface_list_v7(self) -> list[str]:
n = self._get_port_count_v7()
interfaces = [f"g{i}" for i in range(1, n + 1)]
try:
@@ -540,10 +537,10 @@ class NetgearSmartDriver(ConfigLifecycleMixin, SnmpAgentMixin, SwitchDriver):
interfaces.append(f"lag{m.group(1)}")
return interfaces
def _get_interface_list(self) -> List[str]:
def _get_interface_list(self) -> list[str]:
"""Return sorted list of slot/port interface names from ``show port all``."""
output = self._send_command("show port all")
interfaces: List[str] = []
interfaces: list[str] = []
for line in output.splitlines():
# Physical ports: "0/1 ..." and LAG channels: "ch1 ..."
m = re.match(r"^\s*(\d+/\d+|ch\d+)\s+", line)
@@ -554,7 +551,7 @@ class NetgearSmartDriver(ConfigLifecycleMixin, SnmpAgentMixin, SwitchDriver):
key=lambda s: [int(x) if x.isdigit() else x for x in re.split(r"(\d+)", s)],
)
def get_interfaces(self) -> Dict[str, Dict]:
def get_interfaces(self) -> dict[str, dict]:
"""Return per-interface details.
Combines ``show port all`` (link status, speed) with
@@ -583,7 +580,7 @@ class NetgearSmartDriver(ConfigLifecycleMixin, SnmpAgentMixin, SwitchDriver):
self._add_lag_info(interfaces)
return interfaces
def _add_lag_info(self, interfaces: Dict[str, Dict]) -> None:
def _add_lag_info(self, interfaces: dict[str, dict]) -> None:
"""Enrich ``interfaces`` with LAG/trunk membership from ``show port-channel all``.
FASTPATH-based switches (NETGEAR M4xxx / GSxxxTxx Smart Managed Pro)
@@ -618,7 +615,7 @@ class NetgearSmartDriver(ConfigLifecycleMixin, SnmpAgentMixin, SwitchDriver):
interfaces[lag]["lag_members"] = members
interfaces[lag]["lag_mode"] = lag_mode
def _get_interfaces_v7(self) -> Dict[str, Dict]:
def _get_interfaces_v7(self) -> dict[str, dict]:
"""``get_interfaces()`` implementation for the "v7" Cisco-like CLI.
Example ``show interfaces GigabitEthernet 1-10 status`` output::
@@ -629,7 +626,7 @@ class NetgearSmartDriver(ConfigLifecycleMixin, SnmpAgentMixin, SwitchDriver):
g2 Notconnct 1 -- -- Copper
"""
n = self._get_port_count_v7()
interfaces: Dict[str, Dict] = {}
interfaces: dict[str, dict] = {}
if n:
status_out = self._send_command(f"show interfaces GigabitEthernet 1-{n} status")
interfaces.update(self._parse_interface_status_v7(status_out))
@@ -638,8 +635,8 @@ class NetgearSmartDriver(ConfigLifecycleMixin, SnmpAgentMixin, SwitchDriver):
return interfaces
@staticmethod
def _parse_interface_status_v7(output: str) -> Dict[str, Dict]:
interfaces: Dict[str, Dict] = {}
def _parse_interface_status_v7(output: str) -> dict[str, dict]:
interfaces: dict[str, dict] = {}
for line in output.splitlines():
line_s = line.rstrip()
if not line_s:
@@ -669,7 +666,7 @@ class NetgearSmartDriver(ConfigLifecycleMixin, SnmpAgentMixin, SwitchDriver):
}
return interfaces
def _add_lag_info_v7(self, interfaces: Dict[str, Dict]) -> None:
def _add_lag_info_v7(self, interfaces: dict[str, dict]) -> None:
"""Enrich ``interfaces`` with LAG membership from "show lag".
Example output::
@@ -711,9 +708,9 @@ class NetgearSmartDriver(ConfigLifecycleMixin, SnmpAgentMixin, SwitchDriver):
"lag_mode": "lacp" if "lacp" in type_s.lower() else "trunk",
}
def _parse_interfaces(self, port_output: str, intf_output: str = "") -> Dict[str, Dict]:
def _parse_interfaces(self, port_output: str, intf_output: str = "") -> dict[str, dict]:
"""Parse ``show port all`` and ``show interface all`` into NAPALM format."""
interfaces: Dict[str, Dict] = {}
interfaces: dict[str, dict] = {}
# --- parse 'show port all' ---
# Skip header lines until we hit a separator row, then parse data rows.
@@ -766,7 +763,7 @@ class NetgearSmartDriver(ConfigLifecycleMixin, SnmpAgentMixin, SwitchDriver):
}
# --- parse 'show interface all' (block format) ---
current_port: Optional[str] = None
current_port: str | None = None
for line in intf_output.splitlines():
# Block header: "Interface................................ 0/1"
m = re.match(r"^\s*Interface[.:\s]+(\d+/\d+|ch\d+)", line, re.I)
@@ -790,7 +787,7 @@ class NetgearSmartDriver(ConfigLifecycleMixin, SnmpAgentMixin, SwitchDriver):
return interfaces
def get_interfaces_ip(self) -> Dict[str, Dict]:
def get_interfaces_ip(self) -> dict[str, dict]:
"""Return all configured IP addresses grouped by interface.
Parses ``show ip interface`` output. On Smart Managed switches
@@ -803,7 +800,7 @@ class NetgearSmartDriver(ConfigLifecycleMixin, SnmpAgentMixin, SwitchDriver):
Default Gateway.................................. 192.168.0.1
"""
output = self._send_command("show ip interface")
interfaces_ip: Dict[str, Dict] = {}
interfaces_ip: dict[str, dict] = {}
# Smart Managed switches expose a single management IP block;
# some firmwares add an "Interface" header for the routing VLAN.
@@ -873,7 +870,7 @@ class NetgearSmartDriver(ConfigLifecycleMixin, SnmpAgentMixin, SwitchDriver):
full: bool = False,
sanitized: bool = False,
format: str = "text",
) -> Dict[str, str]:
) -> dict[str, str]:
"""Return running and/or startup configuration.
Netgear Smart Managed does not support a candidate config;
@@ -896,7 +893,7 @@ class NetgearSmartDriver(ConfigLifecycleMixin, SnmpAgentMixin, SwitchDriver):
return configs
def get_arp_table(self, vrf: str = "") -> List[Dict]:
def get_arp_table(self, vrf: str = "") -> list[dict]:
"""Return the ARP table.
Example ``show arp`` output::
@@ -949,7 +946,7 @@ class NetgearSmartDriver(ConfigLifecycleMixin, SnmpAgentMixin, SwitchDriver):
return arp_table
def get_mac_address_table(self) -> List[Dict]:
def get_mac_address_table(self) -> list[dict]:
"""Return the MAC address table.
Example ``show mac-addr-table`` output::
@@ -1008,7 +1005,7 @@ class NetgearSmartDriver(ConfigLifecycleMixin, SnmpAgentMixin, SwitchDriver):
return mac_table
def _get_mac_address_table_v7(self) -> List[Dict]:
def _get_mac_address_table_v7(self) -> list[dict]:
"""``get_mac_address_table()`` implementation for the "v7" Cisco-like CLI.
Example ``show mac address-table`` output::
@@ -1055,7 +1052,7 @@ class NetgearSmartDriver(ConfigLifecycleMixin, SnmpAgentMixin, SwitchDriver):
return mac_table
def get_lldp_neighbors(self) -> Dict[str, List[Dict]]:
def get_lldp_neighbors(self) -> dict[str, list[dict]]:
"""Return a dict of LLDP neighbors keyed by local port.
Example ``show lldp remote-device all`` output::
@@ -1065,7 +1062,7 @@ class NetgearSmartDriver(ConfigLifecycleMixin, SnmpAgentMixin, SwitchDriver):
--------- ----- -------------------- --------------- ---------------
0/1 1 00:1a:2b:3c:4d:5e 0/1 router-1
"""
neighbors: Dict[str, List[Dict]] = {}
neighbors: dict[str, list[dict]] = {}
for row in self._get_lldp_table():
entry = {"hostname": row["system_name"], "port": row["port_id"]}
chassis_id = row.get("remote_chassis_id", "")
@@ -1074,12 +1071,12 @@ class NetgearSmartDriver(ConfigLifecycleMixin, SnmpAgentMixin, SwitchDriver):
neighbors.setdefault(row["local_port"], []).append(entry)
return neighbors
def _get_lldp_table(self) -> List[Dict]:
def _get_lldp_table(self) -> list[dict]:
"""Parse the LLDP neighbour table into a list of row dicts."""
if self._detect_cli_v7():
return self._get_lldp_table_v7()
output = self._send_command("show lldp remote-device all")
rows: List[Dict] = []
rows: list[dict] = []
in_table = False
for line in output.splitlines():
@@ -1111,16 +1108,16 @@ class NetgearSmartDriver(ConfigLifecycleMixin, SnmpAgentMixin, SwitchDriver):
return rows
def _get_lldp_table_v7(self) -> List[Dict]:
def _get_lldp_table_v7(self) -> list[dict]:
"""Parse ``show lldp neighbor`` on v7 firmware.
The v7 CLI has neither ``show lldp remote-device all`` nor a bare
``show lldp`` — both answer "Unknown command". Only the singular
``neighbor`` works, and it lays the table out differently::
Port | Device ID | Port ID | SysName | Capabilities | TTL
---- + ----------------- + ---------------- + ----------------- + -------------- + -----
g9 | 10:01:02:44:37:26 |10:01:02:44:37:28 | pve-eze.eze.local | Bridge | 97
Port | Device ID | Port ID | SysName | Capabilities | TTL
---- + ----------------- + ---------------- + ----------------- + -------------- + -----
g9 | 10:01:02:44:37:26 |10:01:02:44:37:28 | pve-eze.eze.local | Bridge | 97
Ports are named ``g9`` rather than ``0/1``, and the columns are
separated by ``|`` with irregular padding — splitting on whitespace
@@ -1132,7 +1129,7 @@ class NetgearSmartDriver(ConfigLifecycleMixin, SnmpAgentMixin, SwitchDriver):
except Exception: # pragma: no cover - transport failure
return []
rows: List[Dict] = []
rows: list[dict] = []
for line in output.splitlines():
if "|" not in line:
continue
@@ -1155,13 +1152,13 @@ class NetgearSmartDriver(ConfigLifecycleMixin, SnmpAgentMixin, SwitchDriver):
)
return rows
def get_lldp_neighbors_detail(self, interface: str = "") -> Dict[str, List[Dict]]:
def get_lldp_neighbors_detail(self, interface: str = "") -> dict[str, list[dict]]:
"""Return detailed LLDP neighbor information.
Fetches per-port detail via ``show lldp remote-device detail <port>``
and falls back to the summary table values for missing fields.
"""
details: Dict[str, List[Dict]] = {}
details: dict[str, list[dict]] = {}
for row in self._get_lldp_table():
if interface and row["local_port"] != interface:
@@ -1183,9 +1180,9 @@ class NetgearSmartDriver(ConfigLifecycleMixin, SnmpAgentMixin, SwitchDriver):
return details
@staticmethod
def _parse_lldp_detail(output: str) -> Dict:
def _parse_lldp_detail(output: str) -> dict:
"""Parse ``show lldp remote-device detail`` block output."""
result: Dict = {
result: dict = {
"parent_interface": "",
"remote_port": "",
"remote_port_description": "",
@@ -1229,7 +1226,7 @@ class NetgearSmartDriver(ConfigLifecycleMixin, SnmpAgentMixin, SwitchDriver):
return result
def get_vlans(self) -> Dict[str, Dict]:
def get_vlans(self) -> dict[str, dict]:
"""Return VLAN information.
Example ``show vlan`` output::
@@ -1243,8 +1240,8 @@ class NetgearSmartDriver(ConfigLifecycleMixin, SnmpAgentMixin, SwitchDriver):
return self._get_vlans_v7()
output = self._send_command("show vlan")
vlans: Dict[str, Dict] = {}
current_id: Optional[str] = None
vlans: dict[str, dict] = {}
current_id: str | None = None
in_table = False
for line in output.splitlines():
@@ -1271,7 +1268,7 @@ class NetgearSmartDriver(ConfigLifecycleMixin, SnmpAgentMixin, SwitchDriver):
return vlans
@staticmethod
def _parse_vlan_ports(ports_raw: str) -> List[str]:
def _parse_vlan_ports(ports_raw: str) -> list[str]:
"""Parse Netgear VLAN port string and expand ranges.
Input examples::
@@ -1279,7 +1276,7 @@ class NetgearSmartDriver(ConfigLifecycleMixin, SnmpAgentMixin, SwitchDriver):
"0/1, 0/3, 0/5" → ["0/1", "0/3", "0/5"]
"0/1-4" → ["0/1", "0/2", "0/3", "0/4"]
"""
interfaces: List[str] = []
interfaces: list[str] = []
for token in ports_raw.split(","):
token = token.strip()
if not token:
@@ -1295,7 +1292,7 @@ class NetgearSmartDriver(ConfigLifecycleMixin, SnmpAgentMixin, SwitchDriver):
interfaces.append(token)
return interfaces
def _get_vlans_v7(self) -> Dict[str, Dict]:
def _get_vlans_v7(self) -> dict[str, dict]:
"""``get_vlans()`` implementation for the "v7" Cisco-like CLI.
Example ``show vlan 1-4094`` output::
@@ -1306,7 +1303,7 @@ class NetgearSmartDriver(ConfigLifecycleMixin, SnmpAgentMixin, SwitchDriver):
10 MGMT Static --- g1-8
"""
output = self._send_paged_command("show vlan 1-4094")
vlans: Dict[str, Dict] = {}
vlans: dict[str, dict] = {}
for line in output.splitlines():
line_s = line.strip()
if not line_s or "|" not in line_s:
@@ -1341,7 +1338,7 @@ class NetgearSmartDriver(ConfigLifecycleMixin, SnmpAgentMixin, SwitchDriver):
def _get_running_config(self) -> str:
return self._send_command("show running-config")
def commit_config(self, message: str = "", revert_in: Optional[int] = None) -> None:
def commit_config(self, message: str = "", revert_in: int | None = None) -> None:
"""Apply the staged candidate configuration to the device and persist it.
1. Saves current running config as rollback backup.
@@ -1361,7 +1358,7 @@ class NetgearSmartDriver(ConfigLifecycleMixin, SnmpAgentMixin, SwitchDriver):
self._backup_config = self._send_command("show running-config")
errors: List[str] = []
errors: list[str] = []
try:
self._enter_config_mode()
errors = self._apply_config_lines(self._candidate_config)
@@ -1401,14 +1398,14 @@ class NetgearSmartDriver(ConfigLifecycleMixin, SnmpAgentMixin, SwitchDriver):
self._backup_config = None
@staticmethod
def _parse_config_blocks(config_text: str) -> Dict[str, List[str]]:
def _parse_config_blocks(config_text: str) -> dict[str, list[str]]:
"""Parse a running-config into ``context → [command lines]`` mapping.
The special key ``"__global__"`` holds top-level commands.
Sub-mode blocks (``interface …``, ``vlan …``) are stored under the
opening line that introduces them.
"""
blocks: Dict[str, List[str]] = {"__global__": []}
blocks: dict[str, list[str]] = {"__global__": []}
ctx = "__global__"
for line in config_text.splitlines():
@@ -1436,7 +1433,7 @@ class NetgearSmartDriver(ConfigLifecycleMixin, SnmpAgentMixin, SwitchDriver):
return blocks
@staticmethod
def _negate_command(cmd: str) -> Optional[str]:
def _negate_command(cmd: str) -> str | None:
"""Return the ``no`` form of *cmd*, or ``None`` if not known."""
for kw in (
"description",
@@ -1450,12 +1447,12 @@ class NetgearSmartDriver(ConfigLifecycleMixin, SnmpAgentMixin, SwitchDriver):
return None
@staticmethod
def _diff_to_commands(backup: str, current: str) -> List[str]:
def _diff_to_commands(backup: str, current: str) -> list[str]:
"""Generate the minimal set of config commands to revert *current* to *backup*."""
backup_blocks = NetgearSmartDriver._parse_config_blocks(backup)
current_blocks = NetgearSmartDriver._parse_config_blocks(current)
cmds: List[str] = []
cmds: list[str] = []
all_ctxs = set(backup_blocks.keys()) | set(current_blocks.keys())
for ctx in sorted(all_ctxs):
@@ -1488,7 +1485,7 @@ class NetgearSmartDriver(ConfigLifecycleMixin, SnmpAgentMixin, SwitchDriver):
# Additional NAPALM getters
# ------------------------------------------------------------------
def get_environment(self) -> Dict:
def get_environment(self) -> dict:
"""Return device environment data (CPU, memory).
Netgear Smart Managed switches do not expose fan, temperature, or
@@ -1526,7 +1523,7 @@ class NetgearSmartDriver(ConfigLifecycleMixin, SnmpAgentMixin, SwitchDriver):
},
}
def get_interfaces_counters(self) -> Dict[str, Dict]:
def get_interfaces_counters(self) -> dict[str, dict]:
"""Return per-interface packet and byte counters.
Parses ``show interface counters`` (or ``show interface ethernet 0/1``).
@@ -1537,11 +1534,11 @@ class NetgearSmartDriver(ConfigLifecycleMixin, SnmpAgentMixin, SwitchDriver):
def _int(s: str) -> int:
return int(s.replace(",", "")) if s.strip().replace(",", "").isdigit() else 0
counters: Dict[str, Dict] = {}
current_port: Optional[str] = None
current: Optional[Dict] = None
counters: dict[str, dict] = {}
current_port: str | None = None
current: dict | None = None
_zero: Dict = {
_zero: dict = {
"tx_errors": 0,
"rx_errors": 0,
"tx_discards": 0,
@@ -1602,7 +1599,7 @@ class NetgearSmartDriver(ConfigLifecycleMixin, SnmpAgentMixin, SwitchDriver):
return counters
def get_users(self) -> Dict[str, Dict]:
def get_users(self) -> dict[str, dict]:
"""Return local user accounts.
Example ``show users`` output::
@@ -1612,7 +1609,7 @@ class NetgearSmartDriver(ConfigLifecycleMixin, SnmpAgentMixin, SwitchDriver):
admin Read/Write 5 Enabled
"""
output = self._send_command("show users")
users: Dict[str, Dict] = {}
users: dict[str, dict] = {}
in_table = False
for line in output.splitlines():
@@ -1632,11 +1629,12 @@ class NetgearSmartDriver(ConfigLifecycleMixin, SnmpAgentMixin, SwitchDriver):
username = parts[0]
role = parts[1].lower()
level = 15 if "write" in role or "admin" in role else 1
users[username] = {"level": level, "password": "", "sshkeys": []}
# NAPALM's get_users shape: the CLI never shows a password, so it is empty.
users[username] = {"level": level, "password": "", "sshkeys": []} # nosec B105
return users
def get_snmp_information(self) -> Dict:
def get_snmp_information(self) -> dict:
"""Return SNMP configuration.
Parses ``show sysinfo`` for contact/location/chassis-ID and
@@ -1648,7 +1646,7 @@ class NetgearSmartDriver(ConfigLifecycleMixin, SnmpAgentMixin, SwitchDriver):
mac = self._parse_key_value(sysinfo, "Burned In MAC Address").replace("-", ":").upper()
snmp_out = self._send_command("show snmp")
communities: Dict[str, Dict] = {}
communities: dict[str, dict] = {}
in_comm = False
for line in snmp_out.splitlines():
@@ -1675,10 +1673,10 @@ class NetgearSmartDriver(ConfigLifecycleMixin, SnmpAgentMixin, SwitchDriver):
"chassis_id": mac,
}
def get_ntp_servers(self) -> Dict[str, Dict]:
def get_ntp_servers(self) -> dict[str, dict]:
"""Return configured NTP/SNTP servers from ``show sntp server``."""
output = self._send_command(["show sntp server", "show ntp server"])
servers: Dict[str, Dict] = {}
servers: dict[str, dict] = {}
for line in output.splitlines():
line_s = line.strip()
@@ -1699,19 +1697,19 @@ class NetgearSmartDriver(ConfigLifecycleMixin, SnmpAgentMixin, SwitchDriver):
return servers
def get_ntp_peers(self) -> Dict[str, Dict]:
def get_ntp_peers(self) -> dict[str, dict]:
"""Return NTP peers (same as servers on Netgear Smart Managed)."""
return self.get_ntp_servers()
def get_ntp_stats(self) -> List[Dict]:
def get_ntp_stats(self) -> list[dict]:
"""Netgear Smart Managed CLI does not expose per-peer NTP statistics."""
return []
def get_optics(self) -> Dict:
def get_optics(self) -> dict:
"""SFP DDM/DOM data is not accessible via the Smart Managed CLI."""
return {}
def get_ipv6_neighbors_table(self) -> List[Dict]:
def get_ipv6_neighbors_table(self) -> list[dict]:
"""IPv6 ND table is not exposed via the Smart Managed CLI."""
return []
@@ -1720,14 +1718,14 @@ class NetgearSmartDriver(ConfigLifecycleMixin, SnmpAgentMixin, SwitchDriver):
destination: str = "",
protocol: str = "",
longer: bool = False,
) -> Dict[str, List[Dict]]:
) -> dict[str, list[dict]]:
"""Return routing table entries.
Smart Managed Pro switches support only a default gateway;
``show ip route`` is parsed where available.
"""
output = self._send_command("show ip route")
routes: Dict[str, List[Dict]] = {}
routes: dict[str, list[dict]] = {}
proto_map = {"c": "connected", "s": "static"}
for line in output.splitlines():
@@ -1783,7 +1781,7 @@ class NetgearSmartDriver(ConfigLifecycleMixin, SnmpAgentMixin, SwitchDriver):
count: int = 5,
vrf: str = "",
source_interface: str = "",
) -> Dict:
) -> dict:
"""Ping *destination* from the device.
Uses ``ping <dst> count <N>``. Source IP and packet size selection
@@ -1832,13 +1830,13 @@ class NetgearSmartDriver(ConfigLifecycleMixin, SnmpAgentMixin, SwitchDriver):
def cli(
self,
commands: List[str],
commands: list[str],
encoding: str = "text",
) -> Dict[str, Union[str, Dict[str, Any]]]:
) -> dict[str, str | dict[str, Any]]:
"""Execute a list of CLI commands and return their raw output."""
if encoding != "text":
raise NotImplementedError(f"Encoding '{encoding}' is not supported by this driver.")
result: Dict[str, Union[str, Dict[str, Any]]] = {}
result: dict[str, str | dict[str, Any]] = {}
for cmd in commands:
result[cmd] = self._send_command(cmd)
return result
@@ -1887,10 +1885,10 @@ class NetgearSmartDriver(ConfigLifecycleMixin, SnmpAgentMixin, SwitchDriver):
# SNMP agent (napalm-device-types SnmpAgentMixin): the caller's values
# ------------------------------------------------------------------
def _install_snmp_agent(self) -> Dict[str, Any]:
def _install_snmp_agent(self) -> dict[str, Any]:
return {"ok": True, "output": "the SNMP agent is built in"}
def _write_snmp_agent_config(self, settings: Dict[str, Any]) -> Dict[str, Any]:
def _write_snmp_agent_config(self, settings: dict[str, Any]) -> dict[str, Any]:
if self._detect_cli_v7():
# The "v7" Cisco-like CLI (e.g. GS110TPv3 7.x) has no "snmp-server"
# commands at all: SNMP is set in the web UI on this firmware.
@@ -1919,16 +1917,16 @@ class NetgearSmartDriver(ConfigLifecycleMixin, SnmpAgentMixin, SwitchDriver):
return {"ok": True, "output": "community (read-only), contact and location set"}
@staticmethod
def _allow_snmp_sources(settings: Dict[str, Any]) -> Dict[str, Any]:
def _allow_snmp_sources(settings: dict[str, Any]) -> dict[str, Any]:
sources = ", ".join(settings.get("allowed_sources", [])) or "none given"
note = f"no host firewall on the switch; sources not enforced ({sources})"
return {"ok": True, "output": note}
def _start_snmp_agent(self) -> Dict[str, Any]:
def _start_snmp_agent(self) -> dict[str, Any]:
self._save_config()
return {"ok": True, "output": "configuration saved"}
def configure_snmp_agent(self, settings: Dict[str, Any]) -> Dict[str, Any]:
def configure_snmp_agent(self, settings: dict[str, Any]) -> dict[str, Any]:
"""The template's steps, then a check that the switch lists the community."""
result = SnmpAgentMixin.configure_snmp_agent(self, settings)
if not result["success"]:
@@ -1943,7 +1941,7 @@ class NetgearSmartDriver(ConfigLifecycleMixin, SnmpAgentMixin, SwitchDriver):
# PoE
# ------------------------------------------------------------------
def get_poe_status(self) -> Dict[str, Dict]:
def get_poe_status(self) -> dict[str, dict]:
"""Return PoE configuration per port from ``show power inline``.
Example output::
@@ -1967,7 +1965,7 @@ class NetgearSmartDriver(ConfigLifecycleMixin, SnmpAgentMixin, SwitchDriver):
elif "class" in mode:
allocation_method = "PPAM_CLASS"
result: Dict[str, Dict] = {}
result: dict[str, dict] = {}
in_table = False
for line in output.splitlines():
line_s = line.strip()
@@ -1997,7 +1995,7 @@ class NetgearSmartDriver(ConfigLifecycleMixin, SnmpAgentMixin, SwitchDriver):
return result
def set_poe(self, interface: str, config: Dict) -> None:
def set_poe(self, interface: str, config: dict) -> None:
"""Update PoE configuration for a single port.
Uses the ``power inline`` interface sub-commands (ProSafe CLI).
@@ -2032,7 +2030,7 @@ class NetgearSmartDriver(ConfigLifecycleMixin, SnmpAgentMixin, SwitchDriver):
# LAG / trunk membership
# ------------------------------------------------------------------
def set_lag_members(self, lag_name: str, members: List[str]) -> None:
def set_lag_members(self, lag_name: str, members: list[str]) -> None:
"""Set the full member-port list of a LAG/trunk group (e.g. ``"ch1"``).
Diffs ``members`` against the LAG's current members (as reported by
@@ -2052,7 +2050,7 @@ class NetgearSmartDriver(ConfigLifecycleMixin, SnmpAgentMixin, SwitchDriver):
if not to_remove and not to_add:
return
lines: List[str] = []
lines: list[str] = []
for port in to_remove:
lines.append(f"interface {port}")
lines.append(f"deleteport {lag_name}")
napalm_netgear/py.typed
View File
Whitespace-only changes.
pyproject.toml
+52 -18
View File
@@ -1,57 +1,91 @@
[build-system]
requires = ["setuptools>=64", "wheel"]
requires = ["setuptools>=77"]
build-backend = "setuptools.build_meta"
[project]
name = "napalm-netgear"
version = "0.5.0"
# "napalm-netgear" on PyPI belongs to someone else (STANDARDS.md §8).
name = "napalm-netgear-switches"
version = "0.6.0"
description = "NAPALM driver for Netgear switches (Plus/Smart)"
readme = "README.md"
license = { text = "Apache-2.0" }
requires-python = ">=3.8"
license = "Apache-2.0"
requires-python = ">=3.10"
authors = [
{ name = "Christian Manivong" },
]
classifiers = [
"Topic :: Utilities",
"License :: OSI Approved :: Apache Software License",
"Programming Language :: Python :: 3",
"Programming Language :: Python :: 3.8",
"Programming Language :: Python :: 3.9",
"Programming Language :: Python :: 3.10",
"Programming Language :: Python :: 3.11",
"Programming Language :: Python :: 3.12",
"Programming Language :: Python :: 3.13",
"Operating System :: POSIX :: Linux",
"Operating System :: MacOS",
]
dependencies = [
"napalm>=4.0.0",
"napalm-device-types>=4.19.0",
"napalm-device-types>=4.21.1",
"netmiko>=4.0.0",
"paramiko>=5.0.0", # CVE-2026-44405
"netaddr",
"netaddr>=0.7.19",
"py-netgear-plus>=0.6.4",
]
[project.optional-dependencies]
dev = [
"pytest",
"pytest-cov",
"black",
"ruff",
]
dev = ["pytest>=8", "pytest-cov>=5", "mypy>=1.11", "ruff>=0.6"]
[project.entry-points."napalm.drivers"]
netgear_smart = "napalm_netgear:NetgearSmartDriver"
netgear_plus = "napalm_netgear:NetgearPlusDriver"
[project.urls]
Repository = "https://github.com/napalm-automation-community/napalm-netgear"
Repository = "https://git.netork.io/NAPALM/napalm-netgear"
Issues = "https://git.netork.io/NAPALM/napalm-netgear/issues"
Changelog = "https://git.netork.io/NAPALM/napalm-netgear/src/branch/main/CHANGELOG.md"
[tool.napalm-standard]
# The distribution name differs from the import package.
package = "napalm_netgear"
[tool.setuptools.packages.find]
where = ["."]
include = ["napalm_netgear*"]
[tool.setuptools.package-data]
napalm_netgear = ["py.typed"]
[tool.ruff]
line-length = 100
target-version = "py38"
target-version = "py310"
[tool.ruff.lint]
select = ["E", "F", "I", "UP", "B"]
[tool.ruff.lint.per-file-ignores]
# Recorded device output keeps its lines.
"tests/**" = ["E501"]
[tool.mypy]
python_version = "3.10"
strict = true
# napalm itself is untyped, so every driver subclasses Any.
disallow_subclassing_any = false
[[tool.mypy.overrides]]
module = ["napalm.*", "netaddr.*", "py_netgear_plus.*"]
ignore_missing_imports = true
[[tool.mypy.overrides]]
# Not yet under strict mypy (quality baseline); may only shrink.
module = ["napalm_netgear.netgear_plus", "napalm_netgear.netgear_smart"]
ignore_errors = true
[tool.pytest.ini_options]
testpaths = ["tests"]
[tool.coverage.run]
source = ["napalm_netgear"]
[tool.coverage.report]
fail_under = 48
tests/quality_baseline.json
+40
View File
@@ -0,0 +1,40 @@
{
"coverage_floor": 48.0,
"function_complexity": {
"napalm_netgear/netgear_plus.py::NetgearPlusDriver.get_vlans_detail": 14,
"napalm_netgear/netgear_smart.py::NetgearSmartDriver._add_lag_info_v7": 13,
"napalm_netgear/netgear_smart.py::NetgearSmartDriver._get_lldp_table": 12,
"napalm_netgear/netgear_smart.py::NetgearSmartDriver._get_lldp_table_v7": 11,
"napalm_netgear/netgear_smart.py::NetgearSmartDriver._get_vlans_v7": 11,
"napalm_netgear/netgear_smart.py::NetgearSmartDriver._parse_interfaces": 19,
"napalm_netgear/netgear_smart.py::NetgearSmartDriver._parse_lldp_detail": 21,
"napalm_netgear/netgear_smart.py::NetgearSmartDriver.get_interfaces_counters": 12,
"napalm_netgear/netgear_smart.py::NetgearSmartDriver.get_poe_status": 11,
"napalm_netgear/netgear_smart.py::NetgearSmartDriver.get_route_to": 12,
"napalm_netgear/netgear_smart.py::NetgearSmartDriver.get_snmp_config": 16,
"napalm_netgear/netgear_smart.py::NetgearSmartDriver.get_snmp_information": 11
},
"function_lines": {
"napalm_netgear/netgear_plus.py::NetgearPlusDriver.get_vlans_detail": 67,
"napalm_netgear/netgear_smart.py::NetgearSmartDriver._parse_interfaces": 78,
"napalm_netgear/netgear_smart.py::NetgearSmartDriver.get_arp_table": 52,
"napalm_netgear/netgear_smart.py::NetgearSmartDriver.get_facts": 64,
"napalm_netgear/netgear_smart.py::NetgearSmartDriver.get_interfaces_counters": 75,
"napalm_netgear/netgear_smart.py::NetgearSmartDriver.get_interfaces_ip": 53,
"napalm_netgear/netgear_smart.py::NetgearSmartDriver.get_mac_address_table": 58,
"napalm_netgear/netgear_smart.py::NetgearSmartDriver.get_poe_status": 53,
"napalm_netgear/netgear_smart.py::NetgearSmartDriver.get_route_to": 57,
"napalm_netgear/netgear_smart.py::NetgearSmartDriver.ping": 56
},
"mypy_relaxed": [
"napalm_netgear.netgear_plus",
"napalm_netgear.netgear_smart"
],
"python_file_lines": {
"napalm_netgear/netgear_smart.py": 2070
},
"untyped_public": [
"napalm_netgear/netgear_plus.py::NetgearPlusDriver.get_snmp_config",
"napalm_netgear/netgear_smart.py::NetgearSmartDriver.get_snmp_config"
]
}
tests/unit/test_driver.py
+3 -3
View File
@@ -1,10 +1,10 @@
"""Unit tests for NetgearSmartDriver — no real device required."""
import pytest
from unittest.mock import MagicMock, patch
from napalm_netgear.netgear_smart import NetgearSmartDriver
import pytest
from napalm_netgear.netgear_smart import NetgearSmartDriver
# ---------------------------------------------------------------------------
# Fixtures
@@ -745,7 +745,7 @@ GS110TPv3#
"""
class TestGetLldpNeighbors:
class TestGetLldpNeighborsV7:
"""`show lldp neighbor` — the switch does speak LLDP, the driver did not.
Without this the map has nothing to draw for the site this switch serves:
tests/unit/test_ssh_host_key.py
+4 -2
View File
@@ -24,7 +24,9 @@ HANDLER = "napalm_netgear.netgear_smart.ConnectHandler"
def _device() -> MagicMock:
device = MagicMock()
device.remote_conn_pre.get_transport.return_value.get_remote_server_key.return_value = SERVER_KEY
device.remote_conn_pre.get_transport.return_value.get_remote_server_key.return_value = (
SERVER_KEY
)
return device
@@ -74,7 +76,7 @@ def test_a_changed_key_is_refused():
try:
raise paramiko.BadHostKeyException("10.0.0.8", SERVER_KEY, OTHER_KEY)
except paramiko.SSHException as exc: # what netmiko does with it
raise NetmikoTimeoutException(f"A paramiko SSHException occurred: {exc}")
raise NetmikoTimeoutException(f"A paramiko SSHException occurred: {exc}") from exc
driver = _driver(seen, SshHostKey.of(OTHER_KEY))
with patch(HANDLER, side_effect=refuse):