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