fix(cli): the trunk gets its own row, as over REST

Since f35b59e a CLI poll reports trunk members as `3` with
trunk_group `Trk3`, but `show interfaces brief` has no line for Trk3
itself, so the trunk was missing from the interface list. The REST path
built that row on its own.

Both paths now use napalm_device_types.add_lag_interfaces, and the REST
path's copy is gone. One visible change there: `lag_members` is now in
port order (7, 10) instead of API order; the description already was.

The CLI path does not know the mode yet, so its trunks carry no
lag_mode. That needs `show trunks`, whose output is not at hand.

Needs napalm-device-types f3fa75b.

Closes #5

napalm_hpe_aruba_procurve.egg-info/PKG-INFO

@@ -0,0 +1,142 @@
+Metadata-Version: 2.4
+Name: napalm-hpe-aruba-procurve
+Version: 0.2.0
+Summary: NAPALM driver for HPE/Aruba ProCurve switches with auto-detecting transport (REST API, SSH, legacy SSH, Telnet)
+Author: Christian Manivong
+License: Apache-2.0
+Project-URL: Repository, https://github.com/chrismanivong/napalm-hpe-aruba-procurve
+Classifier: Topic :: Utilities
+Classifier: License :: OSI Approved :: Apache Software License
+Classifier: Programming Language :: Python :: 3
+Classifier: Programming Language :: Python :: 3.8
+Classifier: Programming Language :: Python :: 3.9
+Classifier: Programming Language :: Python :: 3.10
+Classifier: Programming Language :: Python :: 3.11
+Classifier: Programming Language :: Python :: 3.12
+Classifier: Operating System :: POSIX :: Linux
+Classifier: Operating System :: MacOS
+Requires-Python: >=3.8
+Description-Content-Type: text/markdown
+Requires-Dist: napalm>=4.0.0
+Requires-Dist: netmiko>=4.0.0
+Requires-Dist: paramiko>=5.0.0
+Requires-Dist: netaddr
+Requires-Dist: requests>=2.25.0
+Requires-Dist: urllib3
+Provides-Extra: dev
+Requires-Dist: pytest; extra == "dev"
+Requires-Dist: pytest-cov; extra == "dev"
+Requires-Dist: black; extra == "dev"
+Requires-Dist: ruff; extra == "dev"
+
+# napalm-hpe-aruba-procurve
+
+NAPALM driver for **HPE / Aruba ProCurve** switches — from ancient 2520G-8-PoE
+to modern 2530/2540 series.
+
+## Transport auto-detection
+
+The driver probes transports in order and uses the first one that succeeds:
+
+| Priority | Transport | Notes |
+|----------|-----------|-------|
+| 1 | REST API (HTTPS → HTTP, v7 → v6) | Newer 2530/2540 with `rest-interface` enabled |
+| 2 | SSH (standard) | hp_procurve netmiko driver |
+| 3 | SSH (legacy KEX) | Forces older kex/cipher negotiation for 2520G-8-PoE etc. |
+| 4 | Telnet | Oldest devices without SSH or with broken SSH |
+
+Override the transport with `optional_args={"transport": "ssh_legacy"}` to skip
+auto-detection.
+
+## Switch prerequisites
+
+### REST API (newer switches, e.g. 2530, 2540)
+
+```
+web-management ssl
+rest-interface
+rest-interface session-idle-timeout 120
+```
+
+### SSH (all ProCurve)
+
+```
+crypto key generate ssh rsa
+ip ssh
+```
+
+Telnet is enabled by default on most ProCurve switches.
+
+## Installation
+
+```bash
+pip install napalm napalm-hpe-aruba-procurve
+```
+
+Or from source:
+
+```bash
+git clone https://github.com/chrismanivong/napalm-hpe-aruba-procurve
+pip install -e napalm-hpe-aruba-procurve/
+```
+
+## Quick start
+
+```python
+from napalm import get_network_driver
+
+Driver = get_network_driver("procurve")
+
+with Driver(
+    "10.0.0.1",
+    "manager",
+    "secret",
+    optional_args={
+        # "transport": "ssh",          # force transport (api/ssh/ssh_legacy/telnet)
+        # "port": 22,
+        # "ssl_verify": False,         # disable SSL cert check for API
+        # "api_version": "v7",         # API version hint (v3/v6/v7)
+        # "secret": "enablepassword",  # enable password for CLI
+    },
+) as dev:
+    print(dev.get_facts())
+    print(dev.get_interfaces())
+```
+
+## Supported NAPALM methods
+
+| Method | API | SSH/Telnet |
+|--------|-----|------------|
+| `open()` | ✅ | ✅ |
+| `close()` | ✅ | ✅ |
+| `is_alive()` | ✅ | ✅ |
+| `get_facts()` | ✅ | ✅ |
+| `get_interfaces()` | ✅ | ✅ |
+| `get_interfaces_ip()` | ✅ | ✅ |
+| `get_arp_table()` | ✅ | ✅ |
+| `get_mac_address_table()` | ✅ | ✅ |
+| `get_lldp_neighbors()` | ✅ | ✅ |
+| `get_lldp_neighbors_detail()` | ✅ | ✅ |
+| `get_config()` | ✅ | ✅ |
+| `get_ntp_servers()` | ✅ | ✅ |
+| `get_environment()` | ❌ | ✅ |
+| `get_users()` | ❌ | ✅ |
+| `get_snmp_information()` | ❌ | ✅ |
+| `ping()` | ✅ | ✅ |
+| `cli()` | ✅ | ✅ |
+| `load_merge_candidate()` | ❌ | ✅ |
+| `load_replace_candidate()` | ❌ | ✅ |
+| `compare_config()` | ❌ | ✅ |
+| `commit_config()` | ❌ | ✅ |
+| `discard_config()` | ❌ | ✅ |
+| `rollback()` | ❌ | ✅ |
+
+## Tested devices
+
+- HP ProCurve 2520G-8-PoE (J9565A) — SSH legacy / Telnet
+- HP ProCurve 2920-48G — SSH
+- Aruba 2530-8-PoE+ — SSH + REST API
+
+## License
+
+Apache 2.0

