feat(hypervisor): generalize create_vm_from_cloud_init interface to support arbitrary NIC configs

- Replace fixed mgmt/capture dual-NIC parameters with generic NICConfigDict list
- Add optional disk_resize_gb parameter for post-clone disk expansion
- Update docstrings to reflect generic NIC approach (primary NIC concept)
- Add NICConfigDict TypedDict supporting access VLAN, trunk VLAN, and DHCP flags

README.md

@@ -38,6 +38,7 @@ Requires Python ≥ 3.9 and NAPALM ≥ 4.0.
 | `SwitchDriver` | Ethernet switches | Cisco IOS, Arista EOS, Juniper EX |
 | `FirewallDriver` | Firewalls & UTM appliances | pfSense, Fortinet FortiOS, Cisco ASA |
 | `HypervisorDriver` | Hypervisors & virtualisation platforms | Proxmox VE, VMware ESXi, KVM/libvirt |
+| `OSDriver` | General-purpose operating systems | Linux, BSD, macOS |
 | `StorageDriver` | Storage appliances & NAS/SAN | TrueNAS, Synology DSM, QNAP QTS |
 
 ## Usage
@@ -105,6 +106,34 @@ class ProxmoxDriver(HypervisorDriver):
         ...
 ```
 
+### OS / Linux
+
+```python
+from napalm_device_types import OSDriver
+
+class LinuxDriver(OSDriver):
+
+    def get_packages(self):
+        # return List[PackageDict]
+        ...
+
+    def get_services(self):
+        # return List[ServiceDict]
+        ...
+
+    def get_users(self):
+        # return List[UserDict]
+        ...
+
+    def get_processes(self):
+        # return List[ProcessDict]
+        ...
+
+    def get_cron_jobs(self):
+        # return List[CronJobDict]
+        ...
+```
+
 ### Storage / NAS
 
 ```python

napalm_device_types/__init__.py

@@ -19,19 +19,37 @@ Available base classes:
 * :class:`~napalm_device_types.switch.SwitchDriver`
 * :class:`~napalm_device_types.firewall.FirewallDriver`
 * :class:`~napalm_device_types.hypervisor.HypervisorDriver`
+* :class:`~napalm_device_types.os.OSDriver`
 * :class:`~napalm_device_types.storage.StorageDriver`
+* :class:`~napalm_device_types.residential_gateway.ResidentialGatewayDriver`
+
+Also provided:
+
+* :class:`~napalm_device_types.config_lifecycle.ConfigLifecycleMixin` --
+  stand-alone mixin to reduce duplication of config lifecycle methods across
+  drivers.
 """
 
+from napalm_device_types.base import DeviceTypeDriver, FingerprintRule, PortSpec
 from napalm_device_types.access_point import AccessPointDriver
+from napalm_device_types.config_lifecycle import ConfigLifecycleMixin
 from napalm_device_types.firewall import FirewallDriver
 from napalm_device_types.hypervisor import HypervisorDriver
+from napalm_device_types.os import OSDriver
+from napalm_device_types.residential_gateway import ResidentialGatewayDriver
 from napalm_device_types.storage import StorageDriver
 from napalm_device_types.switch import SwitchDriver
 
 __all__ = [
     "AccessPointDriver",
+    "ConfigLifecycleMixin",
+    "DeviceTypeDriver",
+    "FingerprintRule",
     "FirewallDriver",
     "HypervisorDriver",
+    "OSDriver",
+    "PortSpec",
+    "ResidentialGatewayDriver",
     "StorageDriver",
     "SwitchDriver",
 ]

napalm_device_types/_ucd_metrics.py

@@ -0,0 +1,165 @@
+"""Shared SNMP metric collection helpers for UCD-MIB and IF-MIB."""
+from __future__ import annotations
+
+import asyncio
+import re
+from typing import Awaitable, Callable, Dict, Optional
+
+# ── UCD-MIB OIDs ──────────────────────────────────────────────────────────────
+OID_SYS_UPTIME  = "1.3.6.1.2.1.25.1.1.0"      # hrSystemUptime (centiseconds)
+OID_CPU_IDLE    = "1.3.6.1.4.1.2021.11.11.0"   # UCD ssCpuIdle (%)
+OID_MEM_TOTAL   = "1.3.6.1.4.1.2021.4.5.0"     # memTotalReal (kB)
+OID_MEM_FREE    = "1.3.6.1.4.1.2021.4.6.0"     # memAvailReal (kB)
+OID_MEM_BUFFER  = "1.3.6.1.4.1.2021.4.14.0"    # memBuffer (kB)
+OID_MEM_CACHED  = "1.3.6.1.4.1.2021.4.15.0"    # memCached (kB)
+OID_SWAP_TOTAL  = "1.3.6.1.4.1.2021.4.3.0"     # memTotalSwap (kB)
+OID_SWAP_AVAIL  = "1.3.6.1.4.1.2021.4.4.0"     # memAvailSwap (kB)
+OID_LOAD_1      = "1.3.6.1.4.1.2021.10.1.3.1"  # laLoad 1-min
+OID_LOAD_5      = "1.3.6.1.4.1.2021.10.1.3.2"  # laLoad 5-min
+OID_LOAD_15     = "1.3.6.1.4.1.2021.10.1.3.3"  # laLoad 15-min
+
+# ── IF-MIB OIDs ───────────────────────────────────────────────────────────────
+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"
+
+IF_SKIP_DEFAULT = re.compile(r'^(lo|sit\d|tun\d|docker|veth|br-|virbr)', re.IGNORECASE)
+
+SnmpGetFn  = Callable[[str], Awaitable[Optional[str]]]
+SnmpWalkFn = Callable[[str], Awaitable[Dict[str, str]]]
+
+
+def ticks_to_seconds(raw: str | None) -> int | None:
+    if raw is None:
+        return None
+    try:
+        return int(raw) // 100
+    except (ValueError, TypeError):
+        return None
+
+
+def build_if_metrics(
+    metrics: dict,
+    descr: Dict[str, str],
+    speed: Dict[str, str],
+    in_oct: Dict[str, str],
+    out_oct: Dict[str, str],
+    in_err: Dict[str, str],
+    out_err: Dict[str, str],
+    *,
+    if_skip: re.Pattern | None = None,
+    tx_err_is_drop: bool = False,
+) -> None:
+    if if_skip is None:
+        if_skip = IF_SKIP_DEFAULT
+    interfaces: dict = {}
+    for idx, name in descr.items():
+        if if_skip.match(name):
+            continue
+        iface: dict = {"name": name}
+        try:
+            iface["rx_bytes"]  = int(in_oct.get(idx, 0)  or 0)
+            iface["tx_bytes"]  = int(out_oct.get(idx, 0) or 0)
+            iface["rx_errors"] = int(in_err.get(idx, 0)  or 0)
+            tx_val = int(out_err.get(idx, 0) or 0)
+            if tx_err_is_drop:
+                iface["tx_errors"] = 0
+                iface["tx_drops"]  = tx_val
+            else:
+                iface["tx_errors"] = tx_val
+            spd = speed.get(idx)
+            iface["speed_mbps"] = int(spd) // 1_000_000 if spd and int(spd) > 0 else 0
+        except (ValueError, TypeError):
+            pass
+        interfaces[name] = iface
+    if interfaces:
+        metrics["interfaces"] = interfaces
+
+
+async def collect_ucd_metrics(
+    snmp_get: SnmpGetFn,
+    snmp_walk: SnmpWalkFn,
+    *,
+    tx_err_is_drop: bool = False,
+    if_skip: re.Pattern | None = None,
+) -> dict:
+    """Collect UCD-MIB system metrics + IF-MIB interface counters in parallel."""
+    (
+        uptime_raw, cpu_idle_raw,
+        mem_total_raw, mem_free_raw, mem_buf_raw, mem_cache_raw,
+        swap_total_raw, swap_avail_raw,
+        load1_raw, load5_raw, load15_raw,
+    ), (descr, speed, in_oct, out_oct, in_err, out_err) = await asyncio.gather(
+        asyncio.gather(
+            snmp_get(OID_SYS_UPTIME),
+            snmp_get(OID_CPU_IDLE),
+            snmp_get(OID_MEM_TOTAL),
+            snmp_get(OID_MEM_FREE),
+            snmp_get(OID_MEM_BUFFER),
+            snmp_get(OID_MEM_CACHED),
+            snmp_get(OID_SWAP_TOTAL),
+            snmp_get(OID_SWAP_AVAIL),
+            snmp_get(OID_LOAD_1),
+            snmp_get(OID_LOAD_5),
+            snmp_get(OID_LOAD_15),
+        ),
+        asyncio.gather(
+            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(uptime_raw)
+    if secs is not None:
+        metrics["uptime_seconds"] = secs
+
+    if cpu_idle_raw is not None:
+        try:
+            metrics["cpu_percent"] = round(100.0 - float(cpu_idle_raw), 1)
+        except ValueError:
+            pass
+
+    if mem_total_raw and mem_free_raw:
+        try:
+            total_kb = int(mem_total_raw)
+            free_kb  = int(mem_free_raw)
+            buf_kb   = int(mem_buf_raw)   if mem_buf_raw   else 0
+            cache_kb = int(mem_cache_raw) if mem_cache_raw else 0
+            used_kb  = max(0, total_kb - free_kb - buf_kb - cache_kb)
+            metrics["memory_total_bytes"] = total_kb * 1024
+            metrics["memory_used_bytes"]  = used_kb  * 1024
+            metrics["memory_percent"]     = round(used_kb / total_kb * 100, 1) if total_kb else 0.0
+            metrics["memory_free_bytes"]  = free_kb  * 1024
+        except ValueError:
+            pass
+
+    if swap_total_raw and swap_avail_raw:
+        try:
+            stotal = int(swap_total_raw)
+            savail = int(swap_avail_raw)
+            sused  = stotal - savail
+            metrics["swap_total_bytes"] = stotal * 1024
+            metrics["swap_used_bytes"]  = sused  * 1024
+            metrics["swap_percent"]     = round(sused / stotal * 100, 1) if stotal else 0.0
+        except ValueError:
+            pass
+
+    for key, raw in [("load_1", load1_raw), ("load_5", load5_raw), ("load_15", load15_raw)]:
+        if raw is not None:
+            try:
+                metrics[key] = float(raw)
+            except ValueError:
+                pass
+
+    build_if_metrics(metrics, descr, speed, in_oct, out_oct, in_err, out_err,
+                     if_skip=if_skip, tx_err_is_drop=tx_err_is_drop)
+    return metrics

napalm_device_types/access_point.py

@@ -11,23 +11,29 @@ Usage::
 """
 
 from typing import Any, Dict, List
