Mechanical: ruff format, and ruff check --fix with the rule set every NAPALM driver uses (E, F, I, UP, B at line length 100). typing.List/Dict/Optional become builtin generics and X | None, imports are sorted, the coding cookies are gone. Two tests that read the source text follow the new spelling. What ruff cannot fix comes in the next commit.
185 lines
5.2 KiB
Python
185 lines
5.2 KiB
Python
"""The command channel: how a driver runs a command and opens a byte stream.
|
|
|
|
netOrk's container runtime driver speaks the Docker Engine API over a stream
|
|
from ``docker system dial-stdio`` (NetOrk/netork#765). The driver only provides
|
|
the way there; these are the SSH pieces every SSH driver can reuse. The
|
|
paramiko channel is faked, so the tests pin behaviour, not paramiko.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import pytest
|
|
|
|
from napalm_device_types.channel import (
|
|
CommandResult,
|
|
ParamikoExecStream,
|
|
open_stream_on_transport,
|
|
run_on_transport,
|
|
)
|
|
|
|
|
|
class FakeChannel:
|
|
"""Just enough of paramiko.Channel: scripted stdout/stderr chunks, an exit code."""
|
|
|
|
def __init__(self, stdout=(), stderr=(), exit_code=0, hang=False):
|
|
self.out = list(stdout)
|
|
self.err = list(stderr)
|
|
self.exit_code = exit_code
|
|
self.hang = hang
|
|
self.sent = b""
|
|
self.command = None
|
|
self.write_closed = False
|
|
self.closed = False
|
|
self.timeout = None
|
|
|
|
def exec_command(self, command):
|
|
self.command = command
|
|
|
|
def settimeout(self, timeout):
|
|
self.timeout = timeout
|
|
|
|
def sendall(self, data):
|
|
self.sent += data
|
|
|
|
def shutdown_write(self):
|
|
self.write_closed = True
|
|
|
|
def close(self):
|
|
self.closed = True
|
|
|
|
def recv_ready(self):
|
|
return bool(self.out)
|
|
|
|
def recv(self, n):
|
|
if self.out:
|
|
return self.out.pop(0)
|
|
if self.hang:
|
|
raise TimeoutError()
|
|
return b""
|
|
|
|
def recv_stderr_ready(self):
|
|
return bool(self.err)
|
|
|
|
def recv_stderr(self, n):
|
|
return self.err.pop(0) if self.err else b""
|
|
|
|
def exit_status_ready(self):
|
|
return not self.hang and not self.out and not self.err
|
|
|
|
def recv_exit_status(self):
|
|
return self.exit_code
|
|
|
|
|
|
class FakeTransport:
|
|
def __init__(self, channel):
|
|
self.channel = channel
|
|
|
|
def open_session(self):
|
|
return self.channel
|
|
|
|
|
|
def test_run_collects_stdout_stderr_and_the_exit_code():
|
|
ch = FakeChannel(stdout=[b"hel", b"lo\n"], stderr=[b"warn\n"], exit_code=3)
|
|
|
|
result = run_on_transport(FakeTransport(ch), "echo hello", timeout=5)
|
|
|
|
assert result == CommandResult(stdout="hello\n", stderr="warn\n", exit_code=3)
|
|
assert ch.command == "echo hello"
|
|
assert ch.closed
|
|
|
|
|
|
def test_run_sends_stdin_and_then_closes_the_write_side():
|
|
ch = FakeChannel(stdout=[b"ok"])
|
|
|
|
run_on_transport(FakeTransport(ch), "sudo -S true", stdin=b"pw\n", timeout=5)
|
|
|
|
assert ch.sent == b"pw\n"
|
|
assert ch.write_closed
|
|
|
|
|
|
def test_run_raises_when_the_command_does_not_finish_in_time():
|
|
ch = FakeChannel(hang=True)
|
|
|
|
with pytest.raises(TimeoutError, match="did not finish"):
|
|
run_on_transport(FakeTransport(ch), "sleep 999", timeout=0.05)
|
|
assert ch.closed
|
|
|
|
|
|
def test_a_stream_reads_until_eof_and_drains_stderr_on_the_way():
|
|
ch = FakeChannel(stdout=[b"HTTP/1.1 200 OK\r\n"], stderr=[b"note\n"])
|
|
stream = open_stream_on_transport(FakeTransport(ch), "docker system dial-stdio")
|
|
|
|
assert stream.read(4096, timeout=5) == b"HTTP/1.1 200 OK\r\n"
|
|
assert stream.read(4096, timeout=5) == b""
|
|
assert stream.stderr == "note\n"
|
|
assert ch.command == "docker system dial-stdio"
|
|
|
|
|
|
def test_a_stream_writes_and_closes_its_write_side_only_when_asked():
|
|
"""dial-stdio answers HTTP 499 to a request whose writer closed early
|
|
(netork#771), so write() never implies close_write()."""
|
|
ch = FakeChannel()
|
|
stream = open_stream_on_transport(FakeTransport(ch), "docker system dial-stdio")
|
|
|
|
stream.write(b"GET /_ping HTTP/1.1\r\n\r\n")
|
|
assert not ch.write_closed
|
|
|
|
stream.close_write()
|
|
assert ch.write_closed and ch.sent.startswith(b"GET /_ping")
|
|
|
|
|
|
def test_a_stream_sends_its_stdin_prefix_first():
|
|
"""How a sudo password reaches `sudo -S` before the stream's own bytes."""
|
|
ch = FakeChannel()
|
|
|
|
open_stream_on_transport(FakeTransport(ch), "sudo -S -p '' cmd", stdin_prefix=b"pw\n")
|
|
|
|
assert ch.sent == b"pw\n"
|
|
|
|
|
|
def test_a_stream_times_out_as_timeout_error():
|
|
stream = ParamikoExecStream(FakeChannel(hang=True))
|
|
|
|
with pytest.raises(TimeoutError):
|
|
stream.read(10, timeout=0.01)
|
|
|
|
|
|
def test_the_stream_keeps_only_a_bounded_tail_of_stderr():
|
|
ch = FakeChannel(stderr=[b"x" * 100, b"y" * 100])
|
|
stream = ParamikoExecStream(ch, stderr_limit=50)
|
|
|
|
assert stream.stderr == "y" * 50
|
|
|
|
|
|
def test_exit_status_is_none_while_running_and_the_code_after():
|
|
running = ParamikoExecStream(FakeChannel(hang=True))
|
|
finished = ParamikoExecStream(FakeChannel(exit_code=1))
|
|
|
|
assert running.exit_status is None
|
|
assert finished.exit_status == 1
|
|
|
|
|
|
def test_close_closes_the_channel():
|
|
ch = FakeChannel()
|
|
|
|
ParamikoExecStream(ch).close()
|
|
|
|
assert ch.closed
|
|
|
|
|
|
def test_run_keeps_output_that_arrives_after_the_exit_status():
|
|
"""sshd may send exit-status before the last stdout bytes."""
|
|
|
|
class EarlyExit(FakeChannel):
|
|
def exit_status_ready(self):
|
|
return True
|
|
|
|
def recv_ready(self):
|
|
return False
|
|
|
|
channel = EarlyExit(stdout=[b"head ", b"tail"], stderr=[b"warn"])
|
|
|
|
result = run_on_transport(FakeTransport(channel), "cmd")
|
|
|
|
assert result == CommandResult("head tail", "warn", 0)
|