feat: add service management and package update APIs to AccessPointDriver
- Add ServiceDict and UpdateDict TypedDicts to models - Add get_services() / manage_service() abstract methods for init-system interaction - Add get_available_updates() / apply_updates() for package upgrade workflows - Add _filter_interfaces() helper to exclude lo and phy* interfaces from interface dicts - Extend WirelessClientDict with optional ip, hostname, and lease_end fields - Add optional description field to VPNTunnelDict - Bump version 0.1.0 → 0.2.0
This commit is contained in:
@@ -20,8 +20,10 @@ from napalm_device_types.models import (
|
||||
MeshPeerDict,
|
||||
PackageDict,
|
||||
RadioStatusDict,
|
||||
ServiceDict,
|
||||
SSIDBridgeDict,
|
||||
SSIDDict,
|
||||
UpdateDict,
|
||||
WirelessClientDict,
|
||||
WirelessConfigDict,
|
||||
)
|
||||
@@ -35,6 +37,23 @@ class AccessPointDriver(NetworkDriver):
|
||||
access-point-specific operations that concrete drivers must implement.
|
||||
"""
|
||||
|
||||
# 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"})
|
||||
|
||||
# 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",)
|
||||
|
||||
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_wireless_clients(self) -> List[WirelessClientDict]:
|
||||
"""
|
||||
Returns a list of wireless clients currently associated with this
|
||||
@@ -498,6 +517,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.
|
||||
|
||||
@@ -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):
|
||||
@@ -289,6 +310,7 @@ class VPNTunnelDict(TypedDict):
|
||||
uptime: int
|
||||
bytes_in: int
|
||||
bytes_out: int
|
||||
description: NotRequired[str] # human-readable tunnel description / name
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
+1
-1
@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
|
||||
|
||||
[project]
|
||||
name = "napalm-device-types"
|
||||
version = "0.1.0"
|
||||
version = "0.2.0"
|
||||
description = "Abstract device-type base classes for NAPALM drivers"
|
||||
readme = "README.md"
|
||||
requires-python = ">=3.9"
|
||||
|
||||
Reference in New Issue
Block a user