-from napalm.base import NetworkDriver
+from napalm_device_types.base import DeviceTypeDriver
+from napalm_device_types._ucd_metrics import IF_SKIP_DEFAULT, collect_ucd_metrics
 from napalm_device_types.models import (
+    ChannelScanEntryDict,
     Dot1XConfigDict,
     FastTransitionConfigDict,
+    HealthMetricsDict,
     MACACLDict,
     MeshConfigDict,
     MeshPeerDict,
     PackageDict,
     RadioStatusDict,
+    ServiceDict,
     SSIDBridgeDict,
     SSIDDict,
+    UpdateDict,
     WirelessClientDict,
     WirelessConfigDict,
 )
 
 
-class AccessPointDriver(NetworkDriver):
+class AccessPointDriver(DeviceTypeDriver):
+    TYPE_LABEL: str = "Access Point"
     """
     Abstract intermediate driver for wireless access points.
 
@@ -35,111 +41,41 @@ class AccessPointDriver(NetworkDriver):
     access-point-specific operations that concrete drivers must implement.
     """
 
-    def get_wireless_clients(self) -> List[WirelessClientDict]:
-        """
-        Returns a list of wireless clients currently associated with this
-        access point.
+    # Interfaces that carry no operational meaning on an access point and
+    # should be excluded from get_interfaces() / get_facts() interface_list.
+    _EXCLUDED_INTERFACES: frozenset = frozenset({"lo"})
 
-        Each entry contains:
+    # Interface name *prefixes* to exclude (e.g. Linux phy* are raw radio
+    # devices and have no IP/Ethernet significance at the AP level).
+    _EXCLUDED_INTERFACE_PREFIXES: tuple = ("phy",)
 
-        * mac (string) - client MAC address
-        * ssid (string) - SSID the client is connected to
-        * radio (string) - radio identifier (e.g. ``"radio0"``, ``"5GHz"``)
-        * signal (int) - received signal strength in dBm
-        * noise (int) - noise floor in dBm
-        * tx_rate (float) - TX bitrate in Mbit/s
-        * rx_rate (float) - RX bitrate in Mbit/s
-        * uptime (int) - association duration in seconds
+    _SNMP_SKIP_IF = IF_SKIP_DEFAULT
+    _SNMP_TX_ERR_IS_DROP: bool = False
 
-        Example::
+    @classmethod
+    async def get_health_metrics(cls, snmp_get, snmp_walk) -> HealthMetricsDict:
+        return await collect_ucd_metrics(
+            snmp_get, snmp_walk,
+            tx_err_is_drop=cls._SNMP_TX_ERR_IS_DROP,
+            if_skip=cls._SNMP_SKIP_IF,
+        )
 
-            [
-                {
-                    "mac": "AA:BB:CC:DD:EE:FF",
-                    "ssid": "MyNetwork",
-                    "radio": "radio1",
-                    "signal": -65,
-                    "noise": -95,
-                    "tx_rate": 300.0,
-                    "rx_rate": 144.0,
-                    "uptime": 3600,
-                }
-            ]
-        """
-        raise NotImplementedError
+    def _filter_interfaces(self, interfaces: Dict[str, Any]) -> Dict[str, Any]:
+        """Remove loopback and radio-device (phy*) interfaces from an interface dict."""
+        return {
+            name: data
+            for name, data in interfaces.items()
+            if name not in self._EXCLUDED_INTERFACES
+            and not name.startswith(self._EXCLUDED_INTERFACE_PREFIXES)
+        }
 
-    def get_ssids(self) -> Dict[str, SSIDDict]:
-        """
-        Returns the configured SSIDs (VAPs) on this access point.
-
-        Keys are SSID names.  Each value contains:
-
-        * enabled (bool) - whether the SSID is currently broadcasting
-        * radio (string) - radio the SSID is bound to
-        * bssid (string) - BSSID (MAC) of the VAP
-        * encryption (string) - e.g. ``"WPA2-PSK"``, ``"WPA3-SAE"``, ``"open"``
-        * hidden (bool) - whether the SSID is hidden
-        * clients (int) - number of currently associated clients
-
-        Example::
-
-            {
-                "MyNetwork": {
-                    "enabled": True,
-                    "radio": "radio1",
-                    "bssid": "AA:BB:CC:DD:EE:F0",
-                    "encryption": "WPA2-PSK",
-                    "hidden": False,
-                    "clients": 3,
-                },
-                "GuestNet": {
-                    "enabled": True,
-                    "radio": "radio0",
-                    "bssid": "AA:BB:CC:DD:EE:F1",
-                    "encryption": "WPA2-PSK",
-                    "hidden": False,
-                    "clients": 1,
-                },
-            }
-        """
-        raise NotImplementedError
-
-    def get_radio_status(self) -> Dict[str, RadioStatusDict]:
-        """
-        Returns the status of each radio interface.
-
-        Keys are radio identifiers (e.g. ``"radio0"``, ``"radio1"``).
-        Each value contains:
-
-        * enabled (bool) - whether the radio is active
-        * band (string) - frequency band, e.g. ``"2.4GHz"``, ``"5GHz"``, ``"6GHz"``
-        * channel (int) - operating channel number
-        * channel_width (int) - channel width in MHz (e.g. 20, 40, 80, 160)
-        * tx_power (int) - transmit power in dBm
-        * frequency (float) - center frequency in MHz
-
-        Example::
-
-            {
-                "radio0": {
-                    "enabled": True,
-                    "band": "2.4GHz",
-                    "channel": 6,
-                    "channel_width": 20,
-                    "tx_power": 20,
-                    "frequency": 2437.0,
-                },
-                "radio1": {
-                    "enabled": True,
-                    "band": "5GHz",
-                    "channel": 36,
-                    "channel_width": 80,
-                    "tx_power": 23,
-                    "frequency": 5180.0,
-                },
-            }
-        """
-        raise NotImplementedError
+    # AP-specific methods (get_wireless_clients, get_ssids, get_radio_status,
+    # get_interfaces, get_vlans, …) are intentionally NOT defined here.
+    # AccessPointDriver sits AFTER the mixin classes in the MRO of concrete
+    # drivers (e.g. OpenWrtDriver(InterfaceMixin, VlanMixin, …, AccessPointDriver)).
+    # NetworkDriver (a parent of AccessPointDriver) raises NotImplementedError for
+    # all standard NAPALM methods, so AccessPointDriver must come LAST to avoid
+    # shadowing the mixin implementations.
 
     def get_wireless_config(self) -> WirelessConfigDict:
         """
@@ -498,6 +434,67 @@ class AccessPointDriver(NetworkDriver):
         """
         raise NotImplementedError
 
