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

chore: adopt the NAPALM driver standard #19

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