initial commit

.gitignore

@@ -0,0 +1,45 @@
+# Python
+__pycache__/
+*.py[cod]
+*.pyo
+*.pyd
+*.so
+*.egg
+*.egg-info/
+dist/
+build/
+.eggs/
+wheels/
+
+# Virtual environments
+.venv/
+venv/
+env/
+.env
+
+# Packaging
+*.tar.gz
+*.whl
+MANIFEST
+
+# Testing
+.pytest_cache/
+.coverage
+.coverage.*
+htmlcov/
+coverage.xml
+
+# Type checking
+.mypy_cache/
+.ruff_cache/
+
+# IDEs
+.vscode/
+.idea/
+*.swp
+*~
+napalm-openwrt.code-workspace
+
+# OS
+.DS_Store
+Thumbs.db

README.md

@@ -0,0 +1,117 @@
+# napalm-openwrt
+
+NAPALM community driver for **OpenWrt** routers and access-points.
+
+Communicates over SSH using [Netmiko](https://github.com/ktbyers/netmiko) (`linux` device type).  
+Requires OpenWrt **19.07** or newer.
+
+## Tested devices
+
+| Model | OpenWrt version | Tested |
+|---|---|---|
+| TP-Link TL-WR1043N/ND v5 | 23.05.3 | ✅ |
+
+> Contributions for additional devices and firmware versions are welcome.
+
+## Requirements
+
+| Dependency | Minimum version |
+|---|---|
+| Python | 3.8 |
+| NAPALM | 4.0 |
+| Netmiko | 4.0 |
+
+## Installation
+
+From source:
+
+```bash
+git clone https://github.com/napalm-automation-community/napalm-openwrt
+cd napalm-openwrt
+pip install -e .
+```
+
+## Quick start
+
+```python
+from napalm import get_network_driver
+
+driver = get_network_driver("openwrt")
+device = driver("192.168.1.1", "root", "")
+
+device.open()
+
+facts = device.get_facts()
+print(facts)
+
+interfaces = device.get_interfaces()
+print(interfaces)
+
+device.close()
+```
+
+## Supported NAPALM getters
+
+| Getter | Supported | Notes |
+|---|---|---|
+| `get_facts` | ✅ | Uses `/etc/openwrt_release`, `/tmp/sysinfo/model`, `/proc/uptime` |
+| `get_interfaces` | ✅ | Uses `ip link show` |
+| `get_interfaces_ip` | ✅ | Uses `ip addr show` |
+| `get_interfaces_counters` | ✅ | Uses `/proc/net/dev` |
+| `get_arp_table` | ✅ | Uses `ip neigh show` |
+| `get_mac_address_table` | ✅ | Uses `bridge fdb show` |
+| `get_config` | ✅ | Uses `uci export` |
+| `get_environment` | ✅ | CPU from `/proc/stat`, memory from `/proc/meminfo` |
+| `get_lldp_neighbors` | ✅ | Requires `lldpd` package installed on device |
+| `get_lldp_neighbors_detail` | ✅ | Requires `lldpd` package installed on device |
+
+## Configuration management
+
+Configuration is managed via [UCI](https://openwrt.org/docs/guide-user/base-system/uci)
+(Unified Configuration Interface).
+
+### Merge candidate
+
+```python
+device.load_merge_candidate(config="""
+uci set system.@system[0].hostname='my-router'
+uci set network.lan.ipaddr='10.0.0.1'
+""")
+
+print(device.compare_config())
+device.commit_config()
+```
+
+### Replace candidate
+
+```python
+with open("full-config.uci") as f:
+    device.load_replace_candidate(config=f.read())
+
+print(device.compare_config())
+device.commit_config()
+```
+
+### Rollback
+
+```python
+# Reverts to the config state before the last commit_config call
+device.rollback()
+```
+
+## Development
+
+```bash
+# Create and activate a virtual environment
+python -m venv .venv
+source .venv/bin/activate
+
+# Install with dev dependencies
+pip install -e ".[dev]"
+
+# Run tests
+pytest
+
+# Lint
+ruff check napalm_openwrt/
+```

napalm_openwrt/__init__.py

@@ -0,0 +1,5 @@
+"""NAPALM driver for OpenWrt routers/access-points."""
+
+from napalm_openwrt.openwrt import OpenWrtDriver
+
+__all__ = ["OpenWrtDriver"]

pyproject.toml

@@ -0,0 +1,54 @@
+[build-system]
+requires = ["setuptools>=64", "wheel"]
+build-backend = "setuptools.build_meta"
+
+[project]
+name = "napalm-openwrt"
+version = "0.1.0"
+description = "NAPALM driver for OpenWrt routers/access-points"
+readme = "README.md"
+license = { text = "Apache-2.0" }
+requires-python = ">=3.8"
+authors = [
+    { name = "Christian Manivong" },
+]
+classifiers = [
+    "Topic :: Utilities",
+    "License :: OSI Approved :: Apache Software License",
+    "Programming Language :: Python :: 3",
+    "Programming Language :: Python :: 3.8",
+    "Programming Language :: Python :: 3.9",
+    "Programming Language :: Python :: 3.10",
+    "Programming Language :: Python :: 3.11",
+    "Programming Language :: Python :: 3.12",
+    "Operating System :: POSIX :: Linux",
+    "Operating System :: MacOS",
+]
+dependencies = [
+    "napalm>=4.0.0",
+    "napalm_device_types>=0.1.0",
+    "netmiko>=4.0.0",
+    "netaddr",
+]
+
+[project.optional-dependencies]
+dev = [
+    "pytest",
+    "pytest-cov",
+    "black",
+    "ruff",
+]
+
+[project.entry-points."napalm.drivers"]
+openwrt = "napalm_openwrt:OpenWrtDriver"
+
+[project.urls]
+Repository = "https://github.com/napalm-automation-community/napalm-openwrt"
+
+[tool.setuptools.packages.find]
+where = ["."]
+include = ["napalm_openwrt*"]
+
+[tool.ruff]
+line-length = 100
+target-version = "py38"

requirements.txt

@@ -0,0 +1,4 @@
+napalm>=4.0.0
+napalm_device_types>=0.1.0
+netmiko>=4.0.0
+netaddr

tests/unit/test_driver.py

@@ -0,0 +1,835 @@
+"""Unit tests for OpenWrtDriver — no real device required."""
+
+import pytest
+from unittest.mock import MagicMock, patch
+
+from napalm_openwrt.openwrt import OpenWrtDriver
+
+
+# ---------------------------------------------------------------------------
+# Fixtures
+# ---------------------------------------------------------------------------
+
+@pytest.fixture
+def driver():
+    """Return a driver instance with a mocked Netmiko connection."""
+    with patch("napalm_openwrt.openwrt.ConnectHandler"):
+        drv = OpenWrtDriver(
+            hostname="192.168.1.1",
+            username="root",
+            password="",
+        )
+        drv.device = MagicMock()
+        yield drv
+
+
+# ---------------------------------------------------------------------------
+# Sample command output (as would be returned by the device)
+# ---------------------------------------------------------------------------
+
+OPENWRT_RELEASE = """\
+DISTRIB_ID="OpenWrt"
+DISTRIB_RELEASE="23.05.3"
+DISTRIB_REVISION="r23809-234f1a2efa"
+DISTRIB_TARGET="ath79/generic"
+DISTRIB_ARCH="mips_24kc"
+DISTRIB_CODENAME="Restoring Earth"
+DISTRIB_TAINTS=""
+"""
+
+SYSINFO_MODEL = "TP-Link TL-WR1043N/ND v5"
+
+UPTIME = "352467.12 345823.44"
+
+IP_LINK_SHOW = """\
+1: lo: <LOOPBACK,UP,LOWER_UP> mtu 65536 qdisc noqueue state UNKNOWN mode DEFAULT group default qlen 1000
+    link/loopback 00:00:00:00:00:00 brd 00:00:00:00:00:00
+2: eth0: <BROADCAST,MULTICAST,UP,LOWER_UP> mtu 1500 qdisc fq_codel state UP mode DEFAULT group default qlen 1000
+    link/ether b0:95:75:aa:bb:cc brd ff:ff:ff:ff:ff:ff
+3: eth1: <BROADCAST,MULTICAST> mtu 1500 qdisc noop state DOWN mode DEFAULT group default qlen 1000
+    link/ether b0:95:75:aa:bb:dd brd ff:ff:ff:ff:ff:ff
+4: br-lan: <BROADCAST,MULTICAST,UP,LOWER_UP> mtu 1500 qdisc noqueue state UP mode DEFAULT group default qlen 1000
+    link/ether b0:95:75:aa:bb:cc brd ff:ff:ff:ff:ff:ff
+"""
+
+IP_ADDR_SHOW = """\
+1: lo: <LOOPBACK,UP,LOWER_UP> mtu 65536 qdisc noqueue state UNKNOWN group default qlen 1000
+    link/loopback 00:00:00:00:00:00 brd 00:00:00:00:00:00
+    inet 127.0.0.1/8 scope host lo
+       valid_lft forever preferred_lft forever
+2: eth0: <BROADCAST,MULTICAST,UP,LOWER_UP> mtu 1500 qdisc fq_codel state UP group default qlen 1000
+    link/ether b0:95:75:aa:bb:cc brd ff:ff:ff:ff:ff:ff
+4: br-lan: <BROADCAST,MULTICAST,UP,LOWER_UP> mtu 1500 qdisc noqueue state UP group default qlen 1000
+    link/ether b0:95:75:aa:bb:cc brd ff:ff:ff:ff:ff:ff
+    inet 192.168.1.1/24 brd 192.168.1.255 scope global br-lan
+       valid_lft forever preferred_lft forever
+    inet6 fd00::1/64 scope global
+       valid_lft forever preferred_lft forever
+"""
+
+IP_NEIGH_SHOW = """\
+192.168.1.100 dev br-lan lladdr aa:bb:cc:dd:ee:ff REACHABLE
+192.168.1.101 dev br-lan lladdr 11:22:33:44:55:66 STALE
+192.168.1.102 dev br-lan  FAILED
+"""
+
+BRIDGE_FDB = """\
+aa:bb:cc:dd:ee:ff dev br-lan master br-lan permanent
+11:22:33:44:55:66 dev eth0 vlan 1 master br-lan
+33:33:00:00:00:01 dev br-lan self permanent
+"""
+
+UCI_EXPORT = """\
+package system
+
+config system
+\toption hostname 'OpenWrt'
+\toption timezone 'UTC'
+
+package network
+
+config interface 'loopback'
+\toption device 'lo'
+\toption proto 'static'
+\toption ipaddr '127.0.0.1'
+\toption netmask '255.0.0.0'
+
+config interface 'lan'
+\toption device 'br-lan'
+\toption proto 'static'
+\toption ipaddr '192.168.1.1'
+\toption netmask '255.255.255.0'
+"""
+
+LLDPCTL_KV = """\
+lldp.eth0.via=LLDP
+lldp.eth0.rid=1
+lldp.eth0.age=0 day, 01:23:45
+lldp.eth0.chassis.mac=00:aa:bb:cc:dd:ee
+lldp.eth0.chassis.name=core-router
+lldp.eth0.chassis.descr=RouterOS 7.x
+lldp.eth0.chassis.mgmt-ip=10.0.0.1
+lldp.eth0.chassis.cap.available=Router, Bridge
+lldp.eth0.chassis.cap.enabled=Router
+lldp.eth0.port.ifname=ether1
+lldp.eth0.port.descr=uplink
+"""
+
+PROC_NET_DEV = """\
+Inter-|   Receive                                                |  Transmit
+ face |bytes    packets errs drop fifo frame compressed multicast|bytes    packets errs drop fifo colls carrier compressed
+    lo:    1234      12    0    0    0     0          0         0     1234      12    0    0    0     0       0          0
+  eth0: 9876543   12345    0    0    0     0          0       100  1234567    9876    0    0    0     0       0          0
+br-lan: 8765432   11234    0    0    0     0          0        50  1123456    8765    0    0    0     0       0          0
+"""
+
+HOSTNAME = "OpenWrt"
+
+
+# ---------------------------------------------------------------------------
+# Tests
+# ---------------------------------------------------------------------------
+
+class TestGetFacts:
+    def _make_send(self):
+        """Return a _send_command mock that handles both str and list args."""
+        def _send(cmd, **kw):
+            key = cmd[0] if isinstance(cmd, list) else cmd
+            if "openwrt_release" in key:
+                return OPENWRT_RELEASE
+            if "sysinfo/model" in key:
+                return SYSINFO_MODEL
+            if "uptime" in key:
+                return UPTIME
+            if "hostname" in key.lower() or "system.@system" in key:
+                return HOSTNAME
+            if "ip link" in key:
+                return IP_LINK_SHOW
+            return ""
+        return _send
+
+    def test_returns_required_keys(self, driver):
+        driver._send_command = self._make_send()
+
+        facts = driver.get_facts()
+        assert set(facts.keys()) == {
+            "vendor", "model", "hostname", "fqdn", "os_version",
+            "serial_number", "uptime", "interface_list",
+        }
+
+    def test_vendor(self, driver):
+        driver._send_command = self._make_send()
+        assert driver.get_facts()["vendor"] == "OpenWrt"
+
+    def test_os_version(self, driver):
+        driver._send_command = self._make_send()
+        assert driver.get_facts()["os_version"] == "23.05.3"
+
+    def test_uptime_parsed(self, driver):
+        driver._send_command = self._make_send()
+        facts = driver.get_facts()
+        assert facts["uptime"] == pytest.approx(352467.12)
+
+
+class TestParseOpenwrtRelease:
+    def test_parses_release(self):
+        result = OpenWrtDriver._parse_openwrt_release(OPENWRT_RELEASE)
+        assert result["DISTRIB_RELEASE"] == "23.05.3"
+        assert result["DISTRIB_ID"] == "OpenWrt"
+        assert result["DISTRIB_TARGET"] == "ath79/generic"
+
+
+class TestGetInterfaces:
+    def test_returns_dict(self, driver):
+        driver._send_command = lambda cmd, **kw: IP_LINK_SHOW
+        result = driver.get_interfaces()
+        assert isinstance(result, dict)
+
+    def test_eth0_is_up(self, driver):
+        driver._send_command = lambda cmd, **kw: IP_LINK_SHOW
+        result = driver.get_interfaces()
+        assert "eth0" in result
+        assert result["eth0"]["is_up"] is True
+        assert result["eth0"]["is_enabled"] is True
+
+    def test_eth1_is_down(self, driver):
+        driver._send_command = lambda cmd, **kw: IP_LINK_SHOW
+        result = driver.get_interfaces()
+        assert "eth1" in result
+        assert result["eth1"]["is_up"] is False
+
+    def test_mac_address_populated(self, driver):
+        driver._send_command = lambda cmd, **kw: IP_LINK_SHOW
+        result = driver.get_interfaces()
+        assert result["eth0"]["mac_address"] != ""
+
+    def test_required_keys(self, driver):
+        driver._send_command = lambda cmd, **kw: IP_LINK_SHOW
+        result = driver.get_interfaces()
+        for iface_data in result.values():
+            assert set(iface_data.keys()) >= {
+                "is_up", "is_enabled", "description",
+                "last_flapped", "speed", "mtu", "mac_address",
+            }
+
+
+class TestGetInterfacesIP:
+    def test_br_lan_ipv4(self, driver):
+        driver._send_command = lambda cmd, **kw: IP_ADDR_SHOW
+        result = driver.get_interfaces_ip()
+        assert "br-lan" in result
+        assert "192.168.1.1" in result["br-lan"]["ipv4"]
+        assert result["br-lan"]["ipv4"]["192.168.1.1"]["prefix_length"] == 24
+
+    def test_br_lan_ipv6(self, driver):
+        driver._send_command = lambda cmd, **kw: IP_ADDR_SHOW
+        result = driver.get_interfaces_ip()
+        assert "ipv6" in result["br-lan"]
+        assert "fd00::1" in result["br-lan"]["ipv6"]
+
+
+class TestGetArpTable:
+    def test_returns_list(self, driver):
+        driver._send_command = lambda cmd, **kw: IP_NEIGH_SHOW
+        result = driver.get_arp_table()
+        assert isinstance(result, list)
+
+    def test_entries_count(self, driver):
+        driver._send_command = lambda cmd, **kw: IP_NEIGH_SHOW
+        result = driver.get_arp_table()
+        # FAILED entry should be skipped
+        assert len(result) == 2
+
+    def test_entry_keys(self, driver):
+        driver._send_command = lambda cmd, **kw: IP_NEIGH_SHOW
+        result = driver.get_arp_table()
+        for entry in result:
+            assert set(entry.keys()) >= {"interface", "mac", "ip", "age"}
+
+    def test_ip_value(self, driver):
+        driver._send_command = lambda cmd, **kw: IP_NEIGH_SHOW
+        result = driver.get_arp_table()
+        ips = {e["ip"] for e in result}
+        assert "192.168.1.100" in ips
+        assert "192.168.1.101" in ips
+
+
+class TestGetMacAddressTable:
+    def test_returns_list(self, driver):
+        driver._send_command = lambda cmd, **kw: BRIDGE_FDB
+        result = driver.get_mac_address_table()
+        assert isinstance(result, list)
+
+    def test_multicast_skipped(self, driver):
+        driver._send_command = lambda cmd, **kw: BRIDGE_FDB
+        result = driver.get_mac_address_table()
+        macs = [e["mac"] for e in result]
+        assert not any("33:33" in m for m in macs)
+
+    def test_vlan_parsed(self, driver):
+        driver._send_command = lambda cmd, **kw: BRIDGE_FDB
+        result = driver.get_mac_address_table()
+        vlan1_entries = [e for e in result if e["vlan"] == 1]
+        assert len(vlan1_entries) >= 1
+
+
+class TestGetConfig:
+    def test_returns_uci_export(self, driver):
+        driver._send_command = lambda cmd, **kw: UCI_EXPORT
+        result = driver.get_config()
+        assert "running" in result
+        assert "startup" in result
+        assert "candidate" in result
+        assert "package" in result["running"]
+
+    def test_candidate_always_empty(self, driver):
+        driver._send_command = lambda cmd, **kw: UCI_EXPORT
+        result = driver.get_config()
+        assert result["candidate"] == ""
+
+
+class TestGetLldpNeighbors:
+    def test_returns_dict(self, driver):
+        driver._send_command = lambda cmd, **kw: LLDPCTL_KV
+        result = driver.get_lldp_neighbors()
+        assert isinstance(result, dict)
+
+    def test_eth0_neighbor(self, driver):
+        driver._send_command = lambda cmd, **kw: LLDPCTL_KV
+        result = driver.get_lldp_neighbors()
+        assert "eth0" in result
+        assert result["eth0"][0]["hostname"] == "core-router"
+        assert result["eth0"][0]["port"] == "ether1"
+
+
+BRIDGE_VLAN_SHOW = """\
+port              vlan-id
+eth0              1 PVID Egress Untagged
+                  10
+                  20
+br-lan            1 PVID Egress Untagged
+                  10
+                  20
+eth1              20 PVID Egress Untagged
+"""
+
+UCI_NETWORK_VLANS = """\
+network.@bridge-vlan[0]=bridge-vlan
+network.@bridge-vlan[0].device='br-lan'
+network.@bridge-vlan[0].vlan='10'
+network.@bridge-vlan[0].name='management'
+network.@bridge-vlan[1]=bridge-vlan
+network.@bridge-vlan[1].device='br-lan'
+network.@bridge-vlan[1].vlan='20'
+network.@bridge-vlan[1].name='iot'
+"""
+
+
+class TestGetVlans:
+    def _send(self, cmd, **kw):
+        if "bridge vlan" in cmd:
+            return BRIDGE_VLAN_SHOW
+        if "uci show network" in cmd:
+            return UCI_NETWORK_VLANS
+        return ""
+
+    def test_returns_dict(self, driver):
+        driver._send_command = self._send
+        result = driver.get_vlans()
+        assert isinstance(result, dict)
+
+    def test_vlan_ids_present(self, driver):
+        driver._send_command = self._send
+        result = driver.get_vlans()
+        assert "1" in result
+        assert "10" in result
+        assert "20" in result
+
+    def test_required_keys(self, driver):
+        driver._send_command = self._send
+        result = driver.get_vlans()
+        for vlan_data in result.values():
+            assert "name" in vlan_data
+            assert "interfaces" in vlan_data
+
+    def test_interfaces_for_vlan10(self, driver):
+        driver._send_command = self._send
+        result = driver.get_vlans()
+        assert set(result["10"]["interfaces"]) == {"eth0", "br-lan"}
+
+    def test_interfaces_for_vlan20(self, driver):
+        driver._send_command = self._send
+        result = driver.get_vlans()
+        assert set(result["20"]["interfaces"]) == {"eth0", "br-lan", "eth1"}
+
+    def test_uci_names_applied(self, driver):
+        driver._send_command = self._send
+        result = driver.get_vlans()
+        assert result["10"]["name"] == "management"
+        assert result["20"]["name"] == "iot"
+
+    def test_vlan_without_uci_name_is_empty_string(self, driver):
+        driver._send_command = self._send
+        result = driver.get_vlans()
+        assert result["1"]["name"] == ""
+
+    def test_no_duplicate_interfaces(self, driver):
+        driver._send_command = self._send
+        result = driver.get_vlans()
+        for vlan_data in result.values():
+            assert len(vlan_data["interfaces"]) == len(set(vlan_data["interfaces"]))
+
+    def test_empty_bridge_output(self, driver):
+        driver._send_command = lambda cmd, **kw: ""
+        result = driver.get_vlans()
+        assert result == {}
+
+
+class TestConfigManagement:
+    def test_load_merge_candidate(self, driver):
+        driver.load_merge_candidate(config="uci set system.@system[0].hostname='MyRouter'")
+        assert driver._candidate_config is not None
+        assert driver._candidate_mode == "merge"
+
+    def test_load_replace_candidate(self, driver):
+        driver.load_replace_candidate(config=UCI_EXPORT)
+        assert driver._candidate_config is not None
+        assert driver._candidate_mode == "replace"
+
+    def test_discard_config(self, driver):
+        driver.load_merge_candidate(config="uci set system.@system[0].hostname='test'")
+        driver.discard_config()
+        assert driver._candidate_config is None
+        assert not driver.has_pending_commit()
+
+    def test_compare_merge_candidate(self, driver):
+        driver._send_command = lambda cmd, **kw: UCI_EXPORT
+        driver.load_merge_candidate(config="uci set system.@system[0].hostname='test'")
+        diff = driver.compare_config()
+        assert diff.startswith("+")
+
+    def test_compare_no_candidate(self, driver):
+        assert driver.compare_config() == ""
+
+    def test_has_pending_commit_false_initially(self, driver):
+        assert not driver.has_pending_commit()
+
+    def test_has_pending_commit_true_after_load(self, driver):
+        driver.load_merge_candidate(config="uci set system.@system[0].hostname='test'")
+        assert driver.has_pending_commit()
+
+
+# ---------------------------------------------------------------------------
+# Sample data for new methods
+# ---------------------------------------------------------------------------
+
+PASSWD = """\
+root:$1$xyz:0:0:root:/root:/bin/ash
+daemon:*:1:1:daemon:/var:/bin/false
+nobody:*:65534:65534:nobody:/var:/bin/false
+alice:$6$abc:1001:1001:Alice:/home/alice:/bin/ash
+"""
+
+ROOT_AUTHORIZED_KEYS = "ssh-rsa AAAAB3NzaC1yc2EAAAADAQABroot@host"
+ALICE_AUTHORIZED_KEYS = "ssh-ed25519 AAAAC3NzaC1lZDI1NTE5alice@host"
+
+UCI_SYSTEM_NTP = """\
+system.@system[0]=system
+system.@system[0].hostname='OpenWrt'
+system.@system[0].timezone='UTC'
+system.ntp=timeserver
+system.ntp.server='0.openwrt.pool.ntp.org 1.openwrt.pool.ntp.org 2.openwrt.pool.ntp.org'
+system.ntp.enabled='1'
+system.ntp.enable_server='0'
+"""
+
+NTPQ_OUTPUT = """\
+     remote           refid      st t when poll reach   delay   offset  jitter
+==============================================================================
+*188.114.101.4   188.114.100.1    4 u  107  256  377  164.228 -13.866   2.695
++37.187.56.220   192.53.103.108   2 u   22   64  377   30.112   5.123   1.100
+"""
+
+UCI_SNMPD = """\
+snmpd.@agent[0]=agent
+snmpd.@agent[0].agentaddress='UDP:161'
+snmpd.@com2sec[0]=com2sec
+snmpd.@com2sec[0].secname='public'
+snmpd.@com2sec[0].source='default'
+snmpd.@com2sec[0].community='public'
+snmpd.@com2sec[1]=com2sec
+snmpd.@com2sec[1].secname='private'
+snmpd.@com2sec[1].source='10.0.0.0/8'
+snmpd.@com2sec[1].community='private'
+snmpd.@system[0]=system
+snmpd.@system[0].sysContact='admin@example.com'
+snmpd.@system[0].sysLocation='Server Room'
+snmpd.@system[0].sysName='MyRouter'
+"""
+
+PING_SUCCESS = """\
+PING 8.8.8.8 (8.8.8.8): 56 data bytes
+64 bytes from 8.8.8.8: seq=0 ttl=120 time=7.123 ms
+64 bytes from 8.8.8.8: seq=1 ttl=120 time=6.987 ms
+64 bytes from 8.8.8.8: seq=2 ttl=120 time=7.234 ms
+--- 8.8.8.8 ping statistics ---
+3 packets transmitted, 3 packets received, 0% packet loss
+round-trip min/avg/max = 6.987/7.115/7.234 ms
+"""
+
+PING_LOSS = """\
+PING 10.0.0.99 (10.0.0.99): 56 data bytes
+--- 10.0.0.99 ping statistics ---
+3 packets transmitted, 0 packets received, 100% packet loss
+"""
+
+PING_ERROR = "ping: bad address 'invalid.host'"
+
+IPV6_NEIGH = """\
+2001:db8::1 dev eth0 lladdr aa:bb:cc:dd:ee:ff REACHABLE
+fe80::1 dev br-lan lladdr 11:22:33:44:55:66 STALE
+2001:db8::2 dev eth0  FAILED
+"""
+
+IP_ROUTE_SHOW = """\
+default via 192.168.1.1 dev br-wan proto dhcp src 203.0.113.1 metric 100
+192.168.1.0/24 dev br-lan proto kernel scope link src 192.168.1.1
+10.0.0.0/8 via 192.168.1.254 dev br-wan proto static metric 50
+"""
+
+TRACEROUTE_OUTPUT = """\
+traceroute to 8.8.8.8 (8.8.8.8), 30 hops max, 38 byte packets
+ 1  192.168.1.1 (192.168.1.1)  1.123 ms  1.456 ms  1.789 ms
+ 2  10.0.0.1 (10.0.0.1)  5.123 ms  5.456 ms  5.789 ms
+ 3  * * *
+ 4  8.8.8.8 (8.8.8.8)  7.001 ms  6.999 ms  7.100 ms
+"""
+
+TRACEROUTE_ERROR = "traceroute: unknown host invalid.host"
+
+IP_NETNS_LIST = """\
+vpn (id: 1)
+mgmt (id: 2)
+"""
+
+
+# ---------------------------------------------------------------------------
+# Tests for new methods
+# ---------------------------------------------------------------------------
+
+class TestCli:
+    def test_returns_dict_keyed_by_command(self, driver):
+        driver._send_command = lambda cmd, **kw: f"output of {cmd}"
+        result = driver.cli(["uname -a", "uptime"])
+        assert set(result.keys()) == {"uname -a", "uptime"}
+
+    def test_output_content(self, driver):
+        driver._send_command = lambda cmd, **kw: "Linux OpenWrt"
+        result = driver.cli(["uname -a"])
+        assert result["uname -a"] == "Linux OpenWrt"
+
+    def test_empty_command_list(self, driver):
+        result = driver.cli([])
+        assert result == {}
+
+
+class TestGetUsers:
+    def _send(self, cmd, **kw):
+        key = cmd[0] if isinstance(cmd, list) else cmd
+        if "/etc/passwd" in key:
+            return PASSWD
+        if "root/.ssh/authorized_keys" in key or "dropbear/authorized_keys" in key:
+            return ROOT_AUTHORIZED_KEYS
+        if "alice" in key:
+            return ALICE_AUTHORIZED_KEYS
+        return ""
+
+    def test_returns_dict(self, driver):
+        driver._send_command = self._send
+        assert isinstance(driver.get_users(), dict)
+
+    def test_root_level_15(self, driver):
+        driver._send_command = self._send
+        users = driver.get_users()
+        assert "root" in users
+        assert users["root"]["level"] == 15
+
+    def test_regular_user_level_1(self, driver):
+        driver._send_command = self._send
+        users = driver.get_users()
+        assert "alice" in users
+        assert users["alice"]["level"] == 1
+
+    def test_system_accounts_excluded(self, driver):
+        driver._send_command = self._send
+        users = driver.get_users()
+        assert "daemon" not in users
+        assert "nobody" not in users
+
+    def test_root_has_ssh_key(self, driver):
+        driver._send_command = self._send
+        users = driver.get_users()
+        assert len(users["root"]["sshkeys"]) >= 1
+        assert users["root"]["sshkeys"][0].startswith("ssh-rsa")
+
+    def test_required_keys(self, driver):
+        driver._send_command = self._send
+        users = driver.get_users()
+        for data in users.values():
+            assert "level" in data
+            assert "password" in data
+            assert "sshkeys" in data
+
+
+class TestGetNtpServers:
+    def test_returns_dict(self, driver):
+        driver._send_command = lambda cmd, **kw: UCI_SYSTEM_NTP
+        result = driver.get_ntp_servers()
+        assert isinstance(result, dict)
+
+    def test_servers_present(self, driver):
+        driver._send_command = lambda cmd, **kw: UCI_SYSTEM_NTP
+        result = driver.get_ntp_servers()
+        assert "0.openwrt.pool.ntp.org" in result
+        assert "1.openwrt.pool.ntp.org" in result
+        assert "2.openwrt.pool.ntp.org" in result
+
+    def test_empty_when_no_ntp(self, driver):
+        driver._send_command = lambda cmd, **kw: ""
+        assert driver.get_ntp_servers() == {}
+
+
+class TestGetNtpStats:
+    def test_returns_list(self, driver):
+        driver._send_command = lambda cmd, **kw: NTPQ_OUTPUT
+        result = driver.get_ntp_stats()
+        assert isinstance(result, list)
+
+    def test_synchronized_entry(self, driver):
+        driver._send_command = lambda cmd, **kw: NTPQ_OUTPUT
+        result = driver.get_ntp_stats()
+        synced = [e for e in result if e["synchronized"]]
+        assert len(synced) == 1
+        assert synced[0]["remote"] == "188.114.101.4"
+
+    def test_required_keys(self, driver):
+        driver._send_command = lambda cmd, **kw: NTPQ_OUTPUT
+        result = driver.get_ntp_stats()
+        for entry in result:
+            assert set(entry.keys()) >= {
+                "remote", "referenceid", "synchronized", "stratum",
+                "type", "when", "hostpoll", "reachability", "delay", "offset", "jitter",
+            }
+
+    def test_empty_on_no_tool(self, driver):
+        driver._send_command = lambda cmd, **kw: "sh: ntpq: not found"
+        result = driver.get_ntp_stats()
+        assert result == []
+
+
+class TestGetSnmpInformation:
+    def test_returns_dict(self, driver):
+        driver._send_command = lambda cmd, **kw: UCI_SNMPD
+        result = driver.get_snmp_information()
+        assert isinstance(result, dict)
+
+    def test_required_keys(self, driver):
+        driver._send_command = lambda cmd, **kw: UCI_SNMPD
+        result = driver.get_snmp_information()
+        assert set(result.keys()) >= {"chassis_id", "community", "contact", "location"}
+
+    def test_contact_and_location(self, driver):
+        driver._send_command = lambda cmd, **kw: UCI_SNMPD
+        result = driver.get_snmp_information()
+        assert result["contact"] == "admin@example.com"
+        assert result["location"] == "Server Room"
+
+    def test_community_entries(self, driver):
+        driver._send_command = lambda cmd, **kw: UCI_SNMPD
+        result = driver.get_snmp_information()
+        assert "public" in result["community"]
+        assert "private" in result["community"]
+
+    def test_community_mode(self, driver):
+        driver._send_command = lambda cmd, **kw: UCI_SNMPD
+        result = driver.get_snmp_information()
+        assert result["community"]["public"]["mode"] == "ro"
+        assert result["community"]["private"]["mode"] == "rw"
+
+
+class TestPing:
+    def test_success_result(self, driver):
+        driver._send_command = lambda cmd, **kw: PING_SUCCESS
+        result = driver.ping("8.8.8.8")
+        assert "success" in result
+        assert result["success"]["probes_sent"] == 3
+        assert result["success"]["packet_loss"] == 0
+
+    def test_rtt_values(self, driver):
+        driver._send_command = lambda cmd, **kw: PING_SUCCESS
+        result = driver.ping("8.8.8.8")
+        s = result["success"]
+        assert s["rtt_min"] == 6.987
+        assert s["rtt_max"] == 7.234
+        assert s["rtt_avg"] == 7.115
+
+    def test_probe_results(self, driver):
+        driver._send_command = lambda cmd, **kw: PING_SUCCESS
+        result = driver.ping("8.8.8.8")
+        assert len(result["success"]["results"]) == 3
+        assert result["success"]["results"][0]["ip_address"] == "8.8.8.8"
+
+    def test_100_percent_loss(self, driver):
+        driver._send_command = lambda cmd, **kw: PING_LOSS
+        result = driver.ping("10.0.0.99")
+        assert "success" in result
+        assert result["success"]["packet_loss"] == 3
+
+    def test_error_on_bad_host(self, driver):
+        driver._send_command = lambda cmd, **kw: PING_ERROR
+        result = driver.ping("invalid.host")
+        assert "error" in result
+
+
+class TestGetIpv6NeighborsTable:
+    def test_returns_list(self, driver):
+        driver._send_command = lambda cmd, **kw: IPV6_NEIGH
+        assert isinstance(driver.get_ipv6_neighbors_table(), list)
+
+    def test_failed_entries_excluded(self, driver):
+        driver._send_command = lambda cmd, **kw: IPV6_NEIGH
+        result = driver.get_ipv6_neighbors_table()
+        assert len(result) == 2
+
+    def test_required_keys(self, driver):
+        driver._send_command = lambda cmd, **kw: IPV6_NEIGH
+        for entry in driver.get_ipv6_neighbors_table():
+            assert set(entry.keys()) >= {"interface", "mac", "ip", "age", "state"}
+
+    def test_state_values(self, driver):
+        driver._send_command = lambda cmd, **kw: IPV6_NEIGH
+        states = {e["state"] for e in driver.get_ipv6_neighbors_table()}
+        assert "REACHABLE" in states
+        assert "STALE" in states
+
+
+class TestGetRouteTo:
+    def test_returns_dict(self, driver):
+        driver._send_command = lambda cmd, **kw: IP_ROUTE_SHOW
+        assert isinstance(driver.get_route_to(), dict)
+
+    def test_default_route_present(self, driver):
+        driver._send_command = lambda cmd, **kw: IP_ROUTE_SHOW
+        result = driver.get_route_to()
+        assert "0.0.0.0/0" in result
+
+    def test_next_hop(self, driver):
+        driver._send_command = lambda cmd, **kw: IP_ROUTE_SHOW
+        result = driver.get_route_to()
+        default = result["0.0.0.0/0"][0]
+        assert default["next_hop"] == "192.168.1.1"
+        assert default["outgoing_interface"] == "br-wan"
+
+    def test_connected_route(self, driver):
+        driver._send_command = lambda cmd, **kw: IP_ROUTE_SHOW
+        result = driver.get_route_to()
+        assert "192.168.1.0/24" in result
+        assert result["192.168.1.0/24"][0]["protocol"] == "connected"
+
+    def test_protocol_filter(self, driver):
+        driver._send_command = lambda cmd, **kw: IP_ROUTE_SHOW
+        result = driver.get_route_to(protocol="static")
+        assert "10.0.0.0/8" in result
+        assert "192.168.1.0/24" not in result
+
+    def test_required_keys(self, driver):
+        driver._send_command = lambda cmd, **kw: IP_ROUTE_SHOW
+        for prefix, routes in driver.get_route_to().items():
+            for route in routes:
+                assert set(route.keys()) >= {
+                    "protocol", "current_active", "next_hop",
+                    "outgoing_interface", "preference", "routing_table",
+                }
+
+
+class TestTraceroute:
+    def test_success_result(self, driver):
+        driver._send_command = lambda cmd, **kw: TRACEROUTE_OUTPUT
+        result = driver.traceroute("8.8.8.8")
+        assert "success" in result
+
+    def test_hop_count(self, driver):
+        driver._send_command = lambda cmd, **kw: TRACEROUTE_OUTPUT
+        result = driver.traceroute("8.8.8.8")
+        assert len(result["success"]) == 4
+
+    def test_hop_1_rtt(self, driver):
+        driver._send_command = lambda cmd, **kw: TRACEROUTE_OUTPUT
+        result = driver.traceroute("8.8.8.8")
+        hop1 = result["success"][1]["probes"][1]
+        assert hop1["rtt"] == 1.123
+        assert hop1["ip_address"] == "192.168.1.1"
+
+    def test_star_hop(self, driver):
+        driver._send_command = lambda cmd, **kw: TRACEROUTE_OUTPUT
+        result = driver.traceroute("8.8.8.8")
+        hop3 = result["success"][3]["probes"][1]
+        assert hop3["ip_address"] == "*"
+
+    def test_error_on_unknown_host(self, driver):
+        driver._send_command = lambda cmd, **kw: TRACEROUTE_ERROR
+        result = driver.traceroute("invalid.host")
+        assert "error" in result
+
+
+class TestGetNetworkInstances:
+    def test_default_instance_always_present(self, driver):
+        driver._send_command = lambda cmd, **kw: (
+            IP_LINK_SHOW if "ip link" in cmd else ""
+        )
+        result = driver.get_network_instances()
+        assert "default" in result
+
+    def test_default_instance_type(self, driver):
+        driver._send_command = lambda cmd, **kw: (
+            IP_LINK_SHOW if "ip link" in cmd else ""
+        )
+        result = driver.get_network_instances()
+        assert result["default"]["type"] == "DEFAULT_INSTANCE"
+
+    def test_default_interfaces_populated(self, driver):
+        driver._send_command = lambda cmd, **kw: (
+            IP_LINK_SHOW if "ip link" in cmd else ""
+        )
+        result = driver.get_network_instances()
+        ifaces = result["default"]["interfaces"]["interface"]
+        assert "eth0" in ifaces
+
+    def test_named_namespaces(self, driver):
+        def _send(cmd, **kw):
+            if "netns list" in cmd:
+                return IP_NETNS_LIST
+            if "netns exec" in cmd:
+                return ""  # empty namespace
+            if "ip link" in cmd:
+                return IP_LINK_SHOW
+            return ""
+        driver._send_command = _send
+        result = driver.get_network_instances()
+        assert "vpn" in result
+        assert "mgmt" in result
+        assert result["vpn"]["type"] == "L3VRF"
+
+    def test_name_filter(self, driver):
+        driver._send_command = lambda cmd, **kw: (
+            IP_LINK_SHOW if "ip link" in cmd else ""
+        )
+        result = driver.get_network_instances(name="default")
+        assert list(result.keys()) == ["default"]
+
+    def test_required_keys(self, driver):
+        driver._send_command = lambda cmd, **kw: (
+            IP_LINK_SHOW if "ip link" in cmd else ""
+        )
+        for inst in driver.get_network_instances().values():
+            assert set(inst.keys()) >= {"name", "type", "state", "interfaces"}