Merge pull request 'feat!: report the subscription as a license, not a warning (0.4.0)' (#26) from feat/facts-not-warnings into master

CHANGELOG.md

@@ -8,6 +8,14 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 ## [Unreleased]
 
 ### Changed
+- **Breaking:** `get_device_warnings` is gone; the node reports facts and the caller
+  names the warnings (netOrk #1111). lldpd and the pending updates were already facts
+  (`get_packages`, `get_available_updates`). The subscription is now
+  `get_licenses()` (napalm-device-types 4.14 `LicenseMixin`): one `pve-subscription`
+  entry with the API's status as a `LicenseStatus` (`notfound` → `none`, `new` and
+  `suspended` → `unknown`), its message as `detail`, product name and due date when
+  subscribed. The old check compared against `NotFound` while the API answers
+  `notfound`, so it never fired. Requires napalm-device-types 4.14.0; version 0.4.0.
 - **Breaking:** `run_device_action` (`fix_snmp`, `install_lldpd`) is gone. The driver
   implements napalm-device-types 4.13's fix templates: `configure_snmp_agent(settings)`
   with the caller's community and allowed sources (one `rocommunity` line each, ufw only

napalm_proxmox/driver.py

@@ -40,6 +40,7 @@ from napalm_device_types import (
     HostStatusMixin,
     HypervisorDriver,
     KernelFactsMixin,
+    LicenseMixin,
     ListeningSocketsMixin,
     PortSpec,
     SnmpAgentMixin,
@@ -92,6 +93,7 @@ class ProxmoxDriver(
     SystemIdentityMixin,
     SystemdServicesMixin,
     HostStatusMixin,
+    LicenseMixin,
     SnmpAgentMixin,
     EnsurePackageMixin,
     HypervisorDriver,

napalm_proxmox/system_mixin.py

@@ -30,6 +30,15 @@ logger = logging.getLogger(__name__)
 
 _JsonDict = dict[str, Any]
 
+#: The subscription API's status (lower-cased) as a napalm-device-types
+#: ``LicenseStatus``; "new" and "suspended" have none of their own.
+_SUBSCRIPTION_STATUS = {
+    "notfound": "none",
+    "active": "active",
+    "invalid": "invalid",
+    "expired": "expired",
+}
+
 
 _PACKAGE_NAME = re.compile(r"^[a-zA-Z0-9][a-zA-Z0-9_.+:~-]*$")
 #: What may stand in an snmpd.conf value: no whitespace, no quote, no comment.
@@ -344,51 +353,29 @@ class ProxmoxSystemMixin:
         return result
 
     # ------------------------------------------------------------------ #
-    # Device warnings
+    # Licenses (napalm-device-types' LicenseMixin)
     # ------------------------------------------------------------------ #
 
-    def get_device_warnings(self) -> list[_JsonDict]:
-        """Return warnings for the Proxmox node.
+    def get_licenses(self) -> list[_JsonDict]:
+        """The node's subscription, as the API reports it.
 
-        Currently detects:
-        - lldpd not installed (LLDP neighbor discovery unavailable)
-        - Available package updates (via Proxmox APT API)
-        - Missing / invalid subscription
+        A fact for the caller: whether running without one is a problem is not
+        the driver's to say. An unreadable subscription reports nothing.
         """
-        warnings: list[_JsonDict] = []
-
-        # 0. LLDP daemon
-        try:
-            lldpd_path = self._exec_ssh_command("command -v lldpd 2>/dev/null").strip()
-            if not lldpd_path:
-                warnings.append({"code": "lldpd_not_installed"})
-        except Exception as exc:
-            logger.debug("Failed to check for lldpd: %s", exc)
-
-        # 1. Available package updates
-        try:
-            updates = self.get_available_updates()
-            if updates:
-                warnings.append({
-                    "code": "updates_available",
-                    "meta": {
-                        "count": len(updates),
-                        "packages": [u["name"] for u in updates],
-                    },
-                })
-        except Exception as exc:
-            logger.debug("Failed to check available updates: %s", exc)
-
-        # 2. Subscription status
-        try:
         sub = self._get_node_subscription()
-            status = sub.get("status", "")
-            if status in ("NotFound", "Invalid", "Expired"):
-                warnings.append({"code": "no_subscription", "meta": {"status": status}})
-        except Exception as exc:
-            logger.debug("Failed to check subscription status: %s", exc)
-
-        return warnings
+        raw = str(sub.get("status", ""))
+        if not raw:
+            return []
+        lic: _JsonDict = {
+            "name": "pve-subscription",
+            "status": _SUBSCRIPTION_STATUS.get(raw.lower(), "unknown"),
+            "detail": sub.get("message") or raw,
+        }
+        if sub.get("productname"):
+            lic["edition"] = sub["productname"]
+        if sub.get("nextduedate"):
+            lic["expires"] = sub["nextduedate"]
+        return [lic]
 
     # ------------------------------------------------------------------ #
     # Services (systemd, through napalm-device-types' SystemdServicesMixin)

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "napalm-proxmox"
-version = "0.3.0"
+version = "0.4.0"
 description = "NAPALM driver for Proxmox VE (classic Linux networking, SDN, OVS)"
 readme = "README.md"
 license = { text = "Apache-2.0" }
@@ -25,7 +25,7 @@ classifiers = [
 requires-python = ">=3.9"
 dependencies = [
     "napalm>=5.0.0",
-    "napalm_device_types>=4.13.0",
+    "napalm_device_types>=4.14.0",
     "paramiko>=5.0.0",  # CVE-2026-44405; imported directly for SSH fallback (driver.py)
     "proxmoxer>=2.0.0",
     "netaddr>=0.9.0",

tests/test_licenses.py

@@ -0,0 +1,62 @@
+"""The node's subscription as a fact (netOrk #1111).
+
+The driver named ``no_subscription`` itself, comparing against "NotFound" while
+the API answers "notfound" -- so it never fired. It reports the subscription
+now; whether a missing one is a warning is the caller's decision.
+"""
+
+from __future__ import annotations
+
+import pytest
+
+from napalm_proxmox import ProxmoxDriver
+
+
+def _driver(subscription: dict) -> ProxmoxDriver:
+    d = ProxmoxDriver("pve", "root@pam", "pw")
+    d._get_node_subscription = lambda: subscription
+    return d
+
+
+@pytest.mark.parametrize(
+    "raw, status",
+    [
+        ("notfound", "none"),
+        ("NotFound", "none"),
+        ("active", "active"),
+        ("invalid", "invalid"),
+        ("expired", "expired"),
+        ("new", "unknown"),
+        ("suspended", "unknown"),
+    ],
+)
+def test_the_api_status_as_a_license_status(raw, status):
+    [lic] = _driver({"status": raw, "message": "There is no subscription key"}).get_licenses()
+
+    assert lic["name"] == "pve-subscription"
+    assert lic["status"] == status
+    assert lic["detail"] == "There is no subscription key"
+
+
+def test_level_and_due_date_when_subscribed():
+    sub = {
+        "status": "active",
+        "productname": "Proxmox VE Community Subscription 1 CPU/year",
+        "nextduedate": "2027-03-01",
+    }
+    [lic] = _driver(sub).get_licenses()
+
+    assert lic["edition"] == "Proxmox VE Community Subscription 1 CPU/year"
+    assert lic["expires"] == "2027-03-01"
+
+
+def test_an_unreadable_subscription_reports_nothing():
+    """``_get_node_subscription`` gives {} when the API call fails: no fact, not "none"."""
+    assert _driver({}).get_licenses() == []
+
+
+def test_no_warnings_of_its_own():
+    """lldpd, the updates and the subscription are facts the caller reads
+    (get_packages, get_available_updates, get_licenses)."""
+    assert not any("get_device_warnings" in vars(c) for c in ProxmoxDriver.__mro__[:-1]
+                   if c.__module__.startswith("napalm_proxmox"))