+    def get_services(self) -> List[ServiceDict]:
+        """
+        Returns the list of system services known to the device's init system.
+
+        Each entry contains:
+
+        * name (string)    - service name as registered with the init system
+        * running (bool)   - ``True`` if the service process is currently running
+        * enabled (bool)   - ``True`` if the service starts automatically at boot
+        * pid (int)        - process ID of the main service process; 0 if not running
+
+        Example::
+
+            [
+                {"name": "lldpd",   "running": True,  "enabled": True,  "pid": 2341},
+                {"name": "sshd",    "running": True,  "enabled": True,  "pid": 1198},
+                {"name": "cron",    "running": False, "enabled": False, "pid": 0},
+            ]
+        """
+        raise NotImplementedError
+
+    def manage_service(self, name: str, action: str) -> Dict[str, Any]:
+        """
+        Execute a lifecycle action on a named service.
+
+        :param name: Service name as returned by :meth:`get_services`.
+        :param action: One of ``start``, ``stop``, ``restart``, ``enable``, ``disable``.
+        :returns: ``{"success": bool, "output": str}``
+        :raises ValueError: If ``name`` or ``action`` is invalid.
+        :raises NotImplementedError: If the driver does not support service management.
+        """
+        raise NotImplementedError
+
+    def get_available_updates(self) -> List[UpdateDict]:
+        """
+        Returns the list of installed packages that have a newer version available.
+
+        Uses the local package manager cache — does not run ``opkg update`` / ``apk update``.
+
+        :returns: List of :class:`~napalm_device_types.models.UpdateDict`.
+        :raises NotImplementedError: If the driver does not support update listing.
+
+        Example::
+
+            [
+                {"name": "busybox",  "current_version": "1.36.1-1", "new_version": "1.37.0-1"},
+                {"name": "dropbear", "current_version": "2022.83-2", "new_version": "2024.86-1"},
+            ]
+        """
+        raise NotImplementedError
+
+    def apply_updates(self, packages: List[str]) -> Dict[str, Any]:
+        """
+        Upgrade one or more packages to their newest available version.
+
+        :param packages: List of package names to upgrade.
+        :returns: ``{"success": bool, "output": str}``
+        :raises NotImplementedError: If the driver does not support package upgrades.
+        """
+        raise NotImplementedError
+
     def set_package_config(self, name: str, config: Dict[str, Any]) -> None:
         """
         Writes a new configuration for an installed package.

napalm_device_types/base.py

@@ -0,0 +1,80 @@
+"""
+DeviceTypeDriver — common base for all netOrk device-type drivers.
+
+Provides the fingerprinting contract used by the discovery subsystem to
+identify devices from SNMP / SSH / HTTP probe data without requiring a
+live connection.
+"""
+
+from __future__ import annotations
+
+from typing import NamedTuple
+
+from napalm.base import NetworkDriver
+
+
+class FingerprintRule(NamedTuple):
+    """Single pattern-matching rule for device fingerprinting.
+
+    pattern   — substring matched against lowercased probe data
+    weight    — evidence added to the driver's score on match
+    mandatory — pattern absent → driver immediately disqualified
+    negative  — pattern present → weight subtracted instead of added
+    """
+
+    pattern: str
+    weight: float = 1.0
+    mandatory: bool = False
+    negative: bool = False
+
+
+class PortSpec(NamedTuple):
+    """Port to probe during discovery and its fingerprinting contribution.
+
+    scheme    — "http", "https", or "ssh"
+    port      — TCP port number
+    paths     — URL paths to try (http/https only; ignored for ssh)
+    weight    — evidence added to score when this port responds
+    mandatory — port silent → driver immediately disqualified
+    """
+
+    scheme: str
+    port: int
+    paths: tuple[str, ...] = ("/",)
+    weight: float = 5.0
+    mandatory: bool = False
+
+
+class DeviceTypeDriver(NetworkDriver):
+    """Common base for all netOrk device-type drivers.
+
+    Sits between napalm.base.NetworkDriver and the type-specific abstract
+    classes (FirewallDriver, SwitchDriver, …). Adds the fingerprinting
+    interface consumed by the discovery subsystem; does not implement any
+    NAPALM abstract methods.
+
+    Override these class attributes in each concrete driver:
+
+    VENDOR                Human-readable vendor name ("AVM", "OPNsense", …).
+    DRIVER_NAME           NAPALM entry-point key ("fritzbox", "opnsense", …).
+    PORT_SPECS            Non-standard ports to probe beyond 80/443/22.
+                          None = no driver-specific ports expected.
+    SNMP_OBJECT_ID_PREFIX OID prefix for sysObjectID (1.3.6.1.2.1.1.2.0).
+                          Match → +15.0 score; set but no match → disqualified.
+    SNMP_FINGERPRINT      Rules matched against sysDescr (lowercased).
+    SSH_FINGERPRINT       Rules matched against SSH version string +
+                          pre-auth banner (lowercased, space-joined).
+    HTTP_FINGERPRINT      Rules matched against page title + Server header +
+                          body snippet (lowercased, whitespace-normalised).
+    """
+
+    VENDOR: str = ""
+    DRIVER_NAME: str = ""
+    PORT_SPECS: list[PortSpec] | None = None
+    SNMP_OBJECT_ID_PREFIX: str | None = None
+    SNMP_FINGERPRINT: list[FingerprintRule] = []
+    SSH_FINGERPRINT: list[FingerprintRule] = []
+    HTTP_FINGERPRINT: list[FingerprintRule] = []
+    OUI_PREFIXES: list[str] = []
+    # Format: "AA:BB:CC" — first 3 octets of MAC, uppercase, colon-separated.
+    # A match contributes fixed weight 6.0 to the fingerprint score.

napalm_device_types/config_lifecycle.py

@@ -0,0 +1,113 @@
+# -*- coding: utf-8 -*-
+from __future__ import annotations
+
+import difflib
+from typing import ClassVar
+
+from napalm.base.exceptions import (
+    CommandErrorException,
+    MergeConfigException,
+    ReplaceConfigException,
+)
+
+
+class ConfigLifecycleMixin:
+    """Mixin providing the standard NAPALM config lifecycle for CLI-driven drivers.
+
+    Provides a complete config lifecycle: ``load_merge_candidate``,
+    ``load_replace_candidate``, ``compare_config``, ``discard_config``,
+    ``has_pending_commit``, and a basic ``rollback`` that re-stages the
+    last backup and calls ``commit_config``.
+
+    Concrete drivers **must** provide:
+
+    * ``_get_running_config()`` -- return the current running config as a
+      string (e.g. ``show running-config`` or ``uci export``)
+    * ``commit_config()`` -- vendor-specific apply-and-persist logic
+
+    Override ``rollback()`` if the device needs a smarter revert (e.g.
+    block-diff rollback or config-import-based recovery).
+
+    Config-state attributes (``_candidate_config``, ``_candidate_mode``,
+    ``_backup_config``) are annotated here for type checkers; they default
+    to ``None`` and are written by the lifecycle methods.
+    """
+
+    _candidate_config: str | None = None
+    _candidate_mode: str | None = None
+    _backup_config: str | None = None
+
+    _comment_chars: ClassVar[tuple[str, ...]] = ("!", "#")
+
+    def _get_running_config(self) -> str:
+        raise NotImplementedError
+
+    def commit_config(self, message: str = "", revert_in: int | None = None) -> None:
+        raise NotImplementedError
+
+    def load_merge_candidate(
+        self, filename: str | None = None, config: str | None = None
+    ) -> None:
+        if filename is not None:
+            try:
+                with open(filename) as fh:
+                    config = fh.read()
+            except OSError as exc:
+                raise MergeConfigException(str(exc)) from exc
+        if config is None:
+            raise MergeConfigException("Either 'filename' or 'config' must be provided.")
+        self._candidate_config = config
+        self._candidate_mode = "merge"
+
+    def load_replace_candidate(
+        self, filename: str | None = None, config: str | None = None
+    ) -> None:
+        if filename is not None:
+            try:
+                with open(filename) as fh:
+                    config = fh.read()
+            except OSError as exc:
+                raise ReplaceConfigException(str(exc)) from exc
+        if config is None:
+            raise ReplaceConfigException("Either 'filename' or 'config' must be provided.")
+        self._candidate_config = config
+        self._candidate_mode = "replace"
+
+    def compare_config(self) -> str:
+        if self._candidate_config is None:
+            return ""
+
+        if self._candidate_mode == "merge":
+            lines = []
+            for line in self._candidate_config.splitlines():
+                if line.strip() and not line.strip().startswith(self._comment_chars):
+                    lines.append(f"+{line}")
+            return "\n".join(lines)
+
+        running = self._get_running_config()
+        diff = difflib.unified_diff(
+            running.splitlines(),
+            self._candidate_config.splitlines(),
+            fromfile="running-config",
+            tofile="candidate-config",
+            lineterm="",
+        )
+        return "\n".join(diff)
+
+    def discard_config(self) -> None:
+        self._candidate_config = None
+        self._candidate_mode = None
+
+    def has_pending_commit(self) -> bool:
+        return self._candidate_config is not None
+
+    def rollback(self) -> None:
+        if self._backup_config is None:
+            raise CommandErrorException(
+                "No backup configuration available – "
+                "commit_config has not been called in this session."
+            )
+        self._candidate_config = self._backup_config
+        self._candidate_mode = "merge"
+        self.commit_config()
+        self._backup_config = None

napalm_device_types/firewall.py

@@ -11,8 +11,10 @@ Usage::
 """
 
 from typing import Any, Dict, List
-from napalm.base import NetworkDriver
+from napalm_device_types.base import DeviceTypeDriver
+from napalm_device_types._ucd_metrics import IF_SKIP_DEFAULT, collect_ucd_metrics
 from napalm_device_types.models import (
+    HealthMetricsDict,
     NATTranslationDict,
     PackageDict,
     SecurityZoneDict,
@@ -21,7 +23,8 @@ from napalm_device_types.models import (
 )
 
 
-class FirewallDriver(NetworkDriver):
+class FirewallDriver(DeviceTypeDriver):
+    TYPE_LABEL: str = "Firewall"
     """
     Abstract intermediate driver for firewall/security devices.
 
@@ -30,6 +33,18 @@ class FirewallDriver(NetworkDriver):
     that concrete drivers must implement.
     """
 
+    _SNMP_SKIP_IF = IF_SKIP_DEFAULT
+    # OPNsense reports drops in the out-error counter.
+    _SNMP_TX_ERR_IS_DROP: bool = True
+
+    @classmethod
+    async def get_health_metrics(cls, snmp_get, snmp_walk) -> HealthMetricsDict:
+        return await collect_ucd_metrics(
+            snmp_get, snmp_walk,
+            tx_err_is_drop=cls._SNMP_TX_ERR_IS_DROP,
+            if_skip=cls._SNMP_SKIP_IF,
+        )
+
     def get_nat_translations(self) -> List[NATTranslationDict]:
         """
         Returns a list of active NAT translation entries.

napalm_device_types/hypervisor.py

@@ -11,18 +11,24 @@ Usage::
 """
 
 from typing import Any, Dict, List
-from napalm.base import NetworkDriver
+from napalm_device_types.base import DeviceTypeDriver
+from napalm_device_types._ucd_metrics import IF_SKIP_DEFAULT, collect_ucd_metrics
 from napalm_device_types.models import (
+    HealthMetricsDict,
+    NICConfigDict,
     PackageDict,
     SnapshotDict,
     StorageVolumeDict,
     VMConfigDict,
     VMDict,
+    VMProvisionResultDict,
+    VMStatusDict,
     VirtualNetworkDict,
 )
 
 
-class HypervisorDriver(NetworkDriver):
+class HypervisorDriver(DeviceTypeDriver):
+    TYPE_LABEL: str = "Hypervisor"
     """
     Abstract intermediate driver for hypervisors and virtualisation platforms
     (e.g. Proxmox VE, VMware ESXi, KVM/libvirt, Hyper-V).
@@ -31,6 +37,17 @@ class HypervisorDriver(NetworkDriver):
     hypervisor-specific operations that concrete drivers must implement.
     """
 
+    _SNMP_SKIP_IF = IF_SKIP_DEFAULT
+    _SNMP_TX_ERR_IS_DROP: bool = False
+
+    @classmethod
+    async def get_health_metrics(cls, snmp_get, snmp_walk) -> HealthMetricsDict:
+        return await collect_ucd_metrics(
+            snmp_get, snmp_walk,
+            tx_err_is_drop=cls._SNMP_TX_ERR_IS_DROP,
+            if_skip=cls._SNMP_SKIP_IF,
+        )
+
     # ------------------------------------------------------------------
     # Virtual machines – read
     # ------------------------------------------------------------------
@@ -521,3 +538,120 @@ class HypervisorDriver(NetworkDriver):
             )
         """
         raise NotImplementedError
+
+    # ------------------------------------------------------------------
+    # Virtual machine provisioning
+    # ------------------------------------------------------------------
+
+    def create_vm_from_cloud_init(
+        self,
+        name: str,
+        *,
+        template: str,
+        cpu: int,
+        memory: int,
+        nics: List[NICConfigDict],
+        cloud_init_config: Dict[str, Any],
+        ssh_public_keys: List[str] | None = None,
+        disk_resize_gb: int | None = None,
+        timeout: int = 180,
+    ) -> VMProvisionResultDict:
+        """
+        Create a new virtual machine from a Cloud-Init template.
+
+        Clones a pre-existing VM template, configures virtual network interfaces,
+        and injects Cloud-Init configuration via a storage snippet or similar
+        mechanism.  The resulting VM is left in a running state.
+
+        Args:
+            name (string) - new VM display name
+            template (string) - hypervisor-internal ID/name of the template VM to clone
+            cpu (int) - number of virtual CPUs to assign
+            memory (int) - RAM to assign in megabytes
+            nics (list[NICConfigDict]) - list of network interface configurations.
+              First NIC is primary (DHCP by default); subsequent NICs are optional.
+              Each entry specifies bridge, optional vlan_tag (access) or trunk_vlan_tags,
+              and dhcp flag.
+            cloud_init_config (dict) - user-data dict (will be rendered to YAML).
+              Should include hostname, bootstrap_token, runcmd, and any custom config.
+            ssh_public_keys (list[str] | None) - SSH public keys to inject into guest.
+              If None or empty, no SSH key injection is performed.
+            disk_resize_gb (int | None) - resize root disk to this size in GB.
+              If None, disk remains template size. Default None.
+            timeout (int) - maximum seconds to wait for provisioning completion
+              (clone, config, start). Default 180.
+
+        Returns:
+            VMProvisionResultDict - ``{"vmid": str, "name": str, "node": str}``
+              vmid is the hypervisor-internal VM ID as a string (numeric for Proxmox).
+
+        Raises:
+            RuntimeError - if provisioning fails (storage unavailable, invalid
+              config, timeout, etc.)
+        """
+        raise NotImplementedError
+
+    def destroy_vm(
+        self,
+        vmid: str,
+        *,
+        remove_disk: bool = True,
+        timeout: int = 60,
+    ) -> None:
+        """
+        Destroy a virtual machine and optionally remove its storage.
+
+        Stops the VM (if running) and purges it from the hypervisor. Optionally
+        also removes associated disks and ephemeral storage (e.g. Cloud-Init snippets).
+
+        Args:
+            vmid (string) - hypervisor-internal VM ID (as returned from ``get_vms()``
+              or ``create_vm_from_cloud_init()``).
+            remove_disk (bool) - if True (default), delete all disks and snapshot
+              data associated with the VM.  If False, only the VM configuration
+              is removed; disks are left behind (rare use case).
+            timeout (int) - maximum seconds to wait for the destroy operation
+              (stop + delete). Default 60.
+
+        Raises:
+            RuntimeError - if the VM does not exist or destroy fails.
+        """
+        raise NotImplementedError
+
+    def get_vm_status(
+        self,
+        vmid: str,
+        *,
+        wait_for_ip: bool = False,
+        timeout: int = 300,
+        poll_interval: int = 5,
+    ) -> VMStatusDict:
+        """
+        Get the current runtime status of a virtual machine.
+
+        Optionally waits for the VM to acquire an IP address on its management
+        NIC (net0), useful when provisioning a VM and waiting for it to boot.
+
+        Args:
+            vmid (string) - hypervisor-internal VM ID.
+            wait_for_ip (bool) - if True, poll until the VM's guest-agent reports
+              an IP address on the first NIC (net0, the management NIC).  Useful
+              for ``create_vm_from_cloud_init()`` follow-up.  If False, return
+              immediate status without waiting (may have no IP).
+            timeout (int) - maximum seconds to wait for IP acquisition
+              (if wait_for_ip=True). Raises RuntimeError if timeout is exceeded.
+              Default 300.
+            poll_interval (int) - seconds between status polls (if wait_for_ip=True).
+              Default 5.
+
+        Returns:
+            VMStatusDict - ``{"status": str, "ip_address": str, "hostname": str,
+              "mac_address": str}``.  ip_address, hostname, and mac_address refer to
+              the primary NIC (first interface) and are only present if the VM is
+              running and has network info available.
+
+        Raises:
+            RuntimeError - if the VM does not exist or if wait_for_ip=True and
+              timeout is exceeded.
+        """
+        raise NotImplementedError

napalm_device_types/models.py

@@ -6,7 +6,7 @@ abstract device-type driver classes in this package.
 """
 
 from typing import Dict, List, Optional
-from typing_extensions import TypedDict
+from typing_extensions import NotRequired, TypedDict
 
 
 # ---------------------------------------------------------------------------
@@ -50,6 +50,23 @@ class PackageDict(TypedDict):
     source: str
 
 
+class ServiceDict(TypedDict):
+    """A system service managed by the device's init system (e.g. procd on OpenWrt)."""
+
+    name: str
+    running: bool
+    enabled: bool
+    pid: int  # 0 if not running
+
+
+class UpdateDict(TypedDict):
+    """A software package that has a newer version available in the package repository."""
+
+    name: str
+    current_version: str
+    new_version: str
+
+
 # ---------------------------------------------------------------------------
 # Access Point
 # ---------------------------------------------------------------------------
@@ -64,6 +81,10 @@ class WirelessClientDict(TypedDict):
     tx_rate: float
     rx_rate: float
     uptime: int
+    # Optional fields populated by DHCP cross-reference (e.g. from firewall)
+    ip: NotRequired[str]
+    hostname: NotRequired[str]
+    lease_end: NotRequired[int]  # Unix timestamp when DHCP lease expires
 
 
 class SSIDDict(TypedDict):
@@ -84,6 +105,18 @@ class RadioStatusDict(TypedDict):
     frequency: float
 
 
+class ChannelScanEntryDict(TypedDict):
+    """One neighboring AP entry as returned by get_channel_scan()."""
+
+    bssid: str
+    ssid: str
+    frequency: int      # centre frequency in MHz, e.g. 2437
+    channel: int        # channel number, e.g. 6
+    signal_dbm: int     # RSSI in dBm, e.g. -72
+    channel_width: int  # MHz: 20 / 40 / 80 / 160; 0 if unknown
+    band: str           # "2.4GHz" | "5GHz" | "6GHz"
+
+
 class WirelessConfigDict(TypedDict):
     country_code: str
     regulatory_domain: str
@@ -289,6 +322,44 @@ class VPNTunnelDict(TypedDict):
     uptime: int
     bytes_in: int
     bytes_out: int
+    description: NotRequired[str]  # human-readable tunnel description / name
+
+
+# ---------------------------------------------------------------------------
+# Residential Gateway
+# ---------------------------------------------------------------------------
+
+
+class WANStatusDict(TypedDict):
+    connection_type: str  # e.g. "DSL", "Cable", "PPPoE", "DHCP"
+    is_connected: bool
+    external_ip: str
+    uptime: int
+    bytes_sent: int
+    bytes_received: int
+    max_bitrate_up: int  # kbit/s
+    max_bitrate_down: int  # kbit/s
+    external_ipv6: NotRequired[str]
+    link_status: NotRequired[str]  # physical line state, e.g. "Up" / "Down"
+
+
+class PortForwardDict(TypedDict):
+    name: str
+    protocol: str  # "TCP" or "UDP"
+    external_port: int
+    internal_ip: str
+    internal_port: int
+    enabled: bool
+    remote_host: NotRequired[str]  # restrict forward to a specific remote source
+
+
+class HostDict(TypedDict):
+    mac: str
+    ip: str
+    hostname: str
+    interface_type: str  # "LAN", "WLAN", ...
+    is_active: bool
+    lease_time_remaining: NotRequired[int]  # seconds
 
 
 # ---------------------------------------------------------------------------
@@ -475,3 +546,187 @@ class ReplicationJobDict(TypedDict):
     last_run: float    # Unix epoch; 0.0 if never run
     last_status: str   # "success", "failed", "running", "pending"
     bytes_sent: int    # bytes transferred in the last run
+
+
+# ---------------------------------------------------------------------------
+# OS / General-purpose Linux
+# ---------------------------------------------------------------------------
+
+
+class UserDict(TypedDict):
+    """A local OS user account."""
+
+    username: str
+    uid: int
+    gid: int
+    home: str
+    shell: str
+    groups: List[str]   # all supplementary group names
+
+
+class ProcessDict(TypedDict):
+    """A running OS process."""
+
+    pid: int
+    ppid: int
+    user: str
+    cpu: float          # CPU utilisation 0.0–100.0
+    memory: float       # RSS percentage of total RAM 0.0–100.0
+    vsz: int            # virtual memory size in KiB
+    rss: int            # resident set size in KiB
+    tty: str            # controlling terminal; empty string if none
+    state: str          # "R" running, "S" sleeping, "D" uninterruptible, "Z" zombie, etc.
+    started: str        # start time string as reported by ``ps`` (e.g. "12:34" or "May28")
+    command: str        # full command line
+
+
+class CronJobDict(TypedDict):
+    """A scheduled cron task."""
+
+    user: str           # owner of the crontab entry
+    schedule: str       # five-field cron expression, e.g. "0 * * * *"
+    command: str        # shell command to execute
+    description: NotRequired[str]  # optional inline comment
+
+
+class ApplyUpdatesResultDict(TypedDict):
+    """Result of an ``apply_updates()`` call."""
+
+    success: bool       # True if the package manager exited without error
+    output: str         # raw stdout/stderr captured from the package manager
+    error: NotRequired[str]  # error message if success is False
+
+
+class DockerContainerDict(TypedDict):
+    """A Docker container entry as returned by ``docker ps -a``."""
+
+    id: str             # short container ID
+    name: str           # container name(s)
+    image: str          # image reference
+    image_version: str  # OCI org.opencontainers.image.version label; empty if absent
+    command: str        # entrypoint / command
+    created: str        # creation timestamp string
+    status: str         # human-readable status (e.g. "Up 3 hours")
+    ports: str          # port mapping string
+    state: str          # "running", "exited", "paused", etc.
+
+
+class DockerImageDict(TypedDict):
+    """A local Docker image entry as returned by ``docker images``."""
+
+    id: str             # short image ID
+    repository: str     # image repository
+    tag: str            # image tag
+    size: str           # human-readable size string (e.g. "187MB")
+    created: str        # creation timestamp string
+    version: str        # OCI org.opencontainers.image.version label; empty if absent
+
+
+class DockerVolumeDict(TypedDict):
+    """A Docker volume entry as returned by ``docker volume ls``."""
+
+    name: str           # volume name
+    driver: str         # volume driver (e.g. "local")
+    mountpoint: str     # host filesystem path
+    scope: str          # "local" or "global"
+
+
+class DockerNetworkDict(TypedDict):
+    """A Docker network entry as returned by ``docker network ls``."""
+
+    id: str             # short network ID
+    name: str           # network name
+    driver: str         # network driver (e.g. "bridge", "host", "overlay")
+    scope: str          # network scope
+    ipv6: str           # "true" if IPv6 is enabled, "false" otherwise
+    internal: str       # "true" if the network is internal, "false" otherwise
+
+
+class DockerInfoDict(TypedDict):
+    """Return value of ``get_docker_info()``."""
+
+    available: bool                              # False if Docker is absent/inaccessible
+    permission_denied: NotRequired[bool]         # True when socket access is denied
+    version: NotRequired[str]                    # Docker Engine version string
+    containers: NotRequired[List[DockerContainerDict]]
+    images: NotRequired[List[DockerImageDict]]
+    volumes: NotRequired[List[DockerVolumeDict]]
+    networks: NotRequired[List[DockerNetworkDict]]
+    outdated_images: NotRequired[List[str]]      # image names with a newer remote digest
+
+
+class DeviceActionResultDict(TypedDict):
+    """Return value of ``run_device_action()``."""
+
+    success: bool   # True if the action completed without error
+    output: str     # human-readable output or status message
+
+
+class SNMPConfigDict(TypedDict):
+    """SNMP agent configuration detected on the device."""
+
+    running: bool           # True if the SNMP daemon is active
+    community: str          # read community string (e.g. "public")
+    port: int               # listening port (default 161)
+    version: str            # highest supported version: "1", "2c", or "3"
+
+
+# ---------------------------------------------------------------------------
+# Health Metrics  (returned by get_health_metrics())
+# ---------------------------------------------------------------------------
+
+
+class HealthMetricsIfaceDict(TypedDict):
+    """Per-interface counters inside a HealthMetricsDict."""
+
+    name: str
+    rx_bytes: NotRequired[int]
+    tx_bytes: NotRequired[int]
+    rx_errors: NotRequired[int]
+    tx_errors: NotRequired[int]
+    tx_drops: NotRequired[int]   # populated instead of tx_errors when the driver reports drops
+    speed_mbps: NotRequired[int]
+
+
+class HealthMetricsDict(TypedDict):
+    """Return value of ``get_health_metrics()``."""
+
+    uptime_seconds: NotRequired[int]
+    cpu_percent: NotRequired[float]
+    memory_total_bytes: NotRequired[int]
+    memory_used_bytes: NotRequired[int]
+    memory_free_bytes: NotRequired[int]
+    memory_percent: NotRequired[float]
+    swap_total_bytes: NotRequired[int]
+    swap_used_bytes: NotRequired[int]
+    swap_percent: NotRequired[float]
+    load_1: NotRequired[float]
+    load_5: NotRequired[float]
+    load_15: NotRequired[float]
+    interfaces: NotRequired[Dict[str, HealthMetricsIfaceDict]]
+
+
+class NICConfigDict(TypedDict):
+    """Network interface configuration for VM provisioning."""
+
+    bridge: str  # Bridge or network name
+    vlan_tag: NotRequired[int | None]  # Access VLAN (None = untagged)
+    trunk_vlan_tags: NotRequired[list[int]]  # Trunk VLAN list (alternative to vlan_tag)
+    dhcp: NotRequired[bool]  # Enable DHCP (default True for first NIC, False for others)
+
+
+class VMProvisionResultDict(TypedDict):
+    """Return value of ``create_vm_from_cloud_init()`` — provisioned VM identifier."""
+
+    vmid: str  # Hypervisor-internal VM ID (string for portability across hypervisors)
+    name: str  # VM display name
+    node: str  # Cluster node name (empty string for standalone hypervisors)
+
+
+class VMStatusDict(TypedDict):
+    """Return value of ``get_vm_status()`` — runtime information of a VM."""
+
+    status: str  # "running", "stopped", "paused", "suspended", etc.
+    ip_address: NotRequired[str]  # Management NIC IP (absent if VM has no IP or is stopped)
+    hostname: NotRequired[str]  # Hostname resolved from IP (if available)
+    mac_address: NotRequired[str]  # MAC of management NIC

napalm_device_types/os.py

@@ -0,0 +1,429 @@
+"""
+Abstract base class for OS / general-purpose server drivers.
+
+Usage::
+
+    from napalm_device_types import OSDriver
+
+    class LinuxDriver(OSDriver):
+        def get_packages(self):
+            ...
+"""
+
+import re
+from typing import List, Optional
+from napalm_device_types.base import DeviceTypeDriver
+from napalm_device_types._ucd_metrics import IF_SKIP_DEFAULT, collect_ucd_metrics
+from napalm_device_types.models import (
+    ApplyUpdatesResultDict,
+    CronJobDict,
+    DeviceActionResultDict,
+    DockerInfoDict,
+    HealthMetricsDict,
+    PackageDict,
+    ProcessDict,
+    ServiceDict,
+    SNMPConfigDict,
+    UpdateDict,
+    UserDict,
+)
+
+
+class OSDriver(DeviceTypeDriver):
+    TYPE_LABEL: str = "OS"
+    """
+    Abstract intermediate driver for general-purpose operating systems
+    (e.g. Linux, BSD, macOS).
+
+    Inherits all standard NAPALM NetworkDriver methods and adds OS-specific
+    operations that concrete drivers must implement.
+    """
+
+    # Interfaces matching this pattern are excluded from health-metric collection.
+    _SNMP_SKIP_IF: re.Pattern = IF_SKIP_DEFAULT
+    # Set True for drivers where the out-error counter actually reports drops (e.g. OPNsense).
+    _SNMP_TX_ERR_IS_DROP: bool = False
+
+    @classmethod
+    async def get_health_metrics(cls, snmp_get, snmp_walk) -> HealthMetricsDict:
+        """Collect SNMP health metrics (CPU, memory, uptime, load, interfaces).
+
+        :param snmp_get:  async callable ``(oid: str) -> Optional[str]``
+        :param snmp_walk: async callable ``(oid: str) -> Dict[str, str]``
+        """
+        return await collect_ucd_metrics(
+            snmp_get, snmp_walk,
+            tx_err_is_drop=cls._SNMP_TX_ERR_IS_DROP,
+            if_skip=cls._SNMP_SKIP_IF,
+        )
+
+    # ------------------------------------------------------------------
+    # Package management
+    # ------------------------------------------------------------------
+
+    def get_packages(self) -> List[PackageDict]:
+        """
+        Returns a list of all installed software packages.
+
+        Each entry contains:
+
+        * name (string) - package name
+        * version (string) - installed version string
+        * installed (bool) - always ``True`` for this method
+        * description (string) - short package description
+        * size (int) - installed size in bytes; ``0`` if unavailable
+        * source (string) - package source / repository name; empty string if unavailable
+
+        Example::
+
+            [
+                {
+                    "name": "openssh-server",
+                    "version": "1:9.2p1-2+deb12u2",
+                    "installed": True,
+                    "description": "secure shell (SSH) server, for secure access from remote machines",
+                    "size": 524288,
+                    "source": "Debian",
+                },
+            ]
+        """
+        raise NotImplementedError
+
+    def get_pending_updates(self) -> List[UpdateDict]:
+        """
+        Returns a list of packages that have a newer version available.
+
+        Each entry contains:
+
+        * name (string) - package name
+        * current_version (string) - currently installed version
+        * new_version (string) - version available in the repository
+
+        Example::
+
+            [
+                {
+                    "name": "openssh-server",
+                    "current_version": "1:9.2p1-2+deb12u1",
+                    "new_version": "1:9.2p1-2+deb12u2",
+                },
+            ]
+        """
+        raise NotImplementedError
+
+    def apply_updates(self, packages: List[str]) -> ApplyUpdatesResultDict:
+        """
+        Upgrades the given packages to the newest available version.
+
+        Only packages that are already installed may be upgraded; this method
+        does **not** install new packages.  Pass an empty list to upgrade
+        **all** packages that have pending updates.
+
+        :param packages: List of package names to upgrade.  Each name must
+            match ``^[a-zA-Z0-9_\\-\\+\\.]+$``; a :exc:`ValueError` is raised
+            for any name that does not conform.
+        :returns: A dict with:
+
+            * success (bool) – ``True`` if the package manager exited without error
+            * output (string) – combined stdout / stderr from the package manager
+            * error (string, optional) – short error message when *success* is ``False``
+
+        :raises ValueError: If any package name fails the safety check.
+
+        Example::
+
+            result = driver.apply_updates(["openssh-server", "curl"])
+            # → {"success": True, "output": "Reading package lists...\\n..."}
+
+            # Upgrade everything:
+            result = driver.apply_updates([])
+        """
+        raise NotImplementedError
+
+    # ------------------------------------------------------------------
+    # Service management
+    # ------------------------------------------------------------------
+
+    def get_services(self) -> List[ServiceDict]:
+        """
+        Returns a list of system services and their current state.
+
+        Each entry contains:
+
+        * name (string) - service unit name (without ``.service`` suffix)
+        * running (bool) - ``True`` if the service is currently active
+        * enabled (bool) - ``True`` if the service starts automatically on boot
+        * pid (int) - main process ID; ``0`` if not running
+
+        Example::
+
+            [
+                {
+                    "name": "ssh",
+                    "running": True,
+                    "enabled": True,
+                    "pid": 1234,
+                },
+                {
+                    "name": "cron",
+                    "running": True,
+                    "enabled": True,
+                    "pid": 5678,
+                },
+            ]
+        """
+        raise NotImplementedError
+
+    # ------------------------------------------------------------------
+    # Users
+    # ------------------------------------------------------------------
+
+    def get_users(self) -> List[UserDict]:
+        """
+        Returns a list of local OS user accounts.
+
+        Each entry contains:
+
+        * username (string) - login name
+        * uid (int) - numeric user ID
+        * gid (int) - primary group ID
+        * home (string) - home directory path
+        * shell (string) - login shell path
+        * groups (list of strings) - all supplementary group names
+
+        Example::
+
+            [
+                {
+                    "username": "admin",
+                    "uid": 1000,
+                    "gid": 1000,
+                    "home": "/home/admin",
+                    "shell": "/bin/bash",
+                    "groups": ["sudo", "docker", "adm"],
+                },
+            ]
+        """
+        raise NotImplementedError
+
+    # ------------------------------------------------------------------
+    # Processes
+    # ------------------------------------------------------------------
+
+    def get_processes(self) -> List[ProcessDict]:
+        """
+        Returns a snapshot of currently running processes.
+
+        Each entry contains:
+
+        * pid (int) - process ID
+        * ppid (int) - parent process ID
+        * user (string) - effective user name
+        * cpu (float) - CPU utilisation percentage
+        * memory (float) - RSS as a percentage of total RAM
+        * vsz (int) - virtual memory size in KiB
+        * rss (int) - resident set size in KiB
+        * tty (string) - controlling terminal; empty string if none
+        * state (string) - process state: ``"R"`` running, ``"S"`` sleeping,
+          ``"D"`` uninterruptible, ``"Z"`` zombie, ``"T"`` stopped, etc.
+        * started (string) - start time as printed by ``ps`` (e.g. ``"12:34"`` or ``"May28"``)
+        * command (string) - full command line
+
+        Example::
+
+            [
+                {
+                    "pid": 1,
+                    "ppid": 0,
+                    "user": "root",
+                    "cpu": 0.0,
+                    "memory": 0.1,
+                    "vsz": 168576,
+                    "rss": 13312,
+                    "tty": "",
+                    "state": "S",
+                    "started": "May28",
+                    "command": "/sbin/init",
+                },
+            ]
+        """
+        raise NotImplementedError
+
+    # ------------------------------------------------------------------
+    # Cron jobs
+    # ------------------------------------------------------------------
+
+    def get_cron_jobs(self) -> List[CronJobDict]:
+        """
+        Returns scheduled cron tasks from all user crontabs and ``/etc/cron.d``.
+
+        Each entry contains:
+
+        * user (string) - owner of the crontab entry
+        * schedule (string) - five-field cron expression (e.g. ``"0 * * * *"``)
+        * command (string) - shell command to execute
+        * description (string, optional) - inline comment text if present
+
+        Example::
+
+            [
+                {
+                    "user": "root",
+                    "schedule": "0 4 * * *",
+                    "command": "/usr/local/bin/backup.sh",
+                    "description": "nightly backup",
+                },
+            ]
+        """
+        raise NotImplementedError
+
+    # ------------------------------------------------------------------
+    # SNMP
+    # ------------------------------------------------------------------
+
+    def get_snmp_config(self) -> Optional[SNMPConfigDict]:
+        """
+        Returns the SNMP agent configuration currently active on the device,
+        or ``None`` if no SNMP daemon is running or detectable.
+
+        The returned dictionary contains:
+
+        * running (bool) - whether the SNMP daemon is currently active
+        * community (string) - the read community string (e.g. ``"public"``)
+        * port (int) - the UDP port the agent listens on (default ``161``)
+        * version (string) - highest supported SNMP version: ``"1"``, ``"2c"``, or ``"3"``
+
+        Example::
+
+            # snmpd running with community "public":
+            {
+                "running": True,
+                "community": "public",
+                "port": 161,
+                "version": "2c",
+            }
+
+            # snmpd not installed / not running:
+            None
+        """
+        raise NotImplementedError
+
+    # ------------------------------------------------------------------
+    # Docker
+    # ------------------------------------------------------------------
+
+    def get_docker_info(self) -> DockerInfoDict:
+        """
+        Returns information about the local Docker environment.
+
+        If Docker is not installed or the current user lacks access to the
+        Docker socket, returns ``{"available": False}``.  When the user has
+        no socket permission, ``permission_denied`` is additionally set to
+        ``True``.
+
+        When Docker is available the dict contains:
+
+        * available (bool) - always ``True``
+        * version (string) - Docker Engine version string
+        * containers (list) - all containers (running and stopped), each with:
+
+          * id (string) - short container ID
+          * name (string) - container name(s)
+          * image (string) - image reference
+          * image_version (string) - OCI ``org.opencontainers.image.version`` label; empty if absent
+          * command (string) - entrypoint / command string
+          * created (string) - creation timestamp string
+          * status (string) - human-readable status (e.g. ``"Up 3 hours"``)
+          * ports (string) - port mapping string
+          * state (string) - ``"running"``, ``"exited"``, ``"paused"``, etc.
+
+        * images (list) - local images, each with:
+
+          * id (string) - short image ID
+          * repository (string) - image repository
+          * tag (string) - image tag
+          * size (string) - human-readable size string (e.g. ``"187MB"``)
+          * created (string) - creation timestamp string
+          * version (string) - OCI ``org.opencontainers.image.version`` label; empty if absent
+
+        * volumes (list) - Docker volumes, each with:
+
+          * name (string) - volume name
+          * driver (string) - volume driver
+          * mountpoint (string) - host filesystem path
+          * scope (string) - ``"local"`` or ``"global"``
+
+        * networks (list) - Docker networks, each with:
+
+          * id (string) - short network ID
+          * name (string) - network name
+          * driver (string) - network driver (e.g. ``"bridge"``, ``"host"``, ``"overlay"``)
+          * scope (string) - network scope
+          * ipv6 (string) - ``"true"`` if IPv6 is enabled
+          * internal (string) - ``"true"`` if the network is internal
+
+        * outdated_images (list of strings) - image names where the local digest
+          differs from the latest remote digest; empty list if all images are
+          current or update checks could not be performed.
+
+        Example::
+
+            # Docker not installed:
+            {"available": False}
+
+            # Docker installed, no socket permission:
+            {"available": False, "permission_denied": True}
+
+            # Docker available:
+            {
+                "available": True,
+                "version": "Docker version 27.3.1, build ce12230",
+                "containers": [
+                    {
+                        "id": "a1b2c3d4e5f6",
+                        "name": "my-app",
+                        "image": "nginx:latest",
+                        "image_version": "1.27.0",
+                        "command": "nginx -g 'daemon off;'",
+                        "created": "2026-05-28 10:00:00 +0000 UTC",
+                        "status": "Up 3 days",
+                        "ports": "0.0.0.0:80->80/tcp",
+                        "state": "running",
+                    },
+                ],
+                "images": [...],
+                "volumes": [],
+                "networks": [...],
+                "outdated_images": ["nginx:latest"],
+            }
+        """
+        raise NotImplementedError
+
+    # ------------------------------------------------------------------
+    # Generic device actions
+    # ------------------------------------------------------------------
+
+    def run_device_action(self, action: str) -> DeviceActionResultDict:
+        """
+        Executes a named administrative action on the device.
+
+        This method is an extensibility point for driver-specific one-off
+        operations that do not fit any other NAPALM API method.  Each driver
+        documents the action names it supports.
+
+        :param action: Action identifier string (e.g. ``"fix_docker_permissions"``).
+        :raises NotImplementedError: If the driver does not implement this method.
+        :raises ValueError: If ``action`` is not a recognised action name for
+            this driver.
+
+        :returns: A dict with:
+
+            * success (bool) – ``True`` if the action completed without error
+            * output (string) – human-readable output or status message
+
+        Example::
+
+            result = driver.run_device_action("fix_docker_permissions")
+            # → {"success": True, "output": "Added 'pi' to the docker group. Reconnect..."}
+        """
+        raise NotImplementedError

