NAPALM/napalm-hpe-aruba-procurve
1
0 Fork 0
Code Issues 6 Pull Requests Actions Packages Projects Releases 1 Wiki Activity

Merge pull request 'chore: adopt the NAPALM driver standard' (#19) from chore/driver-standard into main
driver / standard (push) Successful in 14s
Details
driver / lint (push) Successful in 12s
Details
driver / types (push) Successful in 47s
Details
driver / test (3.10) (push) Successful in 46s
Details
driver / test (3.11) (push) Successful in 44s
Details
driver / test (3.12) (push) Successful in 45s
Details
driver / test (3.13) (push) Successful in 37s
Details
driver / lowest (push) Successful in 23s
Details
driver / security (push) Successful in 46s
Details
driver / build (push) Successful in 22s
Details
driver / publish (push) Successful in 25s
Details
CI / driver (push) Successful in 55m51s
Details

Browse Source
This commit was merged in pull request #19.
This commit is contained in:
christianmanivong committed 2026-10-23 09:46:46 +00:00
2 parents 739283e76b 1a7cf93709
commit b842aa5763
14 files changed
+636 -321
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/*
.gitignore
+4
View File
@@ -4,3 +4,7 @@ __pycache__/
.venv/
build/
dist/
.coverage
.pytest_cache/
.mypy_cache/
.ruff_cache/
CHANGELOG.md
+52
View File
@@ -0,0 +1,52 @@
# 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.
### Changed
- 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 now requires napalm-device-types 4.21.1.
- Exceptions the driver deliberately ignores (logging out of the REST API, the extra
port lookup, per-port details, closing a session) are now logged at debug level.
## [0.5.0] - 2026-10-18
### Changed
- The SSH host key is verified: with a known key in `optional_args["ssh_host_keys"]`
any other key is refused before login, and every key seen is reported to
`optional_args["ssh_host_key_observer"]`. Without either argument the driver
connects as before. Requires napalm-device-types 4.19.0.
## [0.4.0] - 2026-10-17
### Added
- `configure_snmp_agent()`: community (read-only), contact and location from the
caller, saved and verified in `show snmp-server`.
### Removed
- **Breaking:** `fix_snmp` and `run_device_action`; `fix_snmp` set the community
`public`.
## [0.3.0] - 2026-10-16
### Changed
- The driver no longer collects over SNMP: `SNMP_HEALTH_PROFILE` declares where the
switch publishes its health values. The memory value is read as free memory, as
the switch reports it, and interface counters come from the 64-bit ifXTable.
## [0.2.0] - 2026-05-29
### Added
- First version: ProCurve and Aruba switches over the REST API, SSH, SSH with legacy
key exchange or Telnet, chosen automatically.
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
+1 -1
View File
@@ -45,7 +45,7 @@ pip install napalm napalm-hpe-aruba-procurve
Or from source:
```bash
git clone https://github.com/chrismanivong/napalm-hpe-aruba-procurve
git clone https://git.netork.io/NAPALM/napalm-hpe-aruba-procurve
pip install -e napalm-hpe-aruba-procurve/
```
napalm_procurve/api_client.py
+60 -53
View File
@@ -1,4 +1,3 @@
# -*- coding: utf-8 -*-
# Licensed under the Apache License, Version 2.0
"""REST API client for HPE/Aruba ProCurve switches (ArubaOS REST interface).
@@ -8,13 +7,12 @@ Used automatically by ProcurveDriver when the REST API is reachable.
"""
import logging
from typing import Any, Dict, Optional, Tuple
from typing import Any
import requests
import urllib3
from napalm.base import helpers as napalm_helpers
from napalm.base.exceptions import ConnectionException, ConnectAuthError
from napalm.base.exceptions import ConnectAuthError, ConnectionException
from napalm_device_types import add_lag_interfaces
logger = logging.getLogger("napalm_procurve.api")
@@ -37,7 +35,7 @@ class ProcurveApiClient:
password: str,
timeout: int = 60,
ssl_verify: bool = False,
api_version: Optional[str] = None,
api_version: str | None = None,
) -> None:
self.hostname = hostname
self.username = username
@@ -47,7 +45,7 @@ class ProcurveApiClient:
self._api_version = api_version # None = auto-detect
self._proto: str = "https"
self._base_url: str = ""
self._session: Optional[requests.Session] = None
self._session: requests.Session | None = None
# ------------------------------------------------------------------
# Detection
@@ -59,7 +57,7 @@ class ProcurveApiClient:
hostname: str,
timeout: int = 5,
ssl_verify: bool = False,
) -> Tuple[Optional[str], Optional[str]]:
) -> tuple[str | None, str | None]:
"""Probe the device for an ArubaOS REST API.
Returns ``(api_version, proto)`` on success, ``(None, None)`` if the
@@ -86,7 +84,10 @@ class ProcurveApiClient:
if resp.status_code in (200, 301, 302, 303, 401, 403, 405):
logger.debug(
"API detected at %s (HTTP %s, %s %s)",
hostname, resp.status_code, proto, ver,
hostname,
resp.status_code,
proto,
ver,
)
return ver, proto
except requests.exceptions.SSLError:
@@ -115,9 +116,7 @@ class ProcurveApiClient:
self._session = requests.Session()
self._session.verify = self.ssl_verify
self._session.headers.update(
{"Content-Type": "application/json", "Connection": "close"}
)
self._session.headers.update({"Content-Type": "application/json", "Connection": "close"})
url = self._base_url + "login-sessions"
resp = self._session.post(
@@ -148,11 +147,9 @@ class ProcurveApiClient:
"""Log out from the REST API session."""
if self._session and self._base_url:
try:
self._session.delete(
self._base_url + "login-sessions", timeout=self.timeout
)
self._session.delete(self._base_url + "login-sessions", timeout=self.timeout)
except Exception:
pass
logger.debug("REST API logout: ignored", exc_info=True)
if self._session:
self._session.close()
self._session = None
@@ -162,9 +159,7 @@ class ProcurveApiClient:
if not self._session:
return False
try:
resp = self._session.get(
self._base_url + "system/status", timeout=5
)
resp = self._session.get(self._base_url + "system/status", timeout=5)
return resp.status_code == 200
except Exception:
return False
@@ -173,7 +168,7 @@ class ProcurveApiClient:
# HTTP helpers
# ------------------------------------------------------------------
def get(self, endpoint: str) -> Dict[str, Any]:
def get(self, endpoint: str) -> dict[str, Any]:
"""GET ``{base_url}{endpoint}`` and return parsed JSON (or empty dict)."""
url = self._base_url + endpoint
try:
@@ -183,7 +178,9 @@ class ProcurveApiClient:
if resp.status_code == 404:
logger.debug("GET %s returned HTTP 404", url)
else:
logger.warning("GET %s returned HTTP %s: %s", url, resp.status_code, resp.text[:200])
logger.warning(
"GET %s returned HTTP %s: %s", url, resp.status_code, resp.text[:200]
)
except Exception as exc:
logger.warning("GET %s error: %s", url, exc)
return {}
@@ -204,7 +201,7 @@ class ProcurveApiClient:
# NAPALM data getters (API-backed)
# ------------------------------------------------------------------
def get_facts(self) -> Dict:
def get_facts(self) -> dict:
"""Return NAPALM facts from the REST API."""
system = self.get("system/status")
if not system:
@@ -230,6 +227,7 @@ class ProcurveApiClient:
model_raw = system.get("product_model", "")
# Extract J-code part numbers (e.g. J9777A) from the model string
import re as _re
_PN_RE = _re.compile(r"\b(J\d{4}[A-Z]{1,2})\b")
pn_match = _PN_RE.search(model_raw)
part_number = pn_match.group(1) if pn_match else ""
@@ -253,14 +251,14 @@ class ProcurveApiClient:
"interface_list": iface_list,
}
def get_interfaces(self) -> Dict[str, Dict]:
def get_interfaces(self) -> dict[str, dict]:
"""Return NAPALM interfaces from the REST API."""
ports_data = self.get("ports")
stats_data = self.get("port-statistics")
output: Dict[str, Dict] = {}
trunk_groups: Dict[str, list] = {}
trunk_modes: Dict[str, str] = {}
output: dict[str, dict] = {}
trunk_groups: dict[str, list] = {}
trunk_modes: dict[str, str] = {}
port_elements = list(ports_data.get("port_element", []))
@@ -274,16 +272,18 @@ class ProcurveApiClient:
for p in blade.get("data_ports", []):
pid = p.get("port_name", "")
if pid and pid not in seen_ids:
port_elements.append({
port_elements.append(
{
"id": pid,
"name": "",
"is_port_up": p.get("operStatus") == "OPER_UP",
"is_port_enabled": p.get("adminStatus") == "ADMIN_UP",
"trunk_group": "",
})
}
)
seen_ids.add(pid)
except Exception:
pass
logger.debug("system/status/switch for missing ports: ignored", exc_info=True)
for port in port_elements:
pid = port.get("id", "")
@@ -309,15 +309,18 @@ class ProcurveApiClient:
output[pid]["speed"] = float(stat.get("port_speed_mbps", 0))
# One row per LAG/trunk group alongside its member ports.
return add_lag_interfaces(output, {
return add_lag_interfaces(
output,
{
group: "lacp" if trunk_modes.get(group) == "PTT_LACP" else "trunk"
for group in trunk_groups
})
},
)
def get_interfaces_ip(self) -> Dict[str, Dict]:
def get_interfaces_ip(self) -> dict[str, dict]:
"""Return NAPALM interfaces IP from the REST API."""
vlans_data = self.get("vlans")
result: Dict[str, Dict] = {}
result: dict[str, dict] = {}
for vlan in vlans_data.get("vlan_element", []):
vlan_id = str(vlan.get("vlan_id", ""))
@@ -327,10 +330,12 @@ class ProcurveApiClient:
for entry in ipv4_data.get("ipv4_element", []):
addr = entry.get("address", "")
mask = entry.get("mask", "")
if not addr or addr == "0.0.0.0":
# An unset address reported by the switch, not a bind address.
if not addr or addr == "0.0.0.0": # nosec B104
continue
try:
import netaddr
plen = netaddr.IPNetwork(f"{addr}/{mask}").prefixlen
except Exception:
plen = 24
@@ -343,12 +348,12 @@ class ProcurveApiClient:
# PoE
# ------------------------------------------------------------------
def get_poe_ports(self) -> Dict[str, Dict]:
def get_poe_ports(self) -> dict[str, dict]:
"""Return PoE configuration for all ports, keyed by port id."""
data = self.get("poe/ports")
return {p["port_id"]: p for p in data.get("port_poe", []) if p.get("port_id")}
def set_port_poe(self, port_id: str, payload: Dict[str, Any]) -> requests.Response:
def set_port_poe(self, port_id: str, payload: dict[str, Any]) -> requests.Response:
"""PUT PoE configuration changes for a single port."""
return self.put(f"ports/{port_id}/poe", json=payload)
@@ -390,9 +395,9 @@ class ProcurveApiClient:
)
return table
def get_lldp_neighbors(self) -> Dict[str, list]:
def get_lldp_neighbors(self) -> dict[str, list]:
"""Return NAPALM LLDP neighbors from the REST API."""
result: Dict[str, list] = {}
result: dict[str, list] = {}
data = self.get("lldp/remote-device")
for nbr in data.get("lldp_remote_device_element", []):
local_port = nbr.get("local_port", "")
@@ -404,9 +409,9 @@ class ProcurveApiClient:
)
return result
def get_lldp_neighbors_detail(self) -> Dict[str, list]:
def get_lldp_neighbors_detail(self) -> dict[str, list]:
"""Return NAPALM LLDP neighbor details from the REST API."""
result: Dict[str, list] = {}
result: dict[str, list] = {}
data = self.get("lldp/remote-device")
for nbr in data.get("lldp_remote_device_element", []):
local_port = nbr.get("local_port", "")
@@ -424,7 +429,7 @@ class ProcurveApiClient:
)
return result
def get_config(self) -> Dict[str, str]:
def get_config(self) -> dict[str, str]:
"""Return running and startup configuration via REST API.
AOS-Switch returns the config as plain text, not JSON.
@@ -447,17 +452,17 @@ class ProcurveApiClient:
logger.warning("GET %s error: %s", url, exc)
return ""
def get_ntp_servers(self) -> Dict[str, Dict]:
def get_ntp_servers(self) -> dict[str, dict]:
"""Return NTP servers from REST API."""
data = self.get("ntp/server")
servers: Dict[str, Dict] = {}
servers: dict[str, dict] = {}
for entry in data.get("ntp_server_element", []):
addr = entry.get("server_address", "")
if addr:
servers[addr] = {}
return servers
def get_vlans(self) -> Dict[int, Dict]:
def get_vlans(self) -> dict[int, dict]:
"""Return VLAN information with proper tagged/untagged separation.
Fetches ``/vlans`` for VLAN names and ``/vlans-ports`` for port membership.
@@ -491,7 +496,7 @@ class ProcurveApiClient:
ports_data = self.get("vlans-ports")
# Build VLAN skeleton from /vlans
result: Dict[int, Dict] = {}
result: dict[int, dict] = {}
for vlan in vlans_data.get("vlan_element", []):
try:
vid = int(vlan["vlan_id"])
@@ -522,8 +527,7 @@ class ProcurveApiClient:
# Untagged variants: POM_UNTAGGED, POM_NATIVE_UNTAGGED, "Untagged"
is_untagged = any(
kw in mode_low
for kw in ("untagged", "native_untagged", "pom_untagged")
kw in mode_low for kw in ("untagged", "native_untagged", "pom_untagged")
) and "tagged" not in mode_low.replace("untagged", "")
# More precise: contains "untagged" anywhere → untagged
@@ -576,17 +580,17 @@ class ProcurveApiClient:
f"delete_vlan({vlan_id}): REST API returned HTTP {resp.status_code}"
)
def get_snmp_communities(self) -> Dict[str, Dict]:
def get_snmp_communities(self) -> dict[str, dict]:
"""Return configured SNMP communities from REST API."""
data = self.get("snmp-server/community")
result: Dict[str, Dict] = {}
result: dict[str, dict] = {}
for entry in data.get("snmp_community_element", []):
name = entry.get("community_name", "")
if name:
result[name] = {"access_type": entry.get("access_type", "")}
return result
def configure_snmp_community(self, community: str = "public") -> Tuple[bool, str]:
def configure_snmp_community(self, community: str = "public") -> tuple[bool, str]:
"""Ensure an SNMP read-only community exists via REST resource endpoint.
Returns (success, message).
@@ -595,10 +599,13 @@ class ProcurveApiClient:
if community in existing:
return True, f"SNMP community '{community}' already exists."
resp = self.post("snmp-server/community", json={
resp = self.post(
"snmp-server/community",
json={
"community_name": community,
"access_type": "MIB2",
})
},
)
if resp.ok:
return True, f"SNMP community '{community}' configured via REST API."
if resp.status_code == 409:
@@ -615,9 +622,9 @@ class ProcurveApiClient:
size: int = 100,
count: int = 5,
vrf: str = "",
) -> Dict:
) -> dict:
"""Execute ping via REST API (if supported) or return not-implemented."""
payload: Dict = {"destination": destination, "count": count, "timeout": timeout}
payload: dict = {"destination": destination, "count": count, "timeout": timeout}
if source:
payload["source"] = source
napalm_procurve/parsers.py
+64 -61
View File
@@ -1,4 +1,3 @@
# -*- coding: utf-8 -*-
# Licensed under the Apache License, Version 2.0
"""CLI output parsers for HPE/Aruba ProCurve switches.
@@ -9,13 +8,15 @@ backends.
"""
import re
from typing import Dict, List, Optional
#: get_users' password field: the switch never shows a password, so it stays empty.
_PASSWORD_NOT_SHOWN = "" # nosec B105
# ---------------------------------------------------------------------------
# MAC address helpers
# ---------------------------------------------------------------------------
def procurve_mac_to_std(mac: str) -> str:
"""Convert ProCurve MAC format ``a1b2c3-d4e5f6`` to ``a1:b2:c3:d4:e5:f6``."""
clean = mac.replace("-", "").replace(":", "").replace(".", "").lower()
@@ -38,7 +39,8 @@ def _kv(output: str, key: str) -> str:
# show system-information
# ---------------------------------------------------------------------------
def parse_system_info(output: str) -> Dict:
def parse_system_info(output: str) -> dict:
"""Parse ``show system-information`` output.
Returns a dict with keys: hostname, os_version, serial_number, model,
@@ -63,10 +65,8 @@ def parse_system_info(output: str) -> Dict:
# between them, e.g.:
# " Software revision : R.11.27 Base MAC Addr : a1b2c3-d4e5f6"
# We capture all k:v pairs on every line.
_KV_RE = re.compile(
r"([A-Za-z][A-Za-z0-9 ()/#-]{1,40}?)\s*:\s*(\S.*?)(?=\s{3,}[A-Za-z]|$)"
)
kv_map: Dict[str, str] = {}
_KV_RE = re.compile(r"([A-Za-z][A-Za-z0-9 ()/#-]{1,40}?)\s*:\s*(\S.*?)(?=\s{3,}[A-Za-z]|$)")
kv_map: dict[str, str] = {}
for raw_line in output.splitlines():
for m in _KV_RE.finditer(raw_line):
k = m.group(1).strip()
@@ -163,6 +163,7 @@ def _parse_procurve_uptime(uptime_str: str) -> float:
# show version
# ---------------------------------------------------------------------------
def parse_version(output: str) -> str:
"""Extract firmware version from ``show version`` or ``show system-information``.
@@ -245,13 +246,13 @@ _INTF_BRIEF_RE = re.compile(
)
def parse_interfaces_brief(output: str) -> Dict[str, Dict]:
def parse_interfaces_brief(output: str) -> dict[str, dict]:
"""Parse ``show interfaces brief`` tabular output.
Returns dict keyed by port name with sub-keys:
is_up, is_enabled, speed, duplex, description.
"""
interfaces: Dict[str, Dict] = {}
interfaces: dict[str, dict] = {}
for line in output.splitlines():
m = _INTF_BRIEF_RE.match(line)
if not m:
@@ -260,9 +261,7 @@ def parse_interfaces_brief(output: str) -> Dict[str, Dict]:
speed = 0.0
duplex = ""
# Mode examples: "1000FDx", "100HDx", "Unknown"; MDI is "Auto", "MDI", "MDIX", "NA"
sm = next(
filter(None, (re.match(r"(\d+)(FDx|HDx)?", t, re.I) for t in mode_or_mdi)), None
)
sm = next(filter(None, (re.match(r"(\d+)(FDx|HDx)?", t, re.I) for t in mode_or_mdi)), None)
if sm:
speed = float(sm.group(1))
duplex = "full" if (sm.group(2) or "").lower() == "fdx" else "half"
@@ -286,7 +285,8 @@ def parse_interfaces_brief(output: str) -> Dict[str, Dict]:
# show interfaces (per-port detail for MAC / description)
# ---------------------------------------------------------------------------
def parse_interface_detail(output: str, port: str) -> Dict:
def parse_interface_detail(output: str, port: str) -> dict:
"""Parse ``show interfaces <port>`` for MAC address and description."""
mac = procurve_mac_to_std(_kv(output, "MAC address"))
name = _kv(output, "Name") or _kv(output, "Description")
@@ -297,7 +297,8 @@ def parse_interface_detail(output: str, port: str) -> Dict:
# show ip / show ip interface
# ---------------------------------------------------------------------------
def parse_ip_interfaces(output: str) -> Dict[str, Dict]:
def parse_ip_interfaces(output: str) -> dict[str, dict]:
"""Parse ``show ip`` or ``show ip interface brief`` output.
Returns NAPALM-compatible dict::
@@ -314,7 +315,7 @@ def parse_ip_interfaces(output: str) -> Dict[str, Dict]:
IP Configuration for VLAN 99:
...
"""
result: Dict[str, Dict] = {}
result: dict[str, dict] = {}
current_vlan = ""
current_ip = ""
@@ -332,7 +333,8 @@ def parse_ip_interfaces(output: str) -> Dict[str, Dict]:
# Single VLAN output (no header): detect from "IP Address" line
if re.search(r"IP\s+Address\s*:", line, re.I):
ip = line.split(":", 1)[-1].strip()
if ip and ip not in ("0.0.0.0", ""):
# An unset address reported by the switch, not a bind address.
if ip and ip != "0.0.0.0": # nosec B104
current_ip = ip
if not current_vlan:
current_vlan = "VLAN1"
@@ -343,6 +345,7 @@ def parse_ip_interfaces(output: str) -> Dict[str, Dict]:
mask = line.split(":", 1)[-1].strip()
try:
import netaddr
prefix_len = netaddr.IPNetwork(f"{current_ip}/{mask}").prefixlen
except Exception:
prefix_len = 24
@@ -369,16 +372,14 @@ _ARP_RE = re.compile(
)
def parse_arp_table(output: str) -> List[Dict]:
def parse_arp_table(output: str) -> list[dict]:
"""Parse ``show arp`` output into NAPALM ARP table format."""
table = []
for line in output.splitlines():
m = _ARP_RE.search(line)
if not m:
continue
ip, mac, entry_type, port = (
m.group(1), m.group(2), m.group(3), m.group(4) or ""
)
ip, mac, _entry_type, port = (m.group(1), m.group(2), m.group(3), m.group(4) or "")
table.append(
{
"interface": port,
@@ -402,7 +403,7 @@ _MAC_RE = re.compile(
)
def parse_mac_table(output: str) -> List[Dict]:
def parse_mac_table(output: str) -> list[dict]:
"""Parse ``show mac-address`` into NAPALM MAC table format."""
table = []
for line in output.splitlines():
@@ -428,7 +429,8 @@ def parse_mac_table(output: str) -> List[Dict]:
# show lldp info remote-device
# ---------------------------------------------------------------------------
def parse_lldp_neighbors(output: str) -> Dict[str, List[Dict]]:
def parse_lldp_neighbors(output: str) -> dict[str, list[dict]]:
"""Parse ``show lldp info remote-device`` into NAPALM neighbors dict.
Output format::
@@ -437,7 +439,7 @@ def parse_lldp_neighbors(output: str) -> Dict[str, List[Dict]]:
--------- + ------------------- ------------ ------------ ...
1 | a1 b2 c3 d4 e5 f6 Gi 0/1 otherswitch ...
"""
neighbors: Dict[str, List[Dict]] = {}
neighbors: dict[str, list[dict]] = {}
in_table = False
for line in output.splitlines():
@@ -472,21 +474,19 @@ def parse_lldp_neighbors(output: str) -> Dict[str, List[Dict]]:
if re.match(r"^[0-9a-fA-F]{12}$", chassis):
chassis = ":".join(chassis[i : i + 2] for i in range(0, 12, 2))
neighbors.setdefault(local_port, []).append(
{"hostname": sys_name, "port": port_id}
)
neighbors.setdefault(local_port, []).append({"hostname": sys_name, "port": port_id})
return neighbors
def parse_lldp_neighbors_detail(output: str) -> Dict[str, List[Dict]]:
def parse_lldp_neighbors_detail(output: str) -> dict[str, list[dict]]:
"""Parse ``show lldp info remote-device detail`` output.
Returns NAPALM lldp_neighbors_detail format.
"""
neighbors: Dict[str, List[Dict]] = {}
neighbors: dict[str, list[dict]] = {}
current_port = ""
entry: Dict = {}
entry: dict = {}
def _save():
if current_port and entry:
@@ -516,9 +516,11 @@ def parse_lldp_neighbors_detail(output: str) -> Dict[str, List[Dict]]:
low = line.lower()
if "chassis id" in low:
val = line.split(":", 1)[-1].strip()
entry["remote_chassis_id"] = procurve_mac_to_std(
re.sub(r"\s+", "", val)
) if re.search(r"[0-9a-fA-F ]{14,}", val) else val
entry["remote_chassis_id"] = (
procurve_mac_to_std(re.sub(r"\s+", "", val))
if re.search(r"[0-9a-fA-F ]{14,}", val)
else val
)
elif "port id" in low and "descr" not in low:
entry["remote_port"] = line.split(":", 1)[-1].strip()
elif "port description" in low:
@@ -528,9 +530,7 @@ def parse_lldp_neighbors_detail(output: str) -> Dict[str, List[Dict]]:
elif "system description" in low:
entry["remote_system_description"] = line.split(":", 1)[-1].strip()
elif "system capabilities" in low and "enabled" not in low:
entry["remote_system_capab"] = _parse_capabilities(
line.split(":", 1)[-1].strip()
)
entry["remote_system_capab"] = _parse_capabilities(line.split(":", 1)[-1].strip())
elif "enabled capabilities" in low or ("system capabilities" in low and "enabled" in low):
entry["remote_system_enable_capab"] = _parse_capabilities(
line.split(":", 1)[-1].strip()
@@ -540,7 +540,7 @@ def parse_lldp_neighbors_detail(output: str) -> Dict[str, List[Dict]]:
return neighbors
def _parse_capabilities(cap_str: str) -> List[str]:
def _parse_capabilities(cap_str: str) -> list[str]:
"""Convert ProCurve capability abbreviations to NAPALM strings."""
_MAP = {
"B": "bridge",
@@ -566,12 +566,13 @@ def _parse_capabilities(cap_str: str) -> List[str]:
# show ntp associations / show timesync
# ---------------------------------------------------------------------------
def parse_ntp_servers(output: str) -> Dict[str, Dict]:
def parse_ntp_servers(output: str) -> dict[str, dict]:
"""Parse ``show ntp associations`` or ``show ntp status`` output.
Returns dict of {server_ip: {}} as per NAPALM get_ntp_servers.
"""
servers: Dict[str, Dict] = {}
servers: dict[str, dict] = {}
for line in output.splitlines():
# " 10.0.0.1 sys.peer 2 ..." or "NTP server: 10.0.0.1"
m = re.search(r"(\d{1,3}(?:\.\d{1,3}){3})", line)
@@ -588,9 +589,10 @@ def parse_ntp_servers(output: str) -> Dict[str, Dict]:
# show environment
# ---------------------------------------------------------------------------
def parse_environment(output: str) -> Dict:
def parse_environment(output: str) -> dict:
"""Parse ``show environment`` output into NAPALM environment dict."""
env: Dict = {
env: dict = {
"fans": {},
"temperature": {},
"power": {},
@@ -629,9 +631,7 @@ def parse_environment(output: str) -> Dict:
}
# Fan status
for m in re.finditer(
r"Fan\s*(\d+)\s*[:\s]+(\S+)", output, re.I
):
for m in re.finditer(r"Fan\s*(\d+)\s*[:\s]+(\S+)", output, re.I):
fan_id = f"fan_{m.group(1)}"
status = m.group(2).lower()
env["fans"][fan_id] = {"status": status in ("ok", "good", "normal", "operating")}
@@ -643,12 +643,13 @@ def parse_environment(output: str) -> Dict:
# show local-users
# ---------------------------------------------------------------------------
def parse_users(output: str) -> Dict[str, Dict]:
def parse_users(output: str) -> dict[str, dict]:
"""Parse ``show local-users`` or ``show users`` output.
Returns NAPALM users dict.
"""
users: Dict[str, Dict] = {}
users: dict[str, dict] = {}
for line in output.splitlines():
# " manager Operator Enabled ..." or " manager Manager ..."
m = re.match(
@@ -668,7 +669,7 @@ def parse_users(output: str) -> Dict[str, Dict]:
priv = 1
users[uname] = {
"level": priv,
"password": "",
"password": _PASSWORD_NOT_SHOWN,
"sshkeys": [],
}
return users
@@ -678,9 +679,10 @@ def parse_users(output: str) -> Dict[str, Dict]:
# show snmp-server / show snmp
# ---------------------------------------------------------------------------
def parse_snmp_info(output: str) -> Dict:
def parse_snmp_info(output: str) -> dict:
"""Parse ``show snmp-server`` output into NAPALM snmp_information dict."""
communities: Dict[str, Dict] = {}
communities: dict[str, dict] = {}
contact = _kv(output, "System Contact") or _kv(output, "Contact")
location = _kv(output, "System Location") or _kv(output, "Location")
chassis_id = ""
@@ -708,7 +710,8 @@ def parse_snmp_info(output: str) -> Dict:
# VLAN helpers
# ---------------------------------------------------------------------------
def expand_procurve_ports(port_str: str) -> List[str]:
def expand_procurve_ports(port_str: str) -> list[str]:
"""Expand a ProCurve port-list string into a flat list of port names.
Handles:
@@ -717,7 +720,7 @@ def expand_procurve_ports(port_str: str) -> List[str]:
- Trunk ports: ``Trk1,Trk2`` → ``["Trk1","Trk2"]``
- Mixed: ``1-3,Trk1,A1`` → ``["1","2","3","Trk1","A1"]``
"""
ports: List[str] = []
ports: list[str] = []
for token in port_str.split(","):
token = token.strip()
if not token:
@@ -746,7 +749,7 @@ def expand_procurve_ports(port_str: str) -> List[str]:
return ports
def parse_vlans_from_running_config(config_text: str) -> Dict[int, Dict]:
def parse_vlans_from_running_config(config_text: str) -> dict[int, dict]:
"""Parse VLAN blocks from ProCurve ``show running-config`` output.
Extracts ``name``, ``tagged``, and ``untagged`` port lists per VLAN.
@@ -777,8 +780,8 @@ def parse_vlans_from_running_config(config_text: str) -> Dict[int, Dict]:
"untagged": []},
}
"""
vlans: Dict[int, Dict] = {}
current_id: Optional[int] = None
vlans: dict[int, dict] = {}
current_id: int | None = None
for line in config_text.splitlines():
stripped = line.strip()
@@ -818,7 +821,7 @@ def parse_vlans_from_running_config(config_text: str) -> Dict[int, Dict]:
# Populate "interfaces" as union of tagged + untagged (NAPALM standard key)
for vdata in vlans.values():
seen: List[str] = []
seen: list[str] = []
for p in vdata["untagged"] + vdata["tagged"]:
if p not in seen:
seen.append(p)
@@ -827,7 +830,7 @@ def parse_vlans_from_running_config(config_text: str) -> Dict[int, Dict]:
return vlans
def parse_vlans_brief(output: str) -> Dict[int, str]:
def parse_vlans_brief(output: str) -> dict[int, str]:
"""Parse ``show vlans`` brief table for VLAN IDs and names only.
Used as a discovery step when running-config is unavailable.
@@ -841,7 +844,7 @@ def parse_vlans_brief(output: str) -> Dict[int, str]:
Returns ``{vlan_id: name}`` dict.
"""
vlans: Dict[int, str] = {}
vlans: dict[int, str] = {}
in_table = False
for line in output.splitlines():
@@ -857,7 +860,7 @@ def parse_vlans_brief(output: str) -> Dict[int, str]:
return vlans
def parse_vlan_ports_detail(output: str) -> Dict[str, List[str]]:
def parse_vlan_ports_detail(output: str) -> dict[str, list[str]]:
"""Parse ``show vlans <id>`` per-VLAN port detail output.
Returns ``{"tagged": [...], "untagged": [...]}`` for ONE vlan.
@@ -870,8 +873,8 @@ def parse_vlan_ports_detail(output: str) -> Dict[str, List[str]]:
2 Tagged Learn Up
Trk1 Tagged Learn Up
"""
tagged: List[str] = []
untagged: List[str] = []
tagged: list[str] = []
untagged: list[str] = []
in_table = False
for line in output.splitlines():
@@ -892,7 +895,7 @@ def parse_vlan_ports_detail(output: str) -> Dict[str, List[str]]:
return {"tagged": tagged, "untagged": untagged}
def derive_port_pvids(vlans: Dict[int, Dict]) -> Dict[str, int]:
def derive_port_pvids(vlans: dict[int, dict]) -> dict[str, int]:
"""Derive PVID (native/access VLAN) for each port from the ``get_vlans`` result.
On ProCurve a port is untagged in exactly ONE VLAN — that is its PVID.
@@ -910,7 +913,7 @@ def derive_port_pvids(vlans: Dict[int, Dict]) -> Dict[str, int]:
})
# → {"1": 1, "2": 1, "3": 1, "5": 1}
"""
pvids: Dict[str, int] = {}
pvids: dict[str, int] = {}
# First pass: assign PVID from untagged membership
for vid, vdata in sorted(vlans.items()):
napalm_procurve/procurve.py
+120 -125
View File
@@ -1,4 +1,3 @@
# -*- coding: utf-8 -*-
# Licensed under the Apache License, Version 2.0
"""NAPALM driver for HPE/Aruba ProCurve switches.
@@ -18,15 +17,16 @@ Netmiko device type: ``hp_procurve`` / ``hp_procurve_telnet``
import logging
import re
import socket
from typing import Any, ClassVar, Dict, List, Optional, Union
from typing import Any, ClassVar
import netaddr
from netmiko import ConnectHandler
from netmiko.exceptions import (
NetmikoAuthenticationException,
NetmikoTimeoutException,
from napalm.base.exceptions import (
CommandErrorException,
ConnectionClosedException,
ConnectionException,
MergeConfigException,
ReplaceConfigException,
)
from napalm.base.netmiko_helpers import netmiko_args
from napalm_device_types import (
ConfigLifecycleMixin,
FingerprintRule,
@@ -35,24 +35,20 @@ from napalm_device_types import (
run_fix_steps,
)
from napalm_device_types.models import InterfaceConfigDict, VlanConfigDict
from napalm.base import helpers as napalm_helpers
from napalm.base.exceptions import (
CommandErrorException,
ConnectionException,
ConnectionClosedException,
MergeConfigException,
ReplaceConfigException,
)
from napalm.base.netmiko_helpers import netmiko_args
from napalm_device_types.ssh_host_keys import (
KNOWN_KEYS_ARG,
OBSERVER_ARG,
HostKeyCheck,
SshHostKeyMismatch,
)
from netmiko import ConnectHandler
from netmiko.exceptions import (
NetmikoAuthenticationException,
NetmikoTimeoutException,
)
from napalm_procurve.api_client import ProcurveApiClient
from napalm_procurve import parsers
from napalm_procurve.api_client import ProcurveApiClient
logger = logging.getLogger("napalm_procurve")
@@ -118,7 +114,7 @@ class ProcurveDriver(ConfigLifecycleMixin, SwitchDriver):
#: Where the switch publishes its health values over SNMP
#: (napalm-device-types ``SnmpHealthProfile``); the application polls.
#: Memory is NETSWITCH-MIB ``hpLocalMemEntry``: .5 total, .6 free, .7 allocated.
SNMP_HEALTH_PROFILE: ClassVar[Dict[str, Any]] = {
SNMP_HEALTH_PROFILE: ClassVar[dict[str, Any]] = {
"uptime": {"oids": ["1.3.6.1.4.1.11.2.14.11.1.2.4.1.5.1"]},
"cpu": {"oids": ["1.3.6.1.4.1.11.2.14.11.5.1.9.6.1.0", "1.3.6.1.4.1.11.2.3.7.8.6.1.0"]},
"memory": {
@@ -134,7 +130,7 @@ class ProcurveDriver(ConfigLifecycleMixin, 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
@@ -147,11 +143,11 @@ class ProcurveDriver(ConfigLifecycleMixin, SwitchDriver):
# --- optional_args ---
self.port: int = optional_args.get("port", 22)
self.telnet_port: int = optional_args.get("telnet_port", 23)
self.force_transport: Optional[str] = optional_args.get("transport")
self.force_transport: str | None = optional_args.get("transport")
self.ssl_verify: bool = optional_args.get(
"ssl_verify", optional_args.get("verify_ssl", optional_args.get("verify", False))
)
self.api_version_hint: Optional[str] = optional_args.get("api_version")
self.api_version_hint: str | None = optional_args.get("api_version")
# Enable password for CLI (defaults to login password)
self._secret: str = optional_args.get("secret", password)
@@ -160,24 +156,24 @@ class ProcurveDriver(ConfigLifecycleMixin, SwitchDriver):
self.netmiko_optional_args = netmiko_args(optional_args)
# The host keys to expect and where the one seen is reported (netork#1167).
self._host_key_args: Dict[str, Any] = {
self._host_key_args: dict[str, Any] = {
k: optional_args[k] for k in (KNOWN_KEYS_ARG, OBSERVER_ARG) if k in optional_args
}
# --- runtime state ---
# Active transport: "api" | "ssh" | "ssh_legacy" | "telnet"
self._transport: Optional[str] = None
self._transport: str | None = None
# CLI backend (SSH or Telnet)
self._device: Optional[ConnectHandler] = None
self._device: ConnectHandler | None = None
# REST API backend
self._api: Optional[ProcurveApiClient] = None
self._api: ProcurveApiClient | None = None
# Why each transport failed during the current open(), as "<transport>: <reason>"
self._attempts: List[str] = []
self._attempts: list[str] = []
# Config management state (CLI only)
self._candidate_config: Optional[str] = None
self._candidate_mode: Optional[str] = None # "merge" | "replace"
self._backup_config: Optional[str] = None
self._candidate_config: str | None = None
self._candidate_mode: str | None = None # "merge" | "replace"
self._backup_config: str | None = None
# ------------------------------------------------------------------
# Connection management
@@ -242,18 +238,18 @@ class ProcurveDriver(ConfigLifecycleMixin, SwitchDriver):
try:
self._device.disconnect()
except Exception:
pass
logger.debug("closing the CLI session: ignored", exc_info=True)
self._device = None
self._transport = None
def is_alive(self) -> Dict[str, bool]:
def is_alive(self) -> dict[str, bool]:
"""Return transport-level liveness without writing to the channel."""
if self._api:
return {"is_alive": self._api.is_alive()}
if self._device:
try:
return {"is_alive": self._device.remote_conn.transport.is_active()}
except (AttributeError, socket.error, EOFError):
except (OSError, AttributeError, EOFError):
return {"is_alive": False}
return {"is_alive": False}
@@ -267,9 +263,7 @@ class ProcurveDriver(ConfigLifecycleMixin, SwitchDriver):
logger.info("Probing REST API for %s (timeout=%ds)", self.hostname, probe_timeout)
# Probe without SSL verification — no credentials are sent during probing,
# so this is safe and avoids failing on self-signed certificates.
ver, proto = ProcurveApiClient.probe(
self.hostname, timeout=probe_timeout, ssl_verify=False
)
ver, proto = ProcurveApiClient.probe(self.hostname, timeout=probe_timeout, ssl_verify=False)
# Allow hint override (e.g. user knows the API version)
if self.api_version_hint and proto:
ver = self.api_version_hint
@@ -281,8 +275,8 @@ class ProcurveDriver(ConfigLifecycleMixin, SwitchDriver):
# Try connecting with the requested SSL setting first; if it fails due to a
# self-signed certificate (ssl_verify=True), transparently retry unverified.
error: Optional[Exception] = None
for ssl_verify in ([self.ssl_verify] if not self.ssl_verify else [True, False]):
error: Exception | None = None
for ssl_verify in [self.ssl_verify] if not self.ssl_verify else [True, False]:
client = ProcurveApiClient(
hostname=self.hostname,
username=self.username,
@@ -300,8 +294,13 @@ class ProcurveDriver(ConfigLifecycleMixin, SwitchDriver):
continue
self._api = client
self._transport = "api"
logger.info("Connected to %s via REST API (%s %s, ssl_verify=%s)",
self.hostname, proto, ver, ssl_verify)
logger.info(
"Connected to %s via REST API (%s %s, ssl_verify=%s)",
self.hostname,
proto,
ver,
ssl_verify,
)
return True
logger.warning("REST API connect failed for %s — falling back to CLI", self.hostname)
@@ -314,8 +313,11 @@ class ProcurveDriver(ConfigLifecycleMixin, SwitchDriver):
Explicit fields (port, timeout, disabled_algorithms) take precedence and are
excluded from the spread to avoid 'multiple values for keyword argument' errors.
"""
extra = {k: v for k, v in self.netmiko_optional_args.items()
if k not in ("port", "timeout", "disabled_algorithms")}
extra = {
k: v
for k, v in self.netmiko_optional_args.items()
if k not in ("port", "timeout", "disabled_algorithms")
}
return dict(
device_type=self.NETMIKO_DEVICE_TYPE,
host=self.hostname,
@@ -328,7 +330,7 @@ class ProcurveDriver(ConfigLifecycleMixin, SwitchDriver):
**extra,
)
def _ssh_connect(self, legacy: bool = False, port: Optional[int] = None) -> ConnectHandler:
def _ssh_connect(self, legacy: bool = False, port: int | None = None) -> ConnectHandler:
"""A netmiko SSH session, its host key verified and reported (netork#1167).
Every SSH session of this driver opens here. A refused key raises
@@ -397,7 +399,7 @@ class ProcurveDriver(ConfigLifecycleMixin, SwitchDriver):
# CLI helpers
# ------------------------------------------------------------------
def _send_command(self, command: Union[str, List[str]]) -> str:
def _send_command(self, command: str | list[str]) -> str:
"""Send a CLI command and return stripped output.
When a list is supplied, commands are tried in order and the first
@@ -421,11 +423,15 @@ class ProcurveDriver(ConfigLifecycleMixin, SwitchDriver):
for cmd in command:
last = _do(cmd)
# ProCurve says "Invalid input: …"; other firmware "% Invalid …"
if "Invalid input" not in last and "% Invalid" not in last and "Error" not in last:
if (
"Invalid input" not in last
and "% Invalid" not in last
and "Error" not in last
):
return last
return last
return _do(command)
except (socket.error, EOFError) as exc:
except (OSError, EOFError) as exc:
raise ConnectionClosedException(str(exc)) from exc
def _exec_prompt(self) -> str:
@@ -462,13 +468,13 @@ class ProcurveDriver(ConfigLifecycleMixin, 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]:
"""Push config lines to the device while in config mode.
Returns a list of error messages for rejected lines.
"""
ep = self._any_prompt()
errors: List[str] = []
errors: list[str] = []
for line in config_text.splitlines():
stripped = line.strip()
if not stripped or stripped.startswith(("!", "#", ";")):
@@ -486,7 +492,7 @@ class ProcurveDriver(ConfigLifecycleMixin, SwitchDriver):
# NAPALM: get_facts
# ------------------------------------------------------------------
def get_facts(self) -> Dict:
def get_facts(self) -> dict:
"""Return general device information."""
if self._transport == "api":
return self._api.get_facts()
@@ -520,22 +526,20 @@ class ProcurveDriver(ConfigLifecycleMixin, SwitchDriver):
"interface_list": iface_list,
}
def _get_interface_list(self) -> List[str]:
def _get_interface_list(self) -> list[str]:
"""Return sorted list of interface names from ``show interfaces brief``."""
out = self._send_command("show interfaces brief")
ifaces = parsers.parse_interfaces_brief(out)
return sorted(
ifaces.keys(),
key=lambda s: [
int(x) if x.isdigit() else x for x in re.split(r"(\d+)", s)
],
key=lambda s: [int(x) if x.isdigit() else x for x in re.split(r"(\d+)", s)],
)
# ------------------------------------------------------------------
# NAPALM: get_interfaces
# ------------------------------------------------------------------
def get_interfaces(self) -> Dict[str, Dict]:
def get_interfaces(self) -> dict[str, dict]:
"""Return a dictionary of interface details."""
if self._transport == "api":
return self._api.get_interfaces()
@@ -552,7 +556,7 @@ class ProcurveDriver(ConfigLifecycleMixin, SwitchDriver):
if detail.get("description"):
ifaces[port]["description"] = detail["description"]
except Exception:
pass
logger.debug("show interfaces %s: ignored", port, exc_info=True)
# The CLI lists member ports only ("3-Trk3"); the trunk gets its own row.
return add_lag_interfaces(ifaces)
@@ -561,7 +565,7 @@ class ProcurveDriver(ConfigLifecycleMixin, SwitchDriver):
# NAPALM: get_interfaces_ip
# ------------------------------------------------------------------
def get_interfaces_ip(self) -> Dict[str, Dict]:
def get_interfaces_ip(self) -> dict[str, dict]:
"""Return IP addresses assigned to interfaces."""
if self._transport == "api":
return self._api.get_interfaces_ip()
@@ -573,7 +577,7 @@ class ProcurveDriver(ConfigLifecycleMixin, SwitchDriver):
# NAPALM: get_arp_table
# ------------------------------------------------------------------
def get_arp_table(self, vrf: str = "") -> List[Dict]:
def get_arp_table(self, vrf: str = "") -> list[dict]:
"""Return the ARP table."""
if self._transport == "api":
return self._api.get_arp_table(vrf)
@@ -585,7 +589,7 @@ class ProcurveDriver(ConfigLifecycleMixin, SwitchDriver):
# NAPALM: get_mac_address_table
# ------------------------------------------------------------------
def get_mac_address_table(self) -> List[Dict]:
def get_mac_address_table(self) -> list[dict]:
"""Return the MAC address table."""
if self._transport == "api":
return self._api.get_mac_address_table()
@@ -597,21 +601,19 @@ class ProcurveDriver(ConfigLifecycleMixin, SwitchDriver):
# NAPALM: get_lldp_neighbors
# ------------------------------------------------------------------
def get_lldp_neighbors(self) -> Dict[str, List[Dict]]:
def get_lldp_neighbors(self) -> dict[str, list[dict]]:
"""Return LLDP neighbor information."""
if self._transport == "api":
return self._api.get_lldp_neighbors()
out = self._send_command(
["show lldp info remote-device", "show lldp neighbors"]
)
out = self._send_command(["show lldp info remote-device", "show lldp neighbors"])
return parsers.parse_lldp_neighbors(out)
# ------------------------------------------------------------------
# NAPALM: get_lldp_neighbors_detail
# ------------------------------------------------------------------
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."""
if self._transport == "api":
return self._api.get_lldp_neighbors_detail()
@@ -629,7 +631,7 @@ class ProcurveDriver(ConfigLifecycleMixin, SwitchDriver):
def get_config(
self, retrieve: str = "all", full: bool = False, sanitized: bool = False
) -> Dict[str, str]:
) -> dict[str, str]:
"""Return device configuration."""
if self._transport == "api":
result = self._api.get_config()
@@ -641,7 +643,7 @@ class ProcurveDriver(ConfigLifecycleMixin, SwitchDriver):
try:
self._api._session.close()
except Exception:
pass
logger.debug("closing the REST session: ignored", exc_info=True)
running = self._get_config_via_ssh()
return {"running": running, "startup": "", "candidate": ""}
@@ -652,9 +654,7 @@ class ProcurveDriver(ConfigLifecycleMixin, SwitchDriver):
if retrieve in ("all", "running"):
running = self._send_command("show running-config")
if retrieve in ("all", "startup"):
startup = self._send_command(
["show startup-config", "show config"]
)
startup = self._send_command(["show startup-config", "show config"])
return {"running": running, "startup": startup, "candidate": candidate}
@@ -680,14 +680,12 @@ class ProcurveDriver(ConfigLifecycleMixin, SwitchDriver):
# NAPALM: get_environment
# ------------------------------------------------------------------
def get_environment(self) -> Dict:
def get_environment(self) -> dict:
"""Return environment information (temperature, fans, CPU, memory)."""
if self._transport == "api":
raise NotImplementedError("Environment data not available via REST API")
sysinfo = self._send_command(
["show system-information", "show system information"]
)
sysinfo = self._send_command(["show system-information", "show system information"])
env_out = self._send_command(["show environment", "show system"])
combined = sysinfo + "\n" + env_out
return parsers.parse_environment(combined)
@@ -696,25 +694,21 @@ class ProcurveDriver(ConfigLifecycleMixin, SwitchDriver):
# NAPALM: get_ntp_servers
# ------------------------------------------------------------------
def get_ntp_servers(self) -> Dict[str, Dict]:
def get_ntp_servers(self) -> dict[str, dict]:
"""Return configured NTP servers."""
if self._transport == "api":
return self._api.get_ntp_servers()
out = self._send_command(
["show ntp associations", "show ntp status", "show timesync"]
)
out = self._send_command(["show ntp associations", "show ntp status", "show timesync"])
return parsers.parse_ntp_servers(out)
# ------------------------------------------------------------------
# NAPALM: get_ntp_stats
# ------------------------------------------------------------------
def get_ntp_stats(self) -> List[Dict]:
def get_ntp_stats(self) -> list[dict]:
"""Return NTP peer statistics."""
out = self._send_command(
["show ntp associations", "show ntp status"]
)
out = self._send_command(["show ntp associations", "show ntp status"])
stats = []
for line in out.splitlines():
m = re.match(
@@ -746,7 +740,7 @@ class ProcurveDriver(ConfigLifecycleMixin, SwitchDriver):
# NAPALM: get_users
# ------------------------------------------------------------------
def get_users(self) -> Dict[str, Dict]:
def get_users(self) -> dict[str, dict]:
"""Return local users."""
out = self._send_command(["show local-users", "show users"])
return parsers.parse_users(out)
@@ -755,7 +749,7 @@ class ProcurveDriver(ConfigLifecycleMixin, SwitchDriver):
# NAPALM: get_snmp_information
# ------------------------------------------------------------------
def get_snmp_information(self) -> Dict:
def get_snmp_information(self) -> dict:
"""Return SNMP configuration."""
out = self._send_command(["show snmp-server", "show snmp"])
return parsers.parse_snmp_info(out)
@@ -764,7 +758,7 @@ class ProcurveDriver(ConfigLifecycleMixin, SwitchDriver):
# NAPALM: get_vlans (extended: tagged + untagged per VLAN)
# ------------------------------------------------------------------
def get_vlans(self) -> Dict[int, Dict]:
def get_vlans(self) -> dict[int, dict]:
"""Return VLAN information with tagged/untagged port separation.
This method extends the standard NAPALM ``get_vlans()`` return with
@@ -811,18 +805,18 @@ class ProcurveDriver(ConfigLifecycleMixin, SwitchDriver):
# Fallback: show vlans (name discovery) + show vlans <id> per VLAN
return self._get_vlans_via_show_vlans()
def _get_vlans_via_show_vlans(self) -> Dict[int, Dict]:
def _get_vlans_via_show_vlans(self) -> dict[int, dict]:
"""Fallback VLAN discovery via ``show vlans`` + per-VLAN detail queries."""
brief_out = self._send_command("show vlans")
vlan_names = parsers.parse_vlans_brief(brief_out)
result: Dict[int, Dict] = {}
result: dict[int, dict] = {}
for vid, name in vlan_names.items():
detail_out = self._send_command(f"show vlans {vid}")
ports = parsers.parse_vlan_ports_detail(detail_out)
tagged = ports["tagged"]
untagged = ports["untagged"]
interfaces: List[str] = []
interfaces: list[str] = []
for p in untagged + tagged:
if p not in interfaces:
interfaces.append(p)
@@ -839,7 +833,7 @@ class ProcurveDriver(ConfigLifecycleMixin, SwitchDriver):
# Custom: get_port_pvids (PVID per port, derived from get_vlans)
# ------------------------------------------------------------------
def get_port_pvids(self) -> Dict[str, int]:
def get_port_pvids(self) -> dict[str, int]:
"""Return the PVID (native / access VLAN) for every switch port.
On ProCurve, a port is untagged in exactly **one** VLAN — that VLAN's
@@ -906,9 +900,7 @@ class ProcurveDriver(ConfigLifecycleMixin, SwitchDriver):
lines.append("exit")
errors = self._apply_config_lines("\n".join(lines))
if errors:
raise CommandErrorException(
f"set_vlan({vlan_id}) errors: {errors}"
)
raise CommandErrorException(f"set_vlan({vlan_id}) errors: {errors}")
finally:
self._exit_config_mode()
self._save_config()
@@ -917,7 +909,7 @@ class ProcurveDriver(ConfigLifecycleMixin, 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.
Diffs ``members`` against the trunk's current members (as reported
@@ -939,7 +931,7 @@ class ProcurveDriver(ConfigLifecycleMixin, SwitchDriver):
mode = current.get("lag_mode") or "trunk"
trk_name = lag_name.lower()
commands: List[str] = []
commands: list[str] = []
if to_remove:
commands.append(f"no trunk {','.join(to_remove)}")
if to_add:
@@ -979,7 +971,8 @@ class ProcurveDriver(ConfigLifecycleMixin, SwitchDriver):
resp = self._api.put(f"ports/{interface}", json=patch)
if not resp.ok:
raise ConnectionException(
f"set_interface({interface}): ports PUT HTTP {resp.status_code} – {resp.text[:200]}"
f"set_interface({interface}): ports PUT HTTP {resp.status_code} – "
f"{resp.text[:200]}"
)
mode = config.get("mode")
if mode == "trunk":
@@ -1024,7 +1017,7 @@ class ProcurveDriver(ConfigLifecycleMixin, SwitchDriver):
mode = config.get("mode")
self._enter_config_mode()
try:
lines: List[str] = []
lines: list[str] = []
if "enabled" in config or "description" in config:
lines.append(f"interface {interface}")
if "enabled" in config:
@@ -1049,9 +1042,7 @@ class ProcurveDriver(ConfigLifecycleMixin, SwitchDriver):
if lines:
errors = self._apply_config_lines("\n".join(lines))
if errors:
raise CommandErrorException(
f"set_interface({interface}) errors: {errors}"
)
raise CommandErrorException(f"set_interface({interface}) errors: {errors}")
finally:
self._exit_config_mode()
self._save_config()
@@ -1068,9 +1059,7 @@ class ProcurveDriver(ConfigLifecycleMixin, SwitchDriver):
try:
errors = self._apply_config_lines(f"no vlan {vlan_id}")
if errors:
raise CommandErrorException(
f"delete_vlan({vlan_id}) errors: {errors}"
)
raise CommandErrorException(f"delete_vlan({vlan_id}) errors: {errors}")
finally:
self._exit_config_mode()
self._save_config()
@@ -1079,7 +1068,7 @@ class ProcurveDriver(ConfigLifecycleMixin, SwitchDriver):
# PoE
# ------------------------------------------------------------------
def get_poe_status(self) -> Dict[str, Dict]:
def get_poe_status(self) -> dict[str, dict]:
"""Return PoE configuration for all ports, keyed by interface name.
Only available via the REST API transport.
@@ -1088,7 +1077,7 @@ class ProcurveDriver(ConfigLifecycleMixin, SwitchDriver):
return self._api.get_poe_ports()
raise NotImplementedError("PoE status is only available via the REST API")
def set_poe(self, interface: str, config: Dict) -> None:
def set_poe(self, interface: str, config: dict) -> None:
"""Update PoE configuration for a single interface."""
if self._transport != "api":
raise NotImplementedError("PoE configuration is only available via the REST API")
@@ -1125,12 +1114,17 @@ class ProcurveDriver(ConfigLifecycleMixin, SwitchDriver):
size: int = 100,
count: int = 5,
vrf: str = "",
) -> Dict:
) -> dict:
"""Execute a ping from the device."""
if self._transport == "api":
return self._api.ping(
destination, source=source, ttl=ttl,
timeout=timeout, size=size, count=count, vrf=vrf,
destination,
source=source,
ttl=ttl,
timeout=timeout,
size=size,
count=count,
vrf=vrf,
)
# ProCurve CLI ping: "ping <ip> repetitions <n>"
@@ -1144,11 +1138,11 @@ class ProcurveDriver(ConfigLifecycleMixin, SwitchDriver):
# NAPALM: cli
# ------------------------------------------------------------------
def cli(self, commands: List[str], encoding: str = "text") -> Dict[str, str]:
def cli(self, commands: list[str], encoding: str = "text") -> dict[str, str]:
"""Execute raw CLI commands and return output dict."""
if self._transport == "api":
# ArubaOS REST API supports CLI via POST /rest/<ver>/cli
results: Dict[str, str] = {}
results: dict[str, str] = {}
for cmd in commands:
resp = self._api.post("cli", json={"cmd": cmd})
if resp.ok:
@@ -1166,9 +1160,7 @@ class ProcurveDriver(ConfigLifecycleMixin, SwitchDriver):
def _get_running_config(self) -> str:
return self._send_command("show running-config")
def load_merge_candidate(
self, filename: Optional[str] = None, config: Optional[str] = None
) -> None:
def load_merge_candidate(self, filename: str | None = None, config: str | None = None) -> None:
if self._transport == "api":
raise MergeConfigException(
"Merge config is not supported via REST API transport. Use SSH."
@@ -1176,7 +1168,7 @@ class ProcurveDriver(ConfigLifecycleMixin, SwitchDriver):
super().load_merge_candidate(filename, config)
def load_replace_candidate(
self, filename: Optional[str] = None, config: Optional[str] = None
self, filename: str | None = None, config: str | None = None
) -> None:
if self._transport == "api":
raise ReplaceConfigException(
@@ -1184,7 +1176,7 @@ class ProcurveDriver(ConfigLifecycleMixin, SwitchDriver):
)
super().load_replace_candidate(filename, 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 candidate configuration to the device."""
if self._candidate_config is None:
raise MergeConfigException("No candidate config loaded.")
@@ -1201,9 +1193,7 @@ class ProcurveDriver(ConfigLifecycleMixin, SwitchDriver):
raise MergeConfigException(f"Error applying config: {exc}") from exc
if errors:
raise MergeConfigException(
"The following lines were rejected:\n" + "\n".join(errors)
)
raise MergeConfigException("The following lines were rejected:\n" + "\n".join(errors))
self._save_config()
self._candidate_config = None
@@ -1224,11 +1214,15 @@ class ProcurveDriver(ConfigLifecycleMixin, SwitchDriver):
# Use a single command with short read_timeout to avoid blocking the poll.
try:
out = self._device.send_command(
out = (
self._device.send_command(
"show snmp-server",
expect_string=rf"{re.escape(self._device.base_prompt)}[>#]",
read_timeout=10,
).strip() if self._device else ""
).strip()
if self._device
else ""
)
except Exception:
return None
if not out:
@@ -1256,7 +1250,7 @@ class ProcurveDriver(ConfigLifecycleMixin, SwitchDriver):
# SNMP agent (napalm-device-types SnmpAgentMixin): the caller's values
# ------------------------------------------------------------------
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 agent is built in; configure community, contact and location, then save.
The REST API has no writable SNMP endpoint, so over REST a temporary SSH
@@ -1293,10 +1287,10 @@ class ProcurveDriver(ConfigLifecycleMixin, SwitchDriver):
try:
self._device.disconnect()
except Exception: # noqa: BLE001 - closing a temporary session
pass
logger.debug("closing the temporary CLI session: ignored", exc_info=True)
self._api, self._device, self._transport = old_api, old_device, old_transport
def _snmp_configure_cli(self, settings: Dict[str, Any]) -> Dict[str, Any]:
def _snmp_configure_cli(self, settings: dict[str, Any]) -> dict[str, Any]:
community = settings["community"]
if not _CLI_WORD.match(community):
raise ValueError("the community must be one word without quotes or #")
@@ -1317,12 +1311,12 @@ class ProcurveDriver(ConfigLifecycleMixin, SwitchDriver):
return {"ok": True, "output": "community (read-only), contact and location set"}
@staticmethod
def _snmp_sources_note(settings: Dict[str, Any]) -> Dict[str, Any]:
def _snmp_sources_note(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 _snmp_save_and_verify(self, settings: Dict[str, Any]) -> Dict[str, Any]:
def _snmp_save_and_verify(self, settings: dict[str, Any]) -> dict[str, Any]:
self._save_config()
shown = self._send_command("show snmp-server")
if re.search(rf"^\s*{re.escape(settings['community'])}\s", shown, re.M):
@@ -1334,6 +1328,7 @@ class ProcurveDriver(ConfigLifecycleMixin, SwitchDriver):
# Helpers
# ---------------------------------------------------------------------------
def _is_float(s: str) -> bool:
try:
float(s)
@@ -1342,7 +1337,7 @@ def _is_float(s: str) -> bool:
return False
def _parse_ping_output(output: str, count: int) -> Dict:
def _parse_ping_output(output: str, count: int) -> dict:
"""Parse ProCurve CLI ping output into NAPALM ping result dict.
ProCurve ping output::
@@ -1356,7 +1351,8 @@ def _parse_ping_output(output: str, count: int) -> Dict:
# Packet statistics line
m = re.search(
r"(\d+)\s+packets\s+transmitted,\s+(\d+)\s+packets\s+received",
output, re.I,
output,
re.I,
)
if not m:
# Check for total failure
@@ -1399,4 +1395,3 @@ def _parse_ping_output(output: str, count: int) -> Dict:
"results": results,
}
}
napalm_procurve/py.typed
View File
Whitespace-only changes.
pyproject.toml
+54 -16
View File
@@ -1,57 +1,95 @@
[build-system]
requires = ["setuptools>=64", "wheel"]
requires = ["setuptools>=77"]
build-backend = "setuptools.build_meta"
[project]
name = "napalm-hpe-aruba-procurve"
version = "0.5.0"
version = "0.6.0"
description = "NAPALM driver for HPE/Aruba ProCurve switches with auto-detecting transport (REST API, SSH, legacy SSH, Telnet)"
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.10",
"requests>=2.25.0",
"urllib3",
"urllib3>=1.23",
]
[project.optional-dependencies]
dev = [
"pytest",
"pytest-cov",
"black",
"ruff",
"pytest>=8",
"pytest-cov>=5",
"mypy>=1.11",
"ruff>=0.6",
]
[project.entry-points."napalm.drivers"]
procurve = "napalm_procurve:ProcurveDriver"
[project.urls]
Repository = "https://github.com/chrismanivong/napalm-hpe-aruba-procurve"
Repository = "https://git.netork.io/NAPALM/napalm-hpe-aruba-procurve"
Issues = "https://git.netork.io/NAPALM/napalm-hpe-aruba-procurve/issues"
Changelog = "https://git.netork.io/NAPALM/napalm-hpe-aruba-procurve/src/branch/main/CHANGELOG.md"
[tool.setuptools.packages.find]
where = ["."]
include = ["napalm_procurve*"]
[tool.setuptools.package-data]
napalm_procurve = ["py.typed"]
[tool.napalm-standard]
# The import package is not named after the distribution.
package = "napalm_procurve"
[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.*", "netmiko.*", "paramiko.*", "netaddr.*"]
ignore_missing_imports = true
[[tool.mypy.overrides]]
# Not yet under strict mypy (quality baseline); may only shrink.
module = ["napalm_procurve.api_client", "napalm_procurve.parsers", "napalm_procurve.procurve"]
ignore_errors = true
[tool.pytest.ini_options]
testpaths = ["tests"]
[tool.coverage.run]
source = ["napalm_procurve"]
[tool.coverage.report]
fail_under = 54
tests/quality_baseline.json
+35
View File
@@ -0,0 +1,35 @@
{
"coverage_floor": 54.0,
"function_complexity": {
"napalm_procurve/api_client.py::ProcurveApiClient.get_interfaces": 15,
"napalm_procurve/api_client.py::ProcurveApiClient.get_vlans": 15,
"napalm_procurve/parsers.py::_parse_procurve_uptime": 11,
"napalm_procurve/parsers.py::parse_ip_interfaces": 14,
"napalm_procurve/parsers.py::parse_lldp_neighbors_detail": 16,
"napalm_procurve/parsers.py::parse_system_info": 18,
"napalm_procurve/parsers.py::parse_vlans_from_running_config": 11,
"napalm_procurve/procurve.py::ProcurveDriver._cli_set_interface": 13,
"napalm_procurve/procurve.py::ProcurveDriver.open": 13,
"napalm_procurve/procurve.py::ProcurveDriver.set_lag_members": 12
},
"function_lines": {
"napalm_procurve/api_client.py::ProcurveApiClient.get_interfaces": 65,
"napalm_procurve/api_client.py::ProcurveApiClient.get_vlans": 101,
"napalm_procurve/parsers.py::parse_ip_interfaces": 58,
"napalm_procurve/parsers.py::parse_lldp_neighbors_detail": 59,
"napalm_procurve/parsers.py::parse_system_info": 80,
"napalm_procurve/parsers.py::parse_vlans_from_running_config": 79,
"napalm_procurve/procurve.py::_parse_ping_output": 58
},
"mypy_relaxed": [
"napalm_procurve.api_client",
"napalm_procurve.parsers",
"napalm_procurve.procurve"
],
"python_file_lines": {
"napalm_procurve/procurve.py": 1397
},
"untyped_public": [
"napalm_procurve/procurve.py::ProcurveDriver.get_snmp_config"
]
}
tests/unit/test_driver.py
+36 -15
View File
@@ -1,13 +1,12 @@
"""Unit tests for ProcurveDriver and parsers — no real device required."""
import pytest
from unittest.mock import MagicMock, patch
import pytest
from napalm.base.exceptions import ConnectionException
from napalm_procurve.procurve import _SSH_DISABLED_STANDARD, ProcurveDriver, _parse_ping_output
from napalm_procurve import parsers
from napalm_procurve.procurve import _SSH_DISABLED_STANDARD, ProcurveDriver, _parse_ping_output
# ===========================================================================
# Sample CLI output fixtures (captured from real ProCurve devices)
@@ -510,7 +509,9 @@ class TestDriverTransportDetection:
with patch("napalm_procurve.procurve.ConnectHandler") as mock_conn:
mock_conn.return_value = MagicMock()
drv = ProcurveDriver(
"192.168.0.1", "manager", "secret",
"192.168.0.1",
"manager",
"secret",
optional_args={"transport": "ssh"},
)
drv.open()
@@ -521,7 +522,9 @@ class TestDriverTransportDetection:
with patch("napalm_procurve.procurve.ConnectHandler") as mock_conn:
mock_conn.return_value = MagicMock()
drv = ProcurveDriver(
"192.168.0.1", "manager", "secret",
"192.168.0.1",
"manager",
"secret",
optional_args={"transport": "telnet"},
)
drv.open()
@@ -532,7 +535,7 @@ class TestDriverTransportDetection:
with patch("napalm_procurve.procurve.ProcurveApiClient.probe", return_value=(None, None)):
with patch("napalm_procurve.procurve.ConnectHandler", side_effect=Exception("no conn")):
drv = ProcurveDriver("192.168.0.1", "manager", "secret")
with pytest.raises(Exception):
with pytest.raises(ConnectionException):
drv.open()
def test_auth_failure_names_why_earlier_transports_failed(self):
@@ -565,7 +568,9 @@ class TestDriverTransportDetection:
def test_all_transports_fail_names_each_reason(self):
"""The final error gives a reason per transport, not just its name."""
with patch("napalm_procurve.procurve.ProcurveApiClient.probe", return_value=(None, None)):
with patch("napalm_procurve.procurve.ConnectHandler", side_effect=OSError("no route to host")):
with patch(
"napalm_procurve.procurve.ConnectHandler", side_effect=OSError("no route to host")
):
drv = ProcurveDriver("192.168.0.1", "manager", "secret")
with pytest.raises(ConnectionException) as exc_info:
drv.open()
@@ -900,22 +905,36 @@ class TestApiGetInterfacesLag:
client = ProcurveApiClient(hostname="192.168.0.1", username="manager", password="secret")
responses = {
"ports": {"port_element": ports},
"port-statistics": {"port_statistics_element": [
{"id": p["id"], "port_speed_mbps": 1000} for p in ports
]},
"port-statistics": {
"port_statistics_element": [{"id": p["id"], "port_speed_mbps": 1000} for p in ports]
},
"system/status/switch": {},
}
client.get = MagicMock(side_effect=lambda endpoint, **kw: responses[endpoint])
return client
def test_trunk_rows_with_mode(self):
client = self._client([
client = self._client(
[
{"id": "1", "is_port_up": True, "is_port_enabled": True, "trunk_group": ""},
{"id": "3", "is_port_up": False, "is_port_enabled": True, "trunk_group": "trk3", "trunk_mode": "PTT_LACP"},
{"id": "4", "is_port_up": True, "is_port_enabled": True, "trunk_group": "trk3", "trunk_mode": "PTT_LACP"},
{
"id": "3",
"is_port_up": False,
"is_port_enabled": True,
"trunk_group": "trk3",
"trunk_mode": "PTT_LACP",
},
{
"id": "4",
"is_port_up": True,
"is_port_enabled": True,
"trunk_group": "trk3",
"trunk_mode": "PTT_LACP",
},
{"id": "10", "is_port_up": False, "is_port_enabled": True, "trunk_group": "trk6"},
{"id": "7", "is_port_up": False, "is_port_enabled": True, "trunk_group": "trk6"},
])
]
)
ifaces = client.get_interfaces()
assert ifaces["trk3"]["lag_members"] == ["3", "4"]
@@ -931,7 +950,9 @@ class TestApiGetInterfacesLag:
class TestCliGetInterfacesLag:
def test_trunk_rows_from_member_ports(self, driver):
driver._send_command = MagicMock(
side_effect=lambda cmd: SHOW_INTERFACES_BRIEF_INTRUSION if cmd == "show interfaces brief" else ""
side_effect=lambda cmd: (
SHOW_INTERFACES_BRIEF_INTRUSION if cmd == "show interfaces brief" else ""
)
)
ifaces = driver.get_interfaces()
tests/unit/test_ssh_host_key.py
+4 -2
View File
@@ -30,7 +30,9 @@ HANDLER = "napalm_procurve.procurve.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
)
device.send_command.return_value = "hostname x"
return device
@@ -41,7 +43,7 @@ def _refuse(**kwargs):
try:
raise paramiko.BadHostKeyException("10.0.0.4", SERVER_KEY, OTHER_KEY)
except paramiko.SSHException as exc:
raise NetmikoTimeoutException(f"A paramiko SSHException occurred: {exc}")
raise NetmikoTimeoutException(f"A paramiko SSHException occurred: {exc}") from exc
def _driver(seen: list, known: dict | None = None, **extra) -> ProcurveDriver: