Merge pull request 'feat: declare the SNMP health OIDs instead of collecting them (0.2.0)' (#5) from feat/snmp-health-profile into main

napalm_tplink_jetstream/tplink_jetstream.py

@@ -21,7 +21,7 @@ from __future__ import annotations
 
 import re
 import socket
-from typing import Any
+from typing import Any, ClassVar
 
 import netaddr
 from netmiko import ConnectHandler
@@ -64,6 +64,15 @@ class TPLinkJetstreamDriver(ConfigLifecycleMixin, SwitchDriver):
     # Netmiko device type for TP-Link Jetstream
     NETMIKO_DEVICE_TYPE = "tplink_jetstream"
 
+    #: Where the switch publishes its health values over SNMP
+    #: (napalm-device-types ``SnmpHealthProfile``); the application polls.
+    #: CPU and memory are tables with one row per stack unit.
+    SNMP_HEALTH_PROFILE: ClassVar[dict[str, Any]] = {
+        "uptime": {"oids": ["1.3.6.1.4.1.11863.6.1.1.9.0"]},
+        "cpu": {"table": "1.3.6.1.4.1.11863.6.4.1.1.1.1.3", "fold": "first"},
+        "memory": {"percent": {"table": "1.3.6.1.4.1.11863.6.4.1.2.1.1.2", "fold": "first"}},
+    }
+
     # PoE priority mapping between the CLI's "low/high/critical" and the
     # generic priority strings used across drivers.
     _POE_PRIORITY_MAP = {"low": "PPP_LOW", "high": "PPP_HIGH", "critical": "PPP_CRITICAL"}
@@ -2004,56 +2013,3 @@ class TPLinkJetstreamDriver(ConfigLifecycleMixin, SwitchDriver):
             lines.append(f"[warn] Verification failed: {out[:100]}")
 
         return {"success": success, "output": "\n".join(lines)}
-
-    # ------------------------------------------------------------------
-    # Health metrics (SNMP)
-    # ------------------------------------------------------------------
-
-    @classmethod
-    async def get_health_metrics(cls, snmp_get, snmp_walk) -> dict:
-        import asyncio
-        from napalm_device_types._ucd_metrics import build_if_metrics, ticks_to_seconds
-
-        _TPLINK_SYS_UPTIME = "1.3.6.1.4.1.11863.6.1.1.9.0"
-        _TPLINK_CPU_1MIN   = "1.3.6.1.4.1.11863.6.4.1.1.1.1.3"
-        _TPLINK_MEM_UTIL   = "1.3.6.1.4.1.11863.6.4.1.2.1.1.2"
-        _OID_IF_DESCR      = "1.3.6.1.2.1.2.2.1.2"
-        _OID_IF_SPEED      = "1.3.6.1.2.1.2.2.1.5"
-        _OID_IF_IN_OCT     = "1.3.6.1.2.1.2.2.1.10"
-        _OID_IF_OUT_OCT    = "1.3.6.1.2.1.2.2.1.16"
-        _OID_IF_IN_ERR     = "1.3.6.1.2.1.2.2.1.14"
-        _OID_IF_OUT_ERR    = "1.3.6.1.2.1.2.2.1.20"
-
-        (tp_uptime, cpu_rows, mem_rows,
-         descr, speed, in_oct, out_oct, in_err, out_err) = await asyncio.gather(
-            snmp_get(_TPLINK_SYS_UPTIME),
-            snmp_walk(_TPLINK_CPU_1MIN),
-            snmp_walk(_TPLINK_MEM_UTIL),
-            snmp_walk(_OID_IF_DESCR),
-            snmp_walk(_OID_IF_SPEED),
-            snmp_walk(_OID_IF_IN_OCT),
-            snmp_walk(_OID_IF_OUT_OCT),
-            snmp_walk(_OID_IF_IN_ERR),
-            snmp_walk(_OID_IF_OUT_ERR),
-        )
-
-        metrics: dict = {}
-
-        secs = ticks_to_seconds(tp_uptime)
-        if secs is not None:
-            metrics["uptime_seconds"] = secs
-
-        if cpu_rows:
-            try:
-                metrics["cpu_percent"] = float(next(iter(cpu_rows.values())))
-            except (ValueError, StopIteration):
-                pass
-
-        if mem_rows:
-            try:
-                metrics["memory_percent"] = float(next(iter(mem_rows.values())))
-            except (ValueError, StopIteration):
-                pass
-
-        build_if_metrics(metrics, descr, speed, in_oct, out_oct, in_err, out_err)
-        return metrics

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "napalm-tplink-jetstream"
-version = "0.1.2"
+version = "0.2.0"
 description = "NAPALM driver for TP-Link Jetstream managed switches"
 readme = "README.md"
 license = { text = "Apache-2.0" }

tests/unit/test_snmp_profile.py

@@ -0,0 +1,25 @@
+"""The switch's SNMP health OIDs, declared as data (napalm-device-types docs/STRUCTURE.md).
+
+The driver does not collect over SNMP; the application polls with this profile.
+"""
+
+from napalm_tplink_jetstream.tplink_jetstream import TPLinkJetstreamDriver
+
+PROFILE = TPLinkJetstreamDriver.SNMP_HEALTH_PROFILE
+
+
+def test_the_driver_does_not_collect_itself():
+    assert "get_health_metrics" not in vars(TPLinkJetstreamDriver)
+
+
+def test_uptime_in_timeticks_from_the_tp_link_mib():
+    # swt-l2-3-38 answered 129434122 (2026-10-16): TimeTicks, about 15 days.
+    assert PROFILE["uptime"] == {"oids": ["1.3.6.1.4.1.11863.6.1.1.9.0"]}
+
+
+def test_cpu_and_memory_per_unit_take_the_first_row():
+    # One row per stack unit; a single switch answers row 1 only.
+    assert PROFILE["cpu"] == {"table": "1.3.6.1.4.1.11863.6.4.1.1.1.1.3", "fold": "first"}
+    assert PROFILE["memory"] == {
+        "percent": {"table": "1.3.6.1.4.1.11863.6.4.1.2.1.1.2", "fold": "first"}
+    }