feat: a public command channel; no warnings of its own (0.4.0)

run_command and open_stream (napalm-device-types CommandChannelMixin,
which AccessPointDriver carries) run on an SSH exec channel next to
netmiko's shell: stdin, stderr apart, a real exit code. netOrk sent UCI
scripts and the wifi channel through the private _send_command, typing
a file in base64 chunks or printf lines; with stdin it writes it with
cat, and base64 is needed for nothing (netOrk #1113). OpenWrt has no
sudo, so a privileged command from a login other than root is refused
before it runs.

get_device_warnings goes with it: no_base64 was its last code.

napalm_openwrt/openwrt.py

@@ -29,10 +29,13 @@ from netmiko.exceptions import NetmikoTimeoutException, NetmikoAuthenticationExc
 
 from napalm_device_types import (
     AccessPointDriver,
+    ByteStream,
+    CommandResult,
     EnsurePackageMixin,
     FingerprintRule,
     SnmpAgentMixin,
 )
+from napalm_device_types.channel import open_stream_on_transport, run_on_transport
 from napalm_device_types.listening import ListeningSocketsMixin
 from napalm.base.exceptions import (
     ConnectionException,
@@ -183,6 +186,43 @@ class OpenWrtDriver(
         except (socket.error, EOFError) as exc:
             raise ConnectionClosedException(str(exc)) from exc
 
+    # ------------------------------------------------------------------
+    # Command channel (napalm-device-types CommandChannelMixin, via AccessPointDriver)
+    # ------------------------------------------------------------------
+
+    def _transport(self) -> Any:
+        """The SSH transport netmiko already holds, for exec channels next to its shell."""
+        if self.device is None:
+            raise ConnectionClosedException("Not connected")
+        return self.device.remote_conn_pre.get_transport()
+
+    def _check_privilege(self, privileged: bool) -> None:
+        if privileged and self.username != "root":
+            raise PermissionError(
+                f"{self.hostname}: OpenWrt has no sudo; {self.username!r} cannot run as root"
+            )
+
+    def run_command(
+        self,
+        command: str,
+        *,
+        privileged: bool = False,
+        timeout: float = 60,
+        stdin: bytes | None = None,
+    ) -> CommandResult:
+        """Run *command* on an SSH exec channel: no PTY, stderr apart, a real exit code.
+
+        The login is root on OpenWrt as a rule, and there is no sudo: a
+        privileged command from any other login is refused before it runs.
+        """
+        self._check_privilege(privileged)
+        return run_on_transport(self._transport(), command, stdin=stdin, timeout=timeout)
+
+    def open_stream(self, command: str, *, privileged: bool = False) -> ByteStream:
+        """Start *command* on an SSH exec channel and return a stream to it."""
+        self._check_privilege(privileged)
+        return open_stream_on_transport(self._transport(), command)
+
     def _run_listening_sockets_command(self, command: str, *, privileged: bool) -> str:
         """Run the listening-socket command for ``ListeningSocketsMixin``.
 

napalm_openwrt/system_mixin.py

@@ -390,19 +390,6 @@ class OpenWrtSystemMixin:
         output = self._send_command(f"/etc/init.d/{name} {action} 2>&1")
         return {"success": True, "output": output}
 
-    def get_device_warnings(self) -> list[dict[str, Any]]:
-        """Only a missing ``base64`` binary, which BusyBox may lack.
-
-        It is not a package the caller can look up -- BusyBox may provide it as
-        an applet -- so it stays here until the shell commands callers send get
-        a public method. lldpd, auc and the pending updates are facts
-        (``get_packages``, ``get_available_updates``); whether they are a
-        problem is the caller's to decide.
-        """
-        if self._send_command("command -v base64 2>/dev/null").strip():
-            return []
-        return [{"code": "no_base64"}]
-
     def get_available_updates(self) -> list[dict[str, Any]]:
         """Return list of upgradable packages from the local package manager cache.
 

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "napalm-openwrt"
-version = "0.3.0"
+version = "0.4.0"
 description = "NAPALM driver for OpenWrt routers/access-points"
 readme = "README.md"
 license = { text = "Apache-2.0" }

tests/unit/test_command_channel.py

@@ -0,0 +1,79 @@
+"""A public command channel (napalm-device-types ``CommandChannelMixin``, netOrk #1113).
+
+netOrk sent UCI scripts and the wifi channel through the private
+``_send_command``: an interactive shell with no exit code, where a file had to
+be typed in base64 chunks or ``printf`` lines. ``run_command`` runs on an SSH
+exec channel of its own, with stdin, stderr and the exit code -- a script is
+written with ``cat`` from stdin, and ``base64`` is needed for nothing.
+"""
+
+from __future__ import annotations
+
+from unittest.mock import MagicMock, patch
+
+import pytest
+from napalm.base.exceptions import ConnectionClosedException
+from napalm_device_types import CommandResult
+
+from napalm_openwrt.openwrt import OpenWrtDriver
+
+
+def _driver(username: str = "root") -> OpenWrtDriver:
+    with patch("napalm_openwrt.openwrt.ConnectHandler"):
+        drv = OpenWrtDriver(hostname="192.0.2.15", username=username, password="")
+    drv.device = MagicMock()
+    return drv
+
+
+def test_a_command_runs_on_the_ssh_transport_with_its_stdin():
+    drv = _driver()
+    transport = drv.device.remote_conn_pre.get_transport.return_value
+    done = CommandResult("", "", 0)
+    with patch("napalm_openwrt.openwrt.run_on_transport", return_value=done) as run:
+        result = drv.run_command("cat > /tmp/x.sh", stdin=b"uci commit\n", timeout=30)
+
+    assert result is done
+    run.assert_called_once_with(transport, "cat > /tmp/x.sh", stdin=b"uci commit\n", timeout=30)
+
+
+def test_root_runs_privileged_commands_as_they_are():
+    drv = _driver()
+    with patch("napalm_openwrt.openwrt.run_on_transport") as run:
+        drv.run_command("wifi", privileged=True)
+
+    assert run.call_args.args[1] == "wifi"
+
+
+def test_a_login_other_than_root_cannot_gain_root():
+    """OpenWrt has no sudo: refused before anything runs, not half done."""
+    drv = _driver(username="admin")
+    with patch("napalm_openwrt.openwrt.run_on_transport") as run, pytest.raises(PermissionError):
+        drv.run_command("wifi", privileged=True)
+
+    run.assert_not_called()
+
+
+def test_a_stream_on_the_same_transport():
+    drv = _driver()
+    transport = drv.device.remote_conn_pre.get_transport.return_value
+    with patch("napalm_openwrt.openwrt.open_stream_on_transport") as stream:
+        drv.open_stream("logread -f")
+
+    stream.assert_called_once_with(transport, "logread -f")
+
+
+def test_not_connected():
+    drv = _driver()
+    drv.device = None
+    with pytest.raises(ConnectionClosedException):
+        drv.run_command("true")
+
+
+def test_no_warnings_of_its_own():
+    """base64 is no longer needed for anything, so a missing one is no warning;
+    lldpd, auc and the updates are facts the caller reads (netOrk #1111)."""
+    assert not any(
+        "get_device_warnings" in vars(c)
+        for c in OpenWrtDriver.__mro__
+        if c.__module__.startswith("napalm_openwrt")
+    )

tests/unit/test_warnings.py

@@ -1,43 +0,0 @@
-"""What the driver still names itself (netOrk #1111).
-
-lldpd, auc and the pending updates are facts the caller reads (``get_packages``,
-``get_available_updates``) and decides on. A missing ``base64`` stays: BusyBox
-may provide it as an applet, so no package says whether it is there.
-"""
-
-from __future__ import annotations
-
-from napalm_openwrt import OpenWrtDriver
-
-
-def _driver(answers: dict[str, str]) -> OpenWrtDriver:
-    d = OpenWrtDriver("ap", "root", "pw")
-    issued: list[str] = []
-
-    def send(command: str) -> str:
-        issued.append(command)
-        return next((out for frag, out in answers.items() if frag in command), "")
-
-    d._send_command = send
-    d.issued = issued
-    return d
-
-
-def test_nothing_installed_names_only_base64():
-    """No lldpd, no auc and an opkg system: the caller names those from the packages."""
-    d = _driver({"opkg list-upgradable": "luci - 1.0 - 1.1\n"})
-
-    assert [w["code"] for w in d.get_device_warnings()] == ["no_base64"]
-
-
-def test_base64_present_names_nothing():
-    d = _driver({"command -v base64": "/usr/bin/base64\n"})
-
-    assert d.get_device_warnings() == []
-
-
-def test_one_command_per_poll():
-    d = _driver({})
-    d.get_device_warnings()
-
-    assert d.issued == ["command -v base64 2>/dev/null"]