napalm_hpe_aruba_procurve.egg-info/SOURCES.txt

@@ -0,0 +1,12 @@
+README.md
+pyproject.toml
+napalm_hpe_aruba_procurve.egg-info/PKG-INFO
+napalm_hpe_aruba_procurve.egg-info/SOURCES.txt
+napalm_hpe_aruba_procurve.egg-info/dependency_links.txt
+napalm_hpe_aruba_procurve.egg-info/entry_points.txt
+napalm_hpe_aruba_procurve.egg-info/requires.txt
+napalm_hpe_aruba_procurve.egg-info/top_level.txt
+napalm_procurve/__init__.py
+napalm_procurve/api_client.py
+napalm_procurve/parsers.py
+napalm_procurve/procurve.py

napalm_hpe_aruba_procurve.egg-info/dependency_links.txt

@@ -0,0 +1 @@
+

napalm_hpe_aruba_procurve.egg-info/entry_points.txt

@@ -0,0 +1,2 @@
+[napalm.drivers]
+procurve = napalm_procurve:ProcurveDriver

napalm_hpe_aruba_procurve.egg-info/requires.txt

@@ -0,0 +1,12 @@
+napalm>=4.0.0
+netmiko>=4.0.0
+paramiko>=5.0.0
+netaddr
+requests>=2.25.0
+urllib3
+
+[dev]
+pytest
+pytest-cov
+black
+ruff

napalm_hpe_aruba_procurve.egg-info/top_level.txt

@@ -0,0 +1 @@
+napalm_procurve

napalm_procurve/api_client.py

@@ -13,7 +13,9 @@ from typing import Any, Dict, Optional, Tuple
 import requests
 import urllib3
 
+from napalm.base import helpers as napalm_helpers
 from napalm.base.exceptions import ConnectionException, ConnectAuthError
+from napalm_device_types import add_lag_interfaces
 
 logger = logging.getLogger("napalm_procurve.api")
 
@@ -225,9 +227,24 @@ class ProcurveApiClient:
             for port in blade.get("data_ports", []):
                 iface_list.append(port.get("port_name", ""))
 
+        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 ""
+        if pn_match:
+            clean = _PN_RE.sub("", model_raw)
+            clean = _re.sub(r"\(\s*\)", "", clean)
+            clean = _re.sub(r"\s{2,}", " ", clean).strip(" -()")
+            model = clean if clean else model_raw
+        else:
+            model = model_raw
+
         return {
             "vendor": "HPE Aruba",
-            "model": system.get("product_model", ""),
+            "model": model,
+            "part_number": part_number,
             "hostname": hostname,
             "fqdn": fqdn,
             "os_version": system.get("firmware_version", ""),
@@ -291,22 +308,11 @@ class ProcurveApiClient:
             if pid in output:
                 output[pid]["speed"] = float(stat.get("port_speed_mbps", 0))
 
-        # Synthesize a logical interface entry for each configured LAG/trunk
-        # group so it shows up as its own row alongside its member ports.
-        for group, members in trunk_groups.items():
-            output[group] = {
-                "is_up": any(output[m]["is_up"] for m in members),
-                "is_enabled": any(output[m]["is_enabled"] for m in members),
-                "description": f"LAG ({', '.join(sorted(members, key=lambda s: int(s) if s.isdigit() else 0))})",
-                "last_flapped": -1.0,
-                "speed": sum(output[m]["speed"] for m in members),
-                "mtu": -1,
-                "mac_address": "",
-                "lag_members": members,
-                "lag_mode": "lacp" if trunk_modes.get(group) == "PTT_LACP" else "trunk",
-            }
-
-        return output
+        # 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
+        })
 
     def get_interfaces_ip(self) -> Dict[str, Dict]:
         """Return NAPALM interfaces IP from the REST API."""
@@ -366,9 +372,14 @@ class ProcurveApiClient:
         mac_data = self.get("mac-table")
         table = []
         for entry in mac_data.get("mac_table_entry_element", []):
+            raw_mac = entry.get("mac_address", "")
+            try:
+                mac = napalm_helpers.mac(raw_mac)
+            except Exception:
+                mac = raw_mac
             table.append(
                 {
-                    "mac": entry.get("mac_address", ""),
+                    "mac": mac,
                     "interface": entry.get("port_id", ""),
                     "vlan": int(entry.get("vlan_id", 0)),
                     "static": entry.get("mac_addr_type", "").lower() == "static",
@@ -414,15 +425,28 @@ class ProcurveApiClient:
         return result
 
     def get_config(self) -> Dict[str, str]:
-        """Return running and startup configuration via REST API."""
-        running = self.get("running-config")
-        startup = self.get("startup-config")
+        """Return running and startup configuration via REST API.
+
+        AOS-Switch returns the config as plain text, not JSON.
+        """
         return {
-            "running": running.get("config", ""),
-            "startup": startup.get("config", ""),
+            "running": self._get_text("running-config"),
+            "startup": self._get_text("startup-config"),
             "candidate": "",
         }
 
+    def _get_text(self, endpoint: str) -> str:
+        """GET *endpoint* and return the response body as plain text."""
+        url = self._base_url + endpoint
+        try:
+            resp = self._session.get(url, timeout=self.timeout)
+            if resp.ok:
+                return resp.text
+            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 ""
+
     def get_ntp_servers(self) -> Dict[str, Dict]:
         """Return NTP servers from REST API."""
         data = self.get("ntp/server")

napalm_procurve/parsers.py

@@ -89,7 +89,23 @@ def parse_system_info(output: str) -> Dict:
     )
     serial = get("Serial Number") or get("Serial Num")
     base_mac = procurve_mac_to_std(get("Base MAC Addr") or get("MAC Address") or "")
-    model = get("System Model") or get("product_model") or ""
+    model_raw = get("System Model") or get("product_model") or ""
+
+    # Extract HPE/HP J-code part numbers (e.g. "J9298A") from the model string.
+    # The J-code is the PN; the remaining text is the clean model name.
+    # Examples:
+    #   "HP J9298A Switch 2520G-8-PoE"  → model="HP 2520G-8-PoE Switch", pn="J9298A"
+    #   "HP2530-8G Switch(J9777A)"       → model="HP2530-8G Switch",       pn="J9777A"
+    _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 ""
+    if pn_match:
+        clean = _PN_RE.sub("", model_raw)
+        clean = re.sub(r"\(\s*\)", "", clean)
+        clean = re.sub(r"\s{2,}", " ", clean).strip(" -()")
+        model = clean if clean else model_raw
+    else:
+        model = model_raw
 
     # Parse uptime — ProCurve formats: "5 days  3:14" or "0 days  0:05"
     # Also: "5 day(s), 3:14:00"
@@ -101,6 +117,7 @@ def parse_system_info(output: str) -> Dict:
         "serial_number": serial,
         "base_mac": base_mac,
         "model": model,
+        "part_number": part_number,
         "uptime_seconds": uptime_seconds,
     }
 
@@ -171,20 +188,40 @@ def parse_version(output: str) -> str:
     return ""
 
 
-def parse_model_from_version(output: str) -> str:
-    """Extract model from ``show version`` first line.
+def parse_model_from_version(output: str) -> tuple[str, str]:
+    """Extract (model, part_number) from ``show version`` first line.
 
-    Example: ``HP J9565A 2520G-8-PoE Switch``  → ``2520G-8-PoE``
+    Handles both orderings of the HP banner:
+      ``HP J9565A 2520G-8-PoE Switch`` → ("2520G-8-PoE", "J9565A")
+      ``HP J9298A Switch 2520G-8-PoE`` → ("2520G-8-PoE", "J9298A")
+      ``Aruba 2530-8-PoE+ Switch``     → ("2530-8-PoE+", "")
+
+    Returns ("", "") when the first line is not an HP/Aruba device banner
+    (e.g. when show version starts with "Image stamp:" on newer Aruba).
     """
-    first = output.strip().splitlines()[0] if output.strip() else ""
-    m = re.search(r"HP\s+\S+\s+(\S+)\s+Switch", first, re.I)
-    if m:
-        return m.group(1)
-    # Aruba format: "Aruba 2530-8-PoE+ Switch"
-    m = re.search(r"(?:Aruba|HP)\s+(\d\S+)\s+(?:Switch|)", first, re.I)
-    if m:
-        return m.group(1)
-    return ""
+    # Find the HP/Aruba product banner line — may not be the first line
+    # (older ProCurve shows "Image stamp:" first, banner follows later)
+    banner = ""
+    for line in output.splitlines():
+        if re.match(r"\s*(?:HP|HPE|Aruba)\b", line, re.I):
+            banner = line.strip()
+            break
+    if not banner:
+        return "", ""
+    first = banner
+
+    # Extract J-code part number
+    _PN_RE_V = re.compile(r"\b(J\d{4}[A-Z]{1,2})\b")
+    pn_match = _PN_RE_V.search(first)
+    part_number = pn_match.group(1) if pn_match else ""
+
+    # Strip vendor prefix, J-code and "Switch" keyword to get the product name
+    cleaned = re.sub(r"\b(?:HP|HPE|Aruba)\b", "", first, flags=re.I)
+    cleaned = _PN_RE_V.sub("", cleaned)
+    cleaned = re.sub(r"\bSwitch\b", "", cleaned, flags=re.I)
+    cleaned = re.sub(r"\s{2,}", " ", cleaned).strip(" -()")
+
+    return cleaned, part_number
 
 
 # ---------------------------------------------------------------------------
@@ -197,12 +234,13 @@ def parse_model_from_version(output: str) -> str:
 #   1      100/1000T    | Yes      Down    Auto   Unknown      off      0
 
 _INTF_BRIEF_RE = re.compile(
-    r"^\s*(\S+)\s+"          # Port
+    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+"        # Enabled
+    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+"                # MDI (ignored)
-    r"(\S+)",                # Mode (speed/duplex)
+    r"(\S+)\s+(\S+)",         # Mode and MDI, in either order depending on firmware
     re.IGNORECASE,
 )
 
@@ -218,13 +256,13 @@ def parse_interfaces_brief(output: str) -> Dict[str, Dict]:
         m = _INTF_BRIEF_RE.match(line)
         if not m:
             continue
-        port, itype, enabled, link, mode = (
-            m.group(1), m.group(2), m.group(3), m.group(4), m.group(5)
-        )
+        port, trunk_group, itype, enabled, link, *mode_or_mdi = m.groups()
         speed = 0.0
         duplex = ""
-        # Mode examples: "1000FDx", "100HDx", "Unknown", "Auto"
-        sm = re.match(r"(\d+)(FDx|HDx)?", mode, re.I)
+        # 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
+        )
         if sm:
             speed = float(sm.group(1))
             duplex = "full" if (sm.group(2) or "").lower() == "fdx" else "half"
@@ -239,6 +277,8 @@ def parse_interfaces_brief(output: str) -> Dict[str, Dict]:
             "mtu": -1,
             "mac_address": "",
         }
+        if trunk_group:
+            interfaces[port]["trunk_group"] = trunk_group
     return interfaces
 
 

napalm_procurve/procurve.py

@@ -27,7 +27,12 @@ from netmiko.exceptions import (
     NetmikoAuthenticationException,
     NetmikoTimeoutException,
 )
-from napalm_device_types import SwitchDriver
+from napalm_device_types import (
+    ConfigLifecycleMixin,
+    FingerprintRule,
+    SwitchDriver,
+    add_lag_interfaces,
+)
 from napalm_device_types.models import InterfaceConfigDict, VlanConfigDict
 from napalm.base import helpers as napalm_helpers
 from napalm.base.exceptions import (
@@ -71,13 +76,31 @@ _SSH_DISABLED_LEGACY = {
 }
 
 
-class ProcurveDriver(SwitchDriver):
+class ProcurveDriver(ConfigLifecycleMixin, SwitchDriver):
     """NAPALM driver for HPE/Aruba ProCurve switches.
 
     Auto-detects transport on ``open()``.  See module docstring for details.
     """
 
     VENDOR = "HPE"
+    DRIVER_NAME = "procurve"
+    SNMP_OBJECT_ID_PREFIX = "1.3.6.1.4.1.11"
+    OUI_PREFIXES = [
+        "3C:D9:2B",  # Hewlett Packard — IEEE
+        "3C:52:82",  # Hewlett Packard — IEEE
+        "44:31:92",  # Hewlett Packard — IEEE
+        "00:17:A4",  # Hewlett Packard — IEEE
+    ]
+    SSH_FINGERPRINT = [
+        FingerprintRule("hp", weight=5.0),
+        FingerprintRule("aruba", weight=4.0),
+    ]
+    HTTP_FINGERPRINT = [
+        FingerprintRule("hp procurve", weight=8.0, mandatory=True),
+        FingerprintRule("procurve", weight=6.0),
+        FingerprintRule("ehttp", weight=5.0),
+        FingerprintRule("comware", weight=5.0),
+    ]
     NETMIKO_DEVICE_TYPE = "hp_procurve"
     NETMIKO_DEVICE_TYPE_TELNET = "hp_procurve_telnet"
 
@@ -120,6 +143,8 @@ class ProcurveDriver(SwitchDriver):
         self._device: Optional[ConnectHandler] = None
         # REST API backend
         self._api: Optional[ProcurveApiClient] = None
+        # Why each transport failed during the current open(), as "<transport>: <reason>"
+        self._attempts: List[str] = []
 
         # Config management state (CLI only)
         self._candidate_config: Optional[str] = None
@@ -141,38 +166,45 @@ class ProcurveDriver(SwitchDriver):
         4. Telnet
         """
         forced = self.force_transport
-        tried: List[str] = []
+        self._attempts = []
 
         # --- 1. REST API ---
         if not forced or forced == "api":
             if self._try_api():
                 return
-            tried.append("api")
 
         # --- 2. SSH standard ---
         if not forced or forced == "ssh":
             if self._try_ssh(legacy=False):
                 return
-            tried.append("ssh")
 
         # --- 3. SSH legacy KEX ---
         if not forced or forced == "ssh_legacy":
             if self._try_ssh(legacy=True):
                 return
-            tried.append("ssh_legacy")
 
         # --- 4. Telnet ---
         if not forced or forced == "telnet":
             if self._try_telnet():
                 return
-            tried.append("telnet")
 
         raise ConnectionException(
             f"Cannot connect to {self.hostname}. "
-            f"Tried transports: {', '.join(tried)}. "
+            f"Tried transports: {'; '.join(self._attempts)}. "
             "Check connectivity, credentials and whether SSH/Telnet/API is enabled."
         )
 
+    def _auth_failure(self, transport: str, exc: Exception) -> ConnectionException:
+        """An authentication error that also says why the earlier transports failed.
+
+        Without them, a Telnet "Login failed" reads as a wrong password when SSH
+        was merely refused and Telnet was the only transport left to answer (#2).
+        """
+        msg = f"Authentication failed for {self.hostname} via {transport}: {exc}"
+        if self._attempts:
+            msg += f" (earlier: {'; '.join(self._attempts)})"
+        return ConnectionException(msg)
+
     def close(self) -> None:
         """Close the active connection."""
         if self._api:
@@ -216,10 +248,12 @@ class ProcurveDriver(SwitchDriver):
 
         if not ver:
             logger.warning("REST API not detected on %s — falling back to CLI", self.hostname)
+            self._attempts.append("api: not detected")
             return False
 
         # 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]):
             client = ProcurveApiClient(
                 hostname=self.hostname,
@@ -234,6 +268,7 @@ class ProcurveDriver(SwitchDriver):
                 client.connect()
             except Exception as exc:
                 logger.debug("REST API connect failed (ssl_verify=%s): %s", ssl_verify, exc)
+                error = exc
                 continue
             self._api = client
             self._transport = "api"
@@ -242,6 +277,7 @@ class ProcurveDriver(SwitchDriver):
             return True
 
         logger.warning("REST API connect failed for %s — falling back to CLI", self.hostname)
+        self._attempts.append(f"api: {error}")
         return False
 
     def _netmiko_kwargs(self, legacy: bool = False) -> dict:
@@ -267,22 +303,23 @@ class ProcurveDriver(SwitchDriver):
     def _try_ssh(self, legacy: bool = False) -> bool:
         """Probe and connect via SSH.  Returns True on success."""
         label = "SSH-legacy" if legacy else "SSH"
+        transport = "ssh_legacy" if legacy else "ssh"
         logger.debug("Trying %s for %s", label, self.hostname)
         try:
             conn = ConnectHandler(**self._netmiko_kwargs(legacy))
             self._device = conn
-            self._transport = "ssh_legacy" if legacy else "ssh"
+            self._transport = transport
             logger.info("Connected to %s via %s", self.hostname, label)
             return True
         except NetmikoAuthenticationException as exc:
-            raise ConnectionException(
-                f"Authentication failed for {self.hostname}: {exc}"
-            ) from exc
+            raise self._auth_failure(transport, exc) from exc
         except NetmikoTimeoutException:
             logger.debug("%s timeout for %s", label, self.hostname)
+            self._attempts.append(f"{transport}: timed out")
             return False
         except Exception as exc:
             logger.debug("%s failed for %s: %s", label, self.hostname, exc)
+            self._attempts.append(f"{transport}: {exc}")
             return False
 
     def _try_telnet(self) -> bool:
@@ -303,11 +340,10 @@ class ProcurveDriver(SwitchDriver):
             logger.info("Connected to %s via Telnet", self.hostname)
             return True
         except NetmikoAuthenticationException as exc:
-            raise ConnectionException(
-                f"Authentication failed for {self.hostname}: {exc}"
-            ) from exc
+            raise self._auth_failure("telnet", exc) from exc
         except Exception as exc:
             logger.debug("Telnet failed for %s: %s", self.hostname, exc)
+            self._attempts.append(f"telnet: {exc}")
             return False
 
     # ------------------------------------------------------------------
@@ -337,7 +373,8 @@ class ProcurveDriver(SwitchDriver):
                 last = ""
                 for cmd in command:
                     last = _do(cmd)
-                    if "% Invalid" not in last and "Error" not in last:
+                    # ProCurve says "Invalid input: …"; other firmware "% Invalid …"
+                    if "Invalid input" not in last and "% Invalid" not in last and "Error" not in last:
                         return last
                 return last
             return _do(command)
@@ -409,14 +446,16 @@ class ProcurveDriver(SwitchDriver):
 
         # CLI path
         sysinfo = self._send_command(
-            ["show system-information", "show system information"]
+            ["show system-information", "show system information", "show system"]
         )
         parsed = parsers.parse_system_info(sysinfo)
 
-        # get model from show version if not in system-info
+        # get model (and part_number) from show version if not in system-info
         if not parsed["model"]:
             ver_out = self._send_command("show version")
-            parsed["model"] = parsers.parse_model_from_version(ver_out)
+            parsed["model"], ver_pn = parsers.parse_model_from_version(ver_out)
+            if ver_pn and not parsed.get("part_number"):
+                parsed["part_number"] = ver_pn
             if not parsed["os_version"]:
                 parsed["os_version"] = parsers.parse_version(ver_out)
 
@@ -425,6 +464,7 @@ class ProcurveDriver(SwitchDriver):
         return {
             "vendor": self.VENDOR,
             "model": parsed["model"],
+            "part_number": parsed.get("part_number", ""),
             "hostname": parsed["hostname"],
             "fqdn": parsed["hostname"],
             "os_version": parsed["os_version"],
@@ -467,7 +507,8 @@ class ProcurveDriver(SwitchDriver):
             except Exception:
                 pass
 
-        return ifaces
+        # The CLI lists member ports only ("3-Trk3"); the trunk gets its own row.
+        return add_lag_interfaces(ifaces)
 
     # ------------------------------------------------------------------
     # NAPALM: get_interfaces_ip
@@ -544,7 +585,18 @@ class ProcurveDriver(SwitchDriver):
     ) -> Dict[str, str]:
         """Return device configuration."""
         if self._transport == "api":
-            return self._api.get_config()
+            result = self._api.get_config()
+            if result.get("running"):
+                return result
+            # REST endpoint not available on this firmware (e.g. HP 2530 / YA series).
+            # HP switches allow only one session per user — close the REST session
+            # first so the switch accepts the SSH connection.
+            try:
+                self._api._session.close()
+            except Exception:
+                pass
+            running = self._get_config_via_ssh()
+            return {"running": running, "startup": "", "candidate": ""}
 
         running = ""
         startup = ""
@@ -559,6 +611,26 @@ class ProcurveDriver(SwitchDriver):
 
         return {"running": running, "startup": startup, "candidate": candidate}
 
+    def _get_config_via_ssh(self) -> str:
+        """Retrieve running-config via netmiko on port 22.
+
+        HP ProCurve switches do not support exec_channel SSH; they require
+        an interactive shell (invoke_shell), which netmiko's hp_procurve
+        device type handles correctly.  Port 22 is used explicitly because
+        self.port is the REST API port (443) in API transport mode.
+        """
+        try:
+            kwargs = self._netmiko_kwargs()
+            kwargs["port"] = 22  # always SSH — self.port is the REST API port
+            conn = ConnectHandler(**kwargs)
+            try:
+                return conn.send_command("show running-config")
+            finally:
+                conn.disconnect()
+        except Exception as exc:
+            logger.warning("SSH config fallback failed for %s: %s", self.hostname, exc)
+            return ""
+
     # ------------------------------------------------------------------
     # NAPALM: get_environment
     # ------------------------------------------------------------------
@@ -867,27 +939,40 @@ class ProcurveDriver(SwitchDriver):
         mode = config.get("mode")
         if mode == "trunk":
             for vid in config.get("trunk_vlans", []):
-                payload = {
-                    "vlan_id": vid,
-                    "port_id": interface,
-                    "port_mode": "POM_TAGGED_STATIC",
-                }
-                resp = self._api.post("vlans-ports", json=payload)
-                if not resp.ok and resp.status_code != 409:
-                    raise ConnectionException(
-                        f"set_interface({interface}): vlans-ports POST HTTP {resp.status_code}"
-                    )
+                self._api_set_port_vlan(interface, vid, "POM_TAGGED_STATIC")
         elif mode == "access":
             if "access_vlan" in config:
-                payload = {
-                    "vlan_id": config["access_vlan"],
-                    "port_id": interface,
-                    "port_mode": "POM_UNTAGGED",
-                }
+                self._api_set_port_vlan(interface, config["access_vlan"], "POM_UNTAGGED")
+
+    # Statuses the switch uses to say "that association is already there".
+    # v7 firmware answers 400 with {"message":"Association exists"}; others
+    # use the more conventional 409.
+    _ASSOCIATION_EXISTS = (400, 409)
+
+    def _api_set_port_vlan(self, interface: str, vid: int, port_mode: str) -> None:
+        """Put *interface* into VLAN *vid* with *port_mode*.
+
+        Creating the membership and changing an existing one are different
+        operations here. A port that already carries the VLAN — untagged, say,
+        while it is meant to become tagged — cannot be POSTed again: the switch
+        refuses the duplicate. The membership is its own resource, named
+        ``{vlan_id}-{port_id}``, and changing it is a PUT.
+        """
+        payload = {"vlan_id": vid, "port_id": interface, "port_mode": port_mode}
         resp = self._api.post("vlans-ports", json=payload)
-                if not resp.ok and resp.status_code != 409:
+        if resp.ok:
+            return
+        if resp.status_code not in self._ASSOCIATION_EXISTS:
             raise ConnectionException(
                 f"set_interface({interface}): vlans-ports POST HTTP {resp.status_code}"
+                f" – {resp.text[:200]}"
+            )
+
+        resp = self._api.put(f"vlans-ports/{vid}-{interface}", json=payload)
+        if not resp.ok:
+            raise ConnectionException(
+                f"set_interface({interface}): vlans-ports/{vid}-{interface} PUT HTTP "
+                f"{resp.status_code} – {resp.text[:200]}"
             )
 
     def _cli_set_interface(self, interface: str, config: InterfaceConfigDict) -> None:
@@ -1033,61 +1118,26 @@ class ProcurveDriver(SwitchDriver):
     # NAPALM: Configuration management (CLI only)
     # ------------------------------------------------------------------
 
+    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:
-        """Load a candidate configuration for merging."""
         if self._transport == "api":
             raise MergeConfigException(
                 "Merge config is not supported via REST API transport. Use SSH."
             )
-        if filename and config:
-            raise MergeConfigException("Provide either filename or config, not both.")
-        if filename:
-            with open(filename) as f:
-                config = f.read()
-        if config is None:
-            raise MergeConfigException("No configuration provided.")
-        self._candidate_config = config
-        self._candidate_mode = "merge"
+        super().load_merge_candidate(filename, config)
 
     def load_replace_candidate(
         self, filename: Optional[str] = None, config: Optional[str] = None
     ) -> None:
-        """Load a candidate configuration for replacing.
-
-        Note: ProCurve does not natively support atomic config replace.
-        The candidate is applied line-by-line (same as merge) when committed.
-        """
         if self._transport == "api":
             raise ReplaceConfigException(
                 "Replace config is not supported via REST API transport. Use SSH."
             )
-        if filename and config:
-            raise ReplaceConfigException("Provide either filename or config, not both.")
-        if filename:
-            with open(filename) as f:
-                config = f.read()
-        if config is None:
-            raise ReplaceConfigException("No configuration provided.")
-        self._candidate_config = config
-        self._candidate_mode = "replace"
-
-    def compare_config(self) -> str:
-        """Return a diff between the running config and the candidate config."""
-        if self._candidate_config is None:
-            return ""
-        import difflib
-
-        running = self._send_command("show running-config")
-        diff = difflib.unified_diff(
-            running.splitlines(),
-            self._candidate_config.splitlines(),
-            fromfile="running-config",
-            tofile="candidate-config",
-            lineterm="",
-        )
-        return "\n".join(diff)
+        super().load_replace_candidate(filename, config)
 
     def commit_config(self, message: str = "", revert_in: Optional[int] = None) -> None:
         """Apply the candidate configuration to the device."""
@@ -1114,24 +1164,6 @@ class ProcurveDriver(SwitchDriver):
         self._candidate_config = None
         self._candidate_mode = None
 
-    def discard_config(self) -> None:
-        """Discard the loaded candidate configuration."""
-        self._candidate_config = None
-        self._candidate_mode = None
-
-    def rollback(self) -> None:
-        """Re-apply the backup configuration saved before the last commit."""
-        if self._backup_config is None:
-            raise MergeConfigException("No backup config available for rollback.")
-        self._candidate_config = self._backup_config
-        self._candidate_mode = "merge"
-        self.commit_config()
-        self._backup_config = None
-
-    def has_pending_commit(self) -> bool:
-        """Return True if a candidate config is loaded."""
-        return self._candidate_config is not None
-
     # ── SNMP / Health ──────────────────────────────────────────────────────────
 
     def get_device_warnings(self) -> list:

pyproject.toml

@@ -27,6 +27,7 @@ classifiers = [
 dependencies = [
     "napalm>=4.0.0",
     "netmiko>=4.0.0",
+    "paramiko>=5.0.0",  # CVE-2026-44405
     "netaddr",
     "requests>=2.25.0",
     "urllib3",

tests/unit/test_driver.py

@@ -3,7 +3,9 @@
 import pytest
 from unittest.mock import MagicMock, patch
 
-from napalm_procurve.procurve import ProcurveDriver, _parse_ping_output
+from napalm.base.exceptions import ConnectionException
+
+from napalm_procurve.procurve import _SSH_DISABLED_STANDARD, ProcurveDriver, _parse_ping_output
 from napalm_procurve import parsers
 
 
@@ -43,6 +45,30 @@ Status and Counters - General System Information
   Base MAC Addr   : aabbcc-ddeeff
 """
 
+# J.15.09 firmware (#3/#4): no model anywhere, uptime in minutes.
+SHOW_SYSTEM_J15 = """\
+ Status and Counters - General System Information
+
+  System Name        : myswitch
+  System Contact     :
+  System Location    :
+
+  MAC Age Time (sec) : 300
+
+  Time Zone          : 120
+  Daylight Time Rule : None
+
+
+  Software revision  : J.15.09.0028         Base MAC Addr      : a1b2c3-d4e5f6
+  ROM Version        : J.14.05              Serial Number      : SG12345678
+  Allow V1 Modules   : Yes
+
+  Up Time            : 5 mins               Memory   - Total   : 58,720,256
+  CPU Util (%)       : 98                              Free    : 39,550,272
+"""
+
+SHOW_SYSTEM_INFORMATION_INVALID = "Invalid input: system-information"
+
 SHOW_VERSION_2520G = """\
  HP J9565A 2520G-8-PoE Switch
   Software revision  : R.11.27
@@ -59,6 +85,25 @@ Status and Counters - Port Status
   3      100/1000T    | No       Down    Auto   Unknown      off   0
 """
 
+# Layout with the Intrusion Alert column; trunk members are "<port>-Trk<n>" (#3).
+SHOW_INTERFACES_BRIEF_INTRUSION = """\
+ Status and Counters - Port Status
+
+                   | Intrusion                           MDI  Flow Bcast
+  Port   Type      | Alert     Enabled Status Mode       Mode Ctrl Limit
+  ------ --------- + --------- ------- ------ ---------- ---- ---- -----
+  1      100/1000T | No        Yes     Up     1000FDx    MDI  on   0
+  2      100/1000T | No        Yes     Down   1000FDx    MDI  off  0
+  3-Trk3 100/1000T | No        Yes     Down   1000FDx    MDI  off  0
+  4-Trk3 100/1000T | No        Yes     Down   1000FDx    MDI  off  0
+  5      100/1000T | No        Yes     Up     1000FDx    MDI  on   0
+  6-Trk6 100/1000T | No        No      Down   1000FDx    NA   off  0
+  7-Trk6 100/1000T | No        Yes     Up     1000FDx    MDI  off  0
+  8      100/1000T | No        Yes     Up     1000FDx    MDIX off  0
+  9      100/1000T | No        Yes     Down   1000FDx    MDIX off  0
+  10     1000SX    | No        Yes     Up     1000FDx    NA   off  0
+"""
+
 SHOW_INTERFACES_PORT = """\
 Status and Counters - Port Counters for port 1
 
@@ -212,6 +257,13 @@ class TestParseSystemInfo:
         assert info["os_version"] == "YA.16.04.0006"
         assert info["serial_number"] == "SG87654321"
 
+    def test_j15(self):
+        info = parsers.parse_system_info(SHOW_SYSTEM_J15)
+        assert info["hostname"] == "myswitch"
+        assert info["os_version"] == "J.15.09.0028"
+        assert info["serial_number"] == "SG12345678"
+        assert info["base_mac"] == "a1:b2:c3:d4:e5:f6"
+
 
 class TestParseInterfacesBrief:
     def test_parses_ports(self):
@@ -234,6 +286,30 @@ class TestParseInterfacesBrief:
     def test_port_3_disabled(self):
         ifaces = parsers.parse_interfaces_brief(SHOW_INTERFACES_BRIEF)
         assert ifaces["3"]["is_enabled"] is False
+        assert "trunk_group" not in ifaces["3"]
+
+
+class TestParseInterfacesBriefIntrusionAlert:
+    """The layout with an Intrusion Alert column before Enabled (#3)."""
+
+    def test_parses_every_port(self):
+        ifaces = parsers.parse_interfaces_brief(SHOW_INTERFACES_BRIEF_INTRUSION)
+        assert sorted(ifaces, key=int) == [str(n) for n in range(1, 11)]
+
+    def test_alert_column_is_not_read_as_enabled(self):
+        ifaces = parsers.parse_interfaces_brief(SHOW_INTERFACES_BRIEF_INTRUSION)
+        assert ifaces["1"]["is_enabled"] is True
+        assert ifaces["1"]["is_up"] is True
+        assert ifaces["1"]["speed"] == 1000.0
+        assert ifaces["2"]["is_up"] is False
+        assert ifaces["6"]["is_enabled"] is False
+
+    def test_trunk_suffix_becomes_trunk_group(self):
+        ifaces = parsers.parse_interfaces_brief(SHOW_INTERFACES_BRIEF_INTRUSION)
+        assert ifaces["3"]["trunk_group"] == "Trk3"
+        assert ifaces["4"]["trunk_group"] == "Trk3"
+        assert ifaces["7"]["trunk_group"] == "Trk6"
+        assert "trunk_group" not in ifaces["1"]
 
 
 class TestParseArpTable:
@@ -349,6 +425,40 @@ class TestDriverGetInterfaces:
         assert "1" in ifaces
         assert ifaces["1"]["is_up"] is True
 
+    def test_trunk_member_detail_uses_bare_port_name(self, driver):
+        """`show interfaces 3-Trk3` is not a command; the port is `3` (#3)."""
+        driver._send_command = MagicMock(return_value=SHOW_INTERFACES_BRIEF_INTRUSION)
+        driver.get_interfaces()
+        sent = [c.args[0] for c in driver._send_command.call_args_list]
+        assert "show interfaces 3" in sent
+        assert not any("Trk" in c for c in sent)
+
+
+class TestSendCommandAlternatives:
+    """A list of commands falls through to the next on a CLI error (#4)."""
+
+    def test_invalid_input_tries_next_command(self, driver):
+        outputs = {
+            "show system-information": SHOW_SYSTEM_INFORMATION_INVALID,
+            "show system information": SHOW_SYSTEM_J15,
+        }
+        driver._device.send_command.side_effect = lambda cmd, **kw: outputs[cmd]
+        out = driver._send_command(["show system-information", "show system information"])
+        assert out == SHOW_SYSTEM_J15.strip()
+
+    def test_facts_on_j15_firmware(self, driver):
+        outputs = {
+            "show system-information": SHOW_SYSTEM_INFORMATION_INVALID,
+            "show system information": SHOW_SYSTEM_J15,
+            "show interfaces brief": SHOW_INTERFACES_BRIEF_INTRUSION,
+            "show version": "Image stamp:    /ws/swbuildm/J_rel/code/build\n                J.15.09.0028",
+        }
+        driver._device.send_command.side_effect = lambda cmd, **kw: outputs[cmd]
+        facts = driver.get_facts()
+        assert facts["os_version"] == "J.15.09.0028"
+        assert facts["serial_number"] == "SG12345678"
+        assert len(facts["interface_list"]) == 10
+
 
 class TestDriverGetArpTable:
     def test_arp_from_cli(self, driver):
@@ -425,6 +535,60 @@ class TestDriverTransportDetection:
                 with pytest.raises(Exception):
                     drv.open()
 
+    def test_auth_failure_names_why_earlier_transports_failed(self):
+        """A Telnet login failure reports the transport and the earlier failures.
+
+        Otherwise "Login failed" reads as a wrong password when SSH was merely
+        refused and Telnet was the only transport left to answer (#2).
+        """
+        from netmiko.exceptions import NetmikoAuthenticationException, NetmikoTimeoutException
+
+        def connect(**kwargs):
+            if kwargs["device_type"] == ProcurveDriver.NETMIKO_DEVICE_TYPE_TELNET:
+                raise NetmikoAuthenticationException("Login failed: 192.168.0.1")
+            if kwargs["disabled_algorithms"] == _SSH_DISABLED_STANDARD:
+                raise ConnectionRefusedError("[Errno 111] Connection refused")
+            raise NetmikoTimeoutException("TCP connection to device failed")
+
+        with patch("napalm_procurve.procurve.ProcurveApiClient.probe", return_value=(None, None)):
+            with patch("napalm_procurve.procurve.ConnectHandler", side_effect=connect):
+                drv = ProcurveDriver("192.168.0.1", "manager", "secret")
+                with pytest.raises(ConnectionException) as exc_info:
+                    drv.open()
+
+        msg = str(exc_info.value)
+        assert "Authentication failed for 192.168.0.1 via telnet: Login failed: 192.168.0.1" in msg
+        assert "api: not detected" in msg
+        assert "ssh: [Errno 111] Connection refused" in msg
+        assert "ssh_legacy: timed out" in msg
+
+    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")):
+                drv = ProcurveDriver("192.168.0.1", "manager", "secret")
+                with pytest.raises(ConnectionException) as exc_info:
+                    drv.open()
+
+        msg = str(exc_info.value)
+        assert "api: not detected" in msg
+        assert "ssh: no route to host" in msg
+        assert "ssh_legacy: no route to host" in msg
+        assert "telnet: no route to host" in msg
+
+    def test_reopen_does_not_carry_earlier_attempts(self):
+        """Each open() reports only its own attempts."""
+        with patch("napalm_procurve.procurve.ProcurveApiClient.probe", return_value=(None, None)):
+            with patch("napalm_procurve.procurve.ConnectHandler", side_effect=OSError("first")):
+                drv = ProcurveDriver("192.168.0.1", "manager", "secret")
+                with pytest.raises(ConnectionException):
+                    drv.open()
+            with patch("napalm_procurve.procurve.ConnectHandler", side_effect=OSError("second")):
+                with pytest.raises(ConnectionException) as exc_info:
+                    drv.open()
+
+        assert "first" not in str(exc_info.value)
+
 
 # ===========================================================================
 # VLAN parser tests
@@ -651,3 +815,127 @@ class TestDriverGetVlans:
         # Ports 1-4 are untagged in VLAN 1
         assert pvids["1"] == 1
         assert pvids["4"] == 1
+
+
+# ===========================================================================
+# set_interface over the REST API — changing an existing VLAN membership
+# ===========================================================================
+
+
+def _api_driver(post_status: int = 400, post_body: str = '{"message":"Association exists"}'):
+    """Driver on the REST transport with a scripted API client."""
+    with patch("napalm_procurve.procurve.ConnectHandler"):
+        drv = ProcurveDriver(hostname="192.168.0.1", username="manager", password="secret")
+    drv._transport = "api"
+    api = MagicMock()
+    post_resp = MagicMock(ok=post_status < 400, status_code=post_status, text=post_body)
+    api.post.return_value = post_resp
+    api.put.return_value = MagicMock(ok=True, status_code=200, text="{}")
+    drv._api = api
+    return drv, api
+
+
+class TestApiSetInterfaceExistingMembership:
+    """A port that already carries the VLAN needs its mode changed, not a new row.
+
+    Reproduced on a 2530-24G-PoEP (REST v7): port 10 holds VLAN 10 untagged
+    alongside 30/40/50 tagged. Posting the tagged membership answers
+    `400 {"message":"Association exists"}` — only `409` was treated as
+    "already there". The resource is `{vlan_id}-{port_id}` and takes a PUT.
+    """
+
+    def test_trunk_falls_back_to_put_on_association_exists(self):
+        drv, api = _api_driver(post_status=400)
+        drv.set_interface("10", {"mode": "trunk", "trunk_vlans": [10]})
+        api.put.assert_called_once()
+        path = api.put.call_args[0][0]
+        assert path == "vlans-ports/10-10"
+        assert api.put.call_args[1]["json"]["port_mode"] == "POM_TAGGED_STATIC"
+
+    def test_access_falls_back_to_put_on_association_exists(self):
+        drv, api = _api_driver(post_status=400)
+        drv.set_interface("10", {"mode": "access", "access_vlan": 20})
+        assert api.put.call_args[0][0] == "vlans-ports/20-10"
+        assert api.put.call_args[1]["json"]["port_mode"] == "POM_UNTAGGED"
+
+    def test_conflict_status_also_falls_back(self):
+        """Other firmware answers 409 for the same situation."""
+        drv, api = _api_driver(post_status=409)
+        drv.set_interface("10", {"mode": "trunk", "trunk_vlans": [10]})
+        api.put.assert_called_once()
+
+    def test_a_new_membership_still_uses_post_alone(self):
+        drv, api = _api_driver(post_status=200)
+        drv.set_interface("11", {"mode": "trunk", "trunk_vlans": [30]})
+        api.post.assert_called_once()
+        api.put.assert_not_called()
+
+    def test_a_failing_put_reports_the_response_body(self):
+        """The message used to drop it, so "Association exists" never surfaced."""
+        drv, api = _api_driver(post_status=400)
+        api.put.return_value = MagicMock(ok=False, status_code=403, text='{"message":"denied"}')
+        with pytest.raises(Exception) as exc:
+            drv.set_interface("10", {"mode": "trunk", "trunk_vlans": [10]})
+        assert "denied" in str(exc.value)
+
+    def test_an_unrelated_post_failure_reports_the_response_body(self):
+        drv, api = _api_driver(post_status=500, post_body='{"message":"boom"}')
+        with pytest.raises(Exception) as exc:
+            drv.set_interface("10", {"mode": "trunk", "trunk_vlans": [10]})
+        assert "boom" in str(exc.value)
+        api.put.assert_not_called()
+
+
+# ===========================================================================
+# LAG rows (#5)
+# ===========================================================================
+
+
+class TestApiGetInterfacesLag:
+    """The REST path's trunk rows, kept as they were when it moved to add_lag_interfaces."""
+
+    def _client(self, ports):
+        from napalm_procurve.api_client import ProcurveApiClient
+
+        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
+            ]},
+            "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"},
+        ])
+        ifaces = client.get_interfaces()
+
+        assert ifaces["trk3"]["lag_members"] == ["3", "4"]
+        assert ifaces["trk3"]["lag_mode"] == "lacp"
+        assert ifaces["trk3"]["is_up"] is True
+        assert ifaces["trk3"]["speed"] == 2000.0
+        assert ifaces["trk6"]["lag_members"] == ["7", "10"]
+        assert ifaces["trk6"]["lag_mode"] == "trunk"
+        assert ifaces["trk6"]["description"] == "LAG (7, 10)"
+        assert ifaces["3"]["trunk_group"] == "trk3"
+
+
+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 ""
+        )
+        ifaces = driver.get_interfaces()
+
+        assert ifaces["Trk3"]["lag_members"] == ["3", "4"]
+        assert ifaces["Trk6"]["lag_members"] == ["6", "7"]
+        assert ifaces["Trk6"]["is_up"] is True
+        assert ifaces["Trk6"]["is_enabled"] is True