napalm_device_types/residential_gateway.py

@@ -0,0 +1,197 @@
+"""
+Abstract base class for residential gateway drivers.
+
+A "residential gateway" combines the roles of router, firewall and
+wireless access point in a single consumer device (e.g. AVM FritzBox,
+ISP-supplied DSL/cable routers). This base class merges the relevant
+subsets of :class:`~napalm_device_types.firewall.FirewallDriver` and
+:class:`~napalm_device_types.access_point.AccessPointDriver` plus
+gateway-specific operations (WAN status, port forwarding, connected
+hosts).
+
+Usage::
+
+    from napalm_device_types import ResidentialGatewayDriver
+
+    class FritzBoxDriver(ResidentialGatewayDriver):
+        def get_wan_status(self):
+            ...
+"""
+
+from typing import Dict, List
+from napalm_device_types.base import DeviceTypeDriver
+from napalm_device_types._ucd_metrics import IF_SKIP_DEFAULT, collect_ucd_metrics
+from napalm_device_types.models import (
+    HealthMetricsDict,
+    HostDict,
+    NATTranslationDict,
+    PortForwardDict,
+    RadioStatusDict,
+    SSIDDict,
+    VPNTunnelDict,
+    WANStatusDict,
+    WirelessClientDict,
+)
+
+
+class ResidentialGatewayDriver(DeviceTypeDriver):
+    TYPE_LABEL: str = "Gateway"
+    """
+    Abstract intermediate driver for residential gateways (router + firewall + AP).
+
+    Inherits all standard NAPALM NetworkDriver methods and adds the
+    gateway-, firewall- and wireless-specific operations that concrete
+    drivers must implement.
+    """
+
+    _SNMP_SKIP_IF = IF_SKIP_DEFAULT
+    _SNMP_TX_ERR_IS_DROP: bool = False
+
+    @classmethod
+    async def get_health_metrics(cls, snmp_get, snmp_walk) -> HealthMetricsDict:
+        return await collect_ucd_metrics(
+            snmp_get, snmp_walk,
+            tx_err_is_drop=cls._SNMP_TX_ERR_IS_DROP,
+            if_skip=cls._SNMP_SKIP_IF,
+        )
+
+    def get_wan_status(self) -> WANStatusDict:
+        """
+        Returns the status of the device's internet (WAN) uplink.
+
+        Contains:
+
+        * connection_type (string) - e.g. ``"DSL"``, ``"Cable"``, ``"PPPoE"``, ``"DHCP"``
+        * is_connected (bool) - whether the WAN connection is currently established
+        * external_ip (string) - the public IPv4 address assigned to the WAN interface
+        * uptime (int) - seconds since the WAN connection was last (re-)established
+        * bytes_sent (int) - total bytes transmitted on the WAN interface
+        * bytes_received (int) - total bytes received on the WAN interface
+        * max_bitrate_up (int) - upstream sync rate in kbit/s
+        * max_bitrate_down (int) - downstream sync rate in kbit/s
+        * external_ipv6 (string, optional) - the public IPv6 address, if any
+        * link_status (string, optional) - physical line state, e.g. ``"Up"`` / ``"Down"``
+
+        Example::
+
+            {
+                "connection_type": "DSL",
+                "is_connected": True,
+                "external_ip": "203.0.113.7",
+                "uptime": 345600,
+                "bytes_sent": 1234567890,
+                "bytes_received": 9876543210,
+                "max_bitrate_up": 40000,
+                "max_bitrate_down": 250000,
+                "link_status": "Up",
+            }
+        """
+        raise NotImplementedError
+
+    def get_port_forwards(self) -> List[PortForwardDict]:
+        """
+        Returns the configured port forwarding (port mapping) rules.
+
+        Each entry contains:
+
+        * name (string) - the rule's description/name
+        * protocol (string) - ``"TCP"`` or ``"UDP"``
+        * external_port (int) - the WAN-side port
+        * internal_ip (string) - the LAN host the traffic is forwarded to
+        * internal_port (int) - the LAN-side port
+        * enabled (bool) - whether the rule is currently active
+        * remote_host (string, optional) - restricts the forward to a specific
+          remote source address; empty/absent means "any"
+
+        Example::
+
+            [
+                {
+                    "name": "Webserver HTTPS",
+                    "protocol": "TCP",
+                    "external_port": 443,
+                    "internal_ip": "192.168.1.10",
+                    "internal_port": 443,
+                    "enabled": True,
+                }
+            ]
+        """
+        raise NotImplementedError
+
+    def get_hosts(self) -> List[HostDict]:
+        """
+        Returns the list of hosts known to the gateway (LAN clients).
+
+        Each entry contains:
+
+        * mac (string) - the host's MAC address
+        * ip (string) - the host's current IP address
+        * hostname (string) - the host's reported hostname (empty if unknown)
+        * interface_type (string) - how the host is connected, e.g. ``"LAN"``, ``"WLAN"``
+        * is_active (bool) - whether the host is currently online
+        * lease_time_remaining (int, optional) - remaining DHCP lease time in seconds
+
+        Example::
+
+            [
+                {
+                    "mac": "AA:BB:CC:DD:EE:FF",
+                    "ip": "192.168.1.42",
+                    "hostname": "laptop",
+                    "interface_type": "WLAN",
+                    "is_active": True,
+                    "lease_time_remaining": 3600,
+                }
+            ]
+        """
+        raise NotImplementedError
+
+    def get_nat_translations(self) -> List[NATTranslationDict]:
+        """
+        Returns a list of active NAT translation entries.
+
+        See :meth:`napalm_device_types.firewall.FirewallDriver.get_nat_translations`
+        for the entry format. Residential gateways typically derive this from
+        the active port-forwarding/NAT-PT table rather than a live connection
+        tracker; drivers that cannot provide this should return an empty list.
+        """
+        raise NotImplementedError
+
+    def get_vpn_tunnels(self) -> Dict[str, VPNTunnelDict]:
+        """
+        Returns the status of VPN tunnels (e.g. WireGuard road-warrior
+        access, IPsec site-to-site).
+
+        See :meth:`napalm_device_types.firewall.FirewallDriver.get_vpn_tunnels`
+        for the entry format. Drivers that cannot provide this should return
+        an empty dict.
+        """
+        raise NotImplementedError
+
+    def get_wireless_clients(self) -> List[WirelessClientDict]:
+        """
+        Returns the list of wireless clients currently associated with the
+        device's built-in access point(s).
+
+        See :meth:`napalm_device_types.access_point.AccessPointDriver.get_wireless_clients`
+        for the entry format.
+        """
+        raise NotImplementedError
+
+    def get_ssids(self) -> Dict[str, SSIDDict]:
+        """
+        Returns the configured wireless networks (SSIDs).
+
+        See :meth:`napalm_device_types.access_point.AccessPointDriver.get_ssids`
+        for the entry format.
+        """
+        raise NotImplementedError
+
+    def get_radio_status(self) -> Dict[str, RadioStatusDict]:
+        """
+        Returns the status of the device's wireless radios.
+
+        See :meth:`napalm_device_types.access_point.AccessPointDriver.get_radio_status`
+        for the entry format.
+        """
+        raise NotImplementedError

napalm_device_types/storage.py

@@ -11,7 +11,7 @@ Usage::
 """
 
 from typing import Any, Dict, List
-from napalm.base import NetworkDriver
+from napalm_device_types.base import DeviceTypeDriver
 from napalm_device_types.models import (
     DiskPoolDict,
     LogicalVolumeDict,
@@ -25,7 +25,8 @@ from napalm_device_types.models import (
 )
 
 
-class StorageDriver(NetworkDriver):
+class StorageDriver(DeviceTypeDriver):
+    TYPE_LABEL: str = "Storage"
     """
     Abstract intermediate driver for storage appliances and NAS/SAN devices
     (e.g. TrueNAS SCALE/CORE, Synology DSM, QNAP QTS, NetApp ONTAP,

napalm_device_types/switch.py

@@ -10,10 +10,11 @@ Usage::
             ...
 """
 
-from typing import Dict
-from napalm.base import NetworkDriver
+from typing import Dict, List
+from napalm_device_types.base import DeviceTypeDriver
 from napalm_device_types.models import (
     Dot1XPortDict,
+    HealthMetricsDict,
     InterfaceConfigDict,
     MACACLDict,
     PoESummaryDict,
@@ -23,7 +24,8 @@ from napalm_device_types.models import (
 )
 
 
-class SwitchDriver(NetworkDriver):
+class SwitchDriver(DeviceTypeDriver):
+    TYPE_LABEL: str = "Switch"
     """
     Abstract intermediate driver for Ethernet switches.
 
@@ -32,6 +34,15 @@ class SwitchDriver(NetworkDriver):
     operations that concrete drivers must implement.
     """
 
+    @classmethod
+    async def get_health_metrics(cls, snmp_get, snmp_walk) -> HealthMetricsDict:
+        """Collect SNMP health metrics for this switch type.
+
+        Switch vendors use proprietary OIDs — each concrete driver must
+        override this classmethod.
+        """
+        raise NotImplementedError
+
     def get_spanning_tree(self) -> Dict[str, SpanningTreeDict]:
         """
         Returns spanning tree status for each STP instance.
@@ -365,6 +376,27 @@ class SwitchDriver(NetworkDriver):
         """
         raise NotImplementedError
 
+    def set_lag_members(self, lag_name: str, members: List[str]) -> None:
+        """
+        Sets the full member-port list of a LAG/trunk interface.
+
+        Diffs *members* against the LAG's current members (as reported by
+        ``get_interfaces()``'s ``lag_members`` field) and adds/removes ports
+        on the device to match. The LAG itself must already exist on the
+        device; this method only manages its membership.
+
+        :param lag_name: Name of the LAG/trunk interface as returned by
+            ``get_interfaces()`` (e.g. ``"Trk1"``, ``"ch1"``, ``"Lag1"``).
+        :param members: Full desired list of member port names.
+        :raises NotImplementedError: If the driver does not implement this method.
+        :raises ValueError: If *lag_name* does not refer to a valid LAG.
+
+        Example::
+
+            driver.set_lag_members("Lag1", ["GigabitEthernet0/1", "GigabitEthernet0/2"])
+        """
+        raise NotImplementedError
+
     def set_poe_enabled(self, interface: str, enabled: bool) -> None:
         """
         Administratively enables or disables PoE on a single port.

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "napalm-device-types"
-version = "0.1.0"
+version = "0.5.0"
 description = "Abstract device-type base classes for NAPALM drivers"
 readme = "README.md"
 requires-python = ">=3.9"

tests/test_fingerprint_base.py

@@ -0,0 +1,150 @@
+"""Tests for DeviceTypeDriver, FingerprintRule, and PortSpec."""
+
+from __future__ import annotations
+
+import pytest
+from napalm_device_types import (
+    AccessPointDriver,
+    DeviceTypeDriver,
+    FingerprintRule,
+    FirewallDriver,
+    HypervisorDriver,
+    OSDriver,
+    PortSpec,
+    ResidentialGatewayDriver,
+    StorageDriver,
+    SwitchDriver,
+)
+from napalm.base import NetworkDriver
+
+
+# ── Hierarchy ─────────────────────────────────────────────────────────────────
+
+TYPE_DRIVERS = [
+    AccessPointDriver,
+    FirewallDriver,
+    HypervisorDriver,
+    OSDriver,
+    ResidentialGatewayDriver,
+    StorageDriver,
+    SwitchDriver,
+]
+
+
+@pytest.mark.parametrize("cls", TYPE_DRIVERS)
+def test_type_driver_inherits_device_type_driver(cls):
+    assert issubclass(cls, DeviceTypeDriver)
+
+
+@pytest.mark.parametrize("cls", TYPE_DRIVERS)
+def test_type_driver_inherits_network_driver(cls):
+    """Transitiv: DeviceTypeDriver → NetworkDriver muss erhalten bleiben."""
+    assert issubclass(cls, NetworkDriver)
+
+
+def test_device_type_driver_inherits_network_driver():
+    assert issubclass(DeviceTypeDriver, NetworkDriver)
+
+
+# ── Default attributes ────────────────────────────────────────────────────────
+
+def test_defaults_on_device_type_driver():
+    assert DeviceTypeDriver.VENDOR == ""
+    assert DeviceTypeDriver.DRIVER_NAME == ""
+    assert DeviceTypeDriver.PORT_SPECS is None
+    assert DeviceTypeDriver.SNMP_OBJECT_ID_PREFIX is None
+    assert DeviceTypeDriver.SNMP_FINGERPRINT == []
+    assert DeviceTypeDriver.SSH_FINGERPRINT == []
+    assert DeviceTypeDriver.HTTP_FINGERPRINT == []
+    assert DeviceTypeDriver.OUI_PREFIXES == []
+
+
+@pytest.mark.parametrize("cls", TYPE_DRIVERS)
+def test_fingerprint_attributes_inherited(cls):
+    assert hasattr(cls, "HTTP_FINGERPRINT")
+    assert hasattr(cls, "SNMP_FINGERPRINT")
+    assert hasattr(cls, "SSH_FINGERPRINT")
+    assert hasattr(cls, "PORT_SPECS")
+    assert hasattr(cls, "SNMP_OBJECT_ID_PREFIX")
+    assert hasattr(cls, "OUI_PREFIXES")
+
+
+# ── FingerprintRule ───────────────────────────────────────────────────────────
+
+def test_fingerprint_rule_defaults():
+    r = FingerprintRule("fritz!box")
+    assert r.pattern == "fritz!box"
+    assert r.weight == 1.0
+    assert r.mandatory is False
+    assert r.negative is False
+
+
+def test_fingerprint_rule_mandatory():
+    r = FingerprintRule("opnsense", weight=8.0, mandatory=True)
+    assert r.mandatory is True
+    assert r.weight == 8.0
+
+
+def test_fingerprint_rule_negative():
+    r = FingerprintRule("pfsense", weight=5.0, negative=True)
+    assert r.negative is True
+
+
+def test_fingerprint_rule_is_immutable():
+    r = FingerprintRule("test", weight=3.0)
+    with pytest.raises(AttributeError):
+        r.weight = 99.0  # type: ignore[misc]
+
+
+# ── PortSpec ──────────────────────────────────────────────────────────────────
+
+def test_port_spec_defaults():
+    p = PortSpec("https", 8006)
+    assert p.scheme == "https"
+    assert p.port == 8006
+    assert p.paths == ("/",)
+    assert p.weight == 5.0
+    assert p.mandatory is False
+
+
+def test_port_spec_mandatory():
+    p = PortSpec("http", 1400, ("/xml/device_description.xml",), weight=9.0, mandatory=True)
+    assert p.mandatory is True
+    assert p.paths == ("/xml/device_description.xml",)
+
+
+def test_port_spec_is_immutable():
+    p = PortSpec("http", 80)
+    with pytest.raises(AttributeError):
+        p.port = 8080  # type: ignore[misc]
+
+
+# ── Concrete driver subclass ──────────────────────────────────────────────────
+
+def test_concrete_driver_overrides_fingerprint():
+    class MyDriver(FirewallDriver):
+        VENDOR = "Acme"
+        DRIVER_NAME = "acme"
+        HTTP_FINGERPRINT = [FingerprintRule("acme portal", weight=9.0, mandatory=True)]
+        PORT_SPECS = [PortSpec("https", 9443, weight=7.0)]
+        OUI_PREFIXES = ["AA:BB:CC"]
+
+    assert MyDriver.VENDOR == "Acme"
+    assert MyDriver.HTTP_FINGERPRINT[0].pattern == "acme portal"
+    assert MyDriver.PORT_SPECS[0].port == 9443
+    assert MyDriver.OUI_PREFIXES == ["AA:BB:CC"]
+    # base class unaffected
+    assert FirewallDriver.HTTP_FINGERPRINT == []
+    assert FirewallDriver.PORT_SPECS is None
+    assert FirewallDriver.OUI_PREFIXES == []
+
+
+def test_type_label_unaffected_by_refactor():
+    """TYPE_LABEL muss nach dem Refactoring noch korrekt sein."""
+    assert FirewallDriver.TYPE_LABEL == "Firewall"
+    assert SwitchDriver.TYPE_LABEL == "Switch"
+    assert ResidentialGatewayDriver.TYPE_LABEL == "Gateway"
+    assert HypervisorDriver.TYPE_LABEL == "Hypervisor"
+    assert OSDriver.TYPE_LABEL == "OS"
+    assert StorageDriver.TYPE_LABEL == "Storage"
+    assert AccessPointDriver.TYPE_LABEL == "Access Point"

tests/test_type_label.py

@@ -0,0 +1,53 @@
+"""Tests for TYPE_LABEL class attribute on all base driver classes."""
+
+import pytest
+from napalm_device_types import (
+    AccessPointDriver,
+    FirewallDriver,
+    HypervisorDriver,
+    OSDriver,
+    ResidentialGatewayDriver,
+    StorageDriver,
+    SwitchDriver,
+)
+
+
+BASE_CLASSES = [
+    (AccessPointDriver, "Access Point"),
+    (FirewallDriver, "Firewall"),
+    (HypervisorDriver, "Hypervisor"),
+    (OSDriver, "OS"),
+    (ResidentialGatewayDriver, "Gateway"),
+    (StorageDriver, "Storage"),
+    (SwitchDriver, "Switch"),
+]
+
+
+@pytest.mark.parametrize("cls, expected_label", BASE_CLASSES)
+def test_type_label_present(cls, expected_label):
+    assert hasattr(cls, "TYPE_LABEL"), f"{cls.__name__} is missing TYPE_LABEL"
+
+
+@pytest.mark.parametrize("cls, expected_label", BASE_CLASSES)
+def test_type_label_value(cls, expected_label):
+    assert cls.TYPE_LABEL == expected_label
+
+
+@pytest.mark.parametrize("cls, _", BASE_CLASSES)
+def test_type_label_is_nonempty_string(cls, _):
+    assert isinstance(cls.TYPE_LABEL, str) and cls.TYPE_LABEL
+
+
+def test_subclass_inherits_type_label():
+    class MySwitch(SwitchDriver):
+        pass
+
+    assert MySwitch.TYPE_LABEL == "Switch"
+
+
+def test_subclass_can_override_type_label():
+    class MySpecialOS(OSDriver):
+        TYPE_LABEL = "Linux"
+
+    assert MySpecialOS.TYPE_LABEL == "Linux"
+    assert OSDriver.TYPE_LABEL == "OS"  # base class unaffected