Merge pull request 'chore: adopt the NAPALM driver standard' (#20) from chore/driver-standard into main
driver / standard (push) Successful in 14s
driver / lint (push) Successful in 11s
driver / types (push) Successful in 42s
driver / test (3.10) (push) Successful in 45s
driver / test (3.11) (push) Successful in 44s
driver / test (3.12) (push) Successful in 44s
driver / test (3.13) (push) Successful in 41s
driver / lowest (push) Successful in 23s
driver / security (push) Successful in 46s
driver / build (push) Successful in 23s
driver / publish (push) Successful in 26s
CI / driver (push) Successful in 1h1m4s

This commit was merged in pull request #20.
This commit is contained in:
christianmanivong committed 2026-10-23 09:40:30 +00:00
commit 7ca6827421
10 files changed
+129 -86

No files matched your search

+15
View File
@@ -0,0 +1,15 @@
name: CI
on:
push:
branches: [main]
tags: ["v*"]
pull_request:
jobs:
driver:
uses: NAPALM/driver-standard/.gitea/workflows/driver.yml@main
with:
standard-ref: main
secrets:
PYPI_TOKEN: ${{ secrets.PYPI_TOKEN }}
-48
View File
@@ -1,48 +0,0 @@
name: CI
on:
push:
branches: ["**"]
pull_request:
branches: ["**"]
jobs:
test:
runs-on: ubuntu-latest
strategy:
fail-fast: false
matrix:
python-version: ["3.10", "3.11", "3.12"]
steps:
- name: Checkout
uses: actions/checkout@v4
- name: Setup Python
uses: actions/setup-python@v5
with:
python-version: ${{ matrix.python-version }}
cache: pip
- name: Install package with dev extras
run: |
python -m pip install --upgrade pip
# napalm-device-types lives in git.netork.io/NAPALM, not on PyPI: without this
# pip looks there, finds an unrelated 0.1.0 and the job dies before any test.
python -m pip install "napalm-device-types @ git+https://git.netork.io/NAPALM/napalm-device-types.git"
python -m pip install -e ".[dev]"
- name: Run unit tests
run: |
python -m pytest -q --tb=short
- name: Build wheel and sdist
run: |
python -m pip install build
python -m build
- name: Upload dist artifacts
# v4 refuses to run on Gitea ("not currently supported on GHES").
uses: actions/upload-artifact@v3
with:
name: dist-${{ matrix.python-version }}
path: dist/*
+11
View File
@@ -7,6 +7,17 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
## [Unreleased] ## [Unreleased]
## [0.5.0] - 2026-10-23
The first release on PyPI.
### Changed
- The drivers follow the NAPALM driver standard
(https://git.netork.io/NAPALM/driver-standard): CI on Gitea runs ruff, mypy, the
tests on Python 3.10–3.13 and at the lowest declared versions, bandit, pip-audit and
a build check, and a release tag publishes to PyPI. The package now ships `py.typed`.
Python 3.10 stays the floor. napalm-device-types now comes from PyPI, 4.21.1 or later.
### Security ### Security
- The SSH host key is verified (NetOrk/netork#1167). `open()` takes the key to expect from `optional_args["ssh_host_key"]` and refuses any other one before it logs in (`SshHostKeyMismatch`); without one it accepts the first key. Every key it sees goes to `optional_args["ssh_host_key_observer"]`. Replaces paramiko's `AutoAddPolicy`. Requires napalm-device-types 4.19.0; version 0.4.0. - The SSH host key is verified (NetOrk/netork#1167). `open()` takes the key to expect from `optional_args["ssh_host_key"]` and refuses any other one before it logs in (`SshHostKeyMismatch`); without one it accepts the first key. Every key it sees goes to `optional_args["ssh_host_key_observer"]`. Replaces paramiko's `AutoAddPolicy`. Requires napalm-device-types 4.19.0; version 0.4.0.
-1
View File
@@ -93,7 +93,6 @@ machine, also in `hrProcessorLoad`).
## Development ## Development
```bash ```bash
pip install "napalm-device-types @ git+https://git.netork.io/NAPALM/napalm-device-types.git"
pip install -e ".[dev]" pip install -e ".[dev]"
pytest pytest
``` ```
+26 -18
View File
@@ -11,22 +11,21 @@ from __future__ import annotations
import ipaddress import ipaddress
import logging import logging
import re import re
import socket
from shlex import quote from shlex import quote
from typing import Any, Optional from typing import Any
import paramiko import paramiko
from napalm.base.exceptions import ConnectionClosedException, ConnectionException from napalm.base.exceptions import ConnectionClosedException, ConnectionException
from napalm_device_types import EnsurePackageMixin, OSDriver, SnmpAgentMixin from napalm_device_types import EnsurePackageMixin, OSDriver, SnmpAgentMixin
from napalm_device_types.models import FixStepOutcomeDict, SnmpAgentSettingsDict
from napalm_device_types.host_status import HostStatusMixin
from napalm_device_types.ssh_host_keys import HostKeyCheck, SshHostKeyMismatch
from napalm_device_types.channel import ( from napalm_device_types.channel import (
ByteStream, ByteStream,
CommandResult, CommandResult,
open_stream_on_transport, open_stream_on_transport,
run_on_transport, run_on_transport,
) )
from napalm_device_types.host_status import HostStatusMixin
from napalm_device_types.models import FixStepOutcomeDict, SnmpAgentSettingsDict
from napalm_device_types.ssh_host_keys import HostKeyCheck, SshHostKeyMismatch
from napalm_bsd import parse from napalm_bsd import parse
@@ -56,13 +55,15 @@ _PRIVILEGED_STDIN = (
#: Stages content without privilege: a fresh ``mktemp`` file only the login user #: Stages content without privilege: a fresh ``mktemp`` file only the login user
#: may read, never a fixed name another user could create first. Prints its path. #: may read, never a fixed name another user could create first. Prints its path.
_STAGE_TEMPLATE = "/tmp/napalm-bsd-stage.XXXXXX" #: The path is on the device, not a local temp file.
_STAGE_TEMPLATE = "/tmp/napalm-bsd-stage.XXXXXX" # nosec B108
_STAGE_PREFIX = _STAGE_TEMPLATE[: -len("XXXXXX")] _STAGE_PREFIX = _STAGE_TEMPLATE[: -len("XXXXXX")]
_STAGE = ( _STAGE = (
f"umask 077 && t=$(mktemp {_STAGE_TEMPLATE}) && " f"umask 077 && t=$(mktemp {_STAGE_TEMPLATE}) && "
'{ cat > "$t" && printf %s "$t" || { s=$?; rm -f -- "$t"; exit $s; }; }' '{ cat > "$t" && printf %s "$t" || { s=$?; rm -f -- "$t"; exit $s; }; }'
) )
def _source(value: str) -> str: def _source(value: str) -> str:
"""An allowed source as an address or network, or ValueError.""" """An allowed source as an address or network, or ValueError."""
return str(ipaddress.ip_network(value.strip(), strict=False)) return str(ipaddress.ip_network(value.strip(), strict=False))
@@ -113,7 +114,7 @@ class BsdDriver(SnmpAgentMixin, EnsurePackageMixin, HostStatusMixin, OSDriver):
username: str, username: str,
password: str, password: str,
timeout: int = 60, timeout: int = 60,
optional_args: Optional[dict] = None, optional_args: dict | None = None,
) -> None: ) -> None:
self.hostname = hostname self.hostname = hostname
self.username = username self.username = username
@@ -121,14 +122,14 @@ class BsdDriver(SnmpAgentMixin, EnsurePackageMixin, HostStatusMixin, OSDriver):
self.timeout = timeout self.timeout = timeout
args = optional_args or {} args = optional_args or {}
self.port: int = int(args.get("port", 22)) self.port: int = int(args.get("port", 22))
self._key_file: Optional[str] = args.get("key_file") self._key_file: str | None = args.get("key_file")
self._sudo_password: Optional[str] = args.get("sudo_password") self._sudo_password: str | None = args.get("sudo_password")
self._allow_agent: bool = bool(args.get("allow_agent", False)) self._allow_agent: bool = bool(args.get("allow_agent", False))
self._look_for_keys: bool = bool(args.get("look_for_keys", False)) self._look_for_keys: bool = bool(args.get("look_for_keys", False))
# The host key to expect, and where the one seen is reported (netork#1167). # The host key to expect, and where the one seen is reported (netork#1167).
self._host_key = HostKeyCheck(args, hostname, self.port) self._host_key = HostKeyCheck(args, hostname, self.port)
self._client: Optional[paramiko.SSHClient] = None self._client: paramiko.SSHClient | None = None
self._root: Optional[bool] = None self._root: bool | None = None
# -- connection -------------------------------------------------------------- # -- connection --------------------------------------------------------------
@@ -147,7 +148,7 @@ class BsdDriver(SnmpAgentMixin, EnsurePackageMixin, HostStatusMixin, OSDriver):
except SshHostKeyMismatch: except SshHostKeyMismatch:
client.close() # refused before any login; nothing was sent client.close() # refused before any login; nothing was sent
raise raise
except (paramiko.SSHException, socket.error) as exc: except (OSError, paramiko.SSHException) as exc:
client.close() client.close()
raise ConnectionException(f"SSH to {self.hostname} failed: {exc}") from exc raise ConnectionException(f"SSH to {self.hostname} failed: {exc}") from exc
self._client = client self._client = client
@@ -175,7 +176,7 @@ class BsdDriver(SnmpAgentMixin, EnsurePackageMixin, HostStatusMixin, OSDriver):
self._root = run_on_transport(self._transport(), "id -u").stdout.strip() == "0" self._root = run_on_transport(self._transport(), "id -u").stdout.strip() == "0"
return self._root return self._root
def _privileged(self, command: str, privileged: bool) -> tuple[str, Optional[bytes]]: def _privileged(self, command: str, privileged: bool) -> tuple[str, bytes | None]:
"""The command line that runs *command* as asked, and what goes to stdin first. """The command line that runs *command* as asked, and what goes to stdin first.
With a sudo password, ``sudo -S`` reads it from stdin, so it never With a sudo password, ``sudo -S`` reads it from stdin, so it never
@@ -194,7 +195,7 @@ class BsdDriver(SnmpAgentMixin, EnsurePackageMixin, HostStatusMixin, OSDriver):
*, *,
privileged: bool = False, privileged: bool = False,
timeout: float = 60, timeout: float = 60,
stdin: Optional[bytes] = None, stdin: bytes | None = None,
) -> CommandResult: ) -> CommandResult:
"""Run *command* on an exec channel: no PTY, stderr apart, a real exit code. """Run *command* on an exec channel: no PTY, stderr apart, a real exit code.
@@ -217,7 +218,11 @@ class BsdDriver(SnmpAgentMixin, EnsurePackageMixin, HostStatusMixin, OSDriver):
staged = self.run_command(_STAGE, timeout=30, stdin=content.encode()) staged = self.run_command(_STAGE, timeout=30, stdin=content.encode())
lines = staged.stdout.strip().splitlines() lines = staged.stdout.strip().splitlines()
tmp = lines[-1] if lines else "" tmp = lines[-1] if lines else ""
if staged.exit_code != 0 or not tmp.startswith(_STAGE_PREFIX) or "/" in tmp[len(_STAGE_PREFIX):]: if (
staged.exit_code != 0
or not tmp.startswith(_STAGE_PREFIX)
or "/" in tmp[len(_STAGE_PREFIX) :]
):
reason = (staged.stderr or staged.stdout).strip() or f"could not stage {path}" reason = (staged.stderr or staged.stdout).strip() or f"could not stage {path}"
return CommandResult(reason, "", staged.exit_code or 1) return CommandResult(reason, "", staged.exit_code or 1)
tmp_q, part_q, path_q = (quote(p) for p in (tmp, f"{path}.napalm-tmp", path)) tmp_q, part_q, path_q = (quote(p) for p in (tmp, f"{path}.napalm-tmp", path))
@@ -474,7 +479,7 @@ class BsdDriver(SnmpAgentMixin, EnsurePackageMixin, HostStatusMixin, OSDriver):
#: net-snmp's daemon; OpenBSD's base snmpd is /usr/sbin/snmpd. #: net-snmp's daemon; OpenBSD's base snmpd is /usr/sbin/snmpd.
SNMPD_DAEMON = "/usr/local/sbin/snmpd" SNMPD_DAEMON = "/usr/local/sbin/snmpd"
def get_management_peer(self) -> Optional[str]: def get_management_peer(self) -> str | None:
"""The address this host sees the SSH connection come from (``SSH_CONNECTION``).""" """The address this host sees the SSH connection come from (``SSH_CONNECTION``)."""
out = self.run_command(_PEER_COMMAND, timeout=15).stdout.split() out = self.run_command(_PEER_COMMAND, timeout=15).stdout.split()
try: try:
@@ -488,7 +493,10 @@ class BsdDriver(SnmpAgentMixin, EnsurePackageMixin, HostStatusMixin, OSDriver):
def _step(self, command: str, done: str) -> FixStepOutcomeDict: def _step(self, command: str, done: str) -> FixStepOutcomeDict:
result = self.run_command(command, privileged=True, timeout=600) result = self.run_command(command, privileged=True, timeout=600)
output = "\n".join(filter(None, (result.stdout.strip(), result.stderr.strip()))) output = "\n".join(filter(None, (result.stdout.strip(), result.stderr.strip())))
return {"ok": result.exit_code == 0, "output": done if result.exit_code == 0 else output[-300:]} return {
"ok": result.exit_code == 0,
"output": done if result.exit_code == 0 else output[-300:],
}
def _install_snmp_agent(self) -> FixStepOutcomeDict: def _install_snmp_agent(self) -> FixStepOutcomeDict:
return self._step(self.INSTALL_COMMAND.format(name="net-snmp"), "net-snmp installed") return self._step(self.INSTALL_COMMAND.format(name="net-snmp"), "net-snmp installed")
@@ -507,7 +515,7 @@ class BsdDriver(SnmpAgentMixin, EnsurePackageMixin, HostStatusMixin, OSDriver):
def _start_snmp_agent(self) -> FixStepOutcomeDict: def _start_snmp_agent(self) -> FixStepOutcomeDict:
return self._step(self.SNMPD_START, "snmpd enabled and running") return self._step(self.SNMPD_START, "snmpd enabled and running")
def get_snmp_config(self) -> Optional[dict[str, Any]]: def get_snmp_config(self) -> dict[str, Any] | None:
"""net-snmp's community and port, if its daemon runs; None otherwise.""" """net-snmp's community and port, if its daemon runs; None otherwise."""
if not self._out(f"pgrep -f {self.SNMPD_DAEMON}"): if not self._out(f"pgrep -f {self.SNMPD_DAEMON}"):
return None return None
+15 -10
View File
@@ -9,7 +9,7 @@ from __future__ import annotations
import ipaddress import ipaddress
import re import re
from typing import Any, Optional from typing import Any
_IFACE_HEADER = re.compile( _IFACE_HEADER = re.compile(
r"^(?P<name>[^\s:]+): flags=[0-9a-f]+<(?P<flags>[^>]*)>(?:.*?\bmtu (?P<mtu>\d+))?" r"^(?P<name>[^\s:]+): flags=[0-9a-f]+<(?P<flags>[^>]*)>(?:.*?\bmtu (?P<mtu>\d+))?"
@@ -26,7 +26,7 @@ def _prefix_from_netmask(mask: str) -> int:
def ifconfig(text: str) -> dict[str, dict[str, Any]]: def ifconfig(text: str) -> dict[str, dict[str, Any]]:
"""``ifconfig -a`` -> per interface: flags, MTU, MAC, state, speed, addresses.""" """``ifconfig -a`` -> per interface: flags, MTU, MAC, state, speed, addresses."""
interfaces: dict[str, dict[str, Any]] = {} interfaces: dict[str, dict[str, Any]] = {}
current: Optional[dict[str, Any]] = None current: dict[str, Any] | None = None
for line in text.splitlines(): for line in text.splitlines():
header = _IFACE_HEADER.match(line) header = _IFACE_HEADER.match(line)
if header: if header:
@@ -69,7 +69,7 @@ def ifconfig(text: str) -> dict[str, dict[str, Any]]:
return interfaces return interfaces
def _network(destination: str, family: str) -> Optional[str]: def _network(destination: str, family: str) -> str | None:
"""A netstat destination as a network: ``default``, ``10.0.2/24``, ``fe80::%em0/64``.""" """A netstat destination as a network: ``default``, ``10.0.2/24``, ``fe80::%em0/64``."""
if destination == "default": if destination == "default":
return "0.0.0.0/0" if family == "ipv4" else "::/0" return "0.0.0.0/0" if family == "ipv4" else "::/0"
@@ -93,7 +93,7 @@ def routes(text: str) -> list[dict[str, str]]:
""" """
result: list[dict[str, str]] = [] result: list[dict[str, str]] = []
family = "" family = ""
iface_column: Optional[int] = None iface_column: int | None = None
for line in text.splitlines(): for line in text.splitlines():
words = line.split() words = line.split()
if not words: if not words:
@@ -253,15 +253,20 @@ def crontab(text: str, *, system: bool = False, user: str = "") -> list[dict[str
return jobs return jobs
def boottime(text: str) -> Optional[int]: def boottime(text: str) -> int | None:
"""``sysctl -n kern.boottime``: FreeBSD ``{ sec = N, … }``, OpenBSD ``N``.""" """``sysctl -n kern.boottime``: FreeBSD ``{ sec = N, … }``, OpenBSD ``N``."""
match = re.search(r"sec = (\d+)", text) or re.fullmatch(r"\s*(\d+)\s*", text) match = re.search(r"sec = (\d+)", text) or re.fullmatch(r"\s*(\d+)\s*", text)
return int(match[1]) if match else None return int(match[1]) if match else None
#: The wildcard addresses a socket listens on: reported as data, never bound to.
_ANY_V4 = "0.0.0.0" # nosec B104
_ANY_V6 = "::"
def _socket( def _socket(
proto: str, local: str, family: str, process: Optional[str], pid: Optional[int] proto: str, local: str, family: str, process: str | None, pid: int | None
) -> Optional[dict[str, Any]]: ) -> dict[str, Any] | None:
"""A ListeningSocketDict from a local endpoint: ``*:22``, ``[::1]:25``, """A ListeningSocketDict from a local endpoint: ``*:22``, ``[::1]:25``,
``127.0.0.1:25``, ``[fe80::1%lo0]:25`` (sockstat, fstat) or ``*.22``, ``127.0.0.1:25``, ``[fe80::1%lo0]:25`` (sockstat, fstat) or ``*.22``,
``::1.25``, ``fe80::1%lo0.25`` (netstat).""" ``::1.25``, ``fe80::1%lo0.25`` (netstat)."""
@@ -275,7 +280,7 @@ def _socket(
return None return None
address, _, zone = address.partition("%") address, _, zone = address.partition("%")
if address == "*": if address == "*":
address = "0.0.0.0" if family == "4" else "::" address = _ANY_V4 if family == "4" else _ANY_V6
return { return {
"proto": proto, "proto": proto,
"address": address, "address": address,
@@ -288,9 +293,9 @@ def _socket(
} }
def _unique(sockets: list[Optional[dict[str, Any]]]) -> list[dict[str, Any]]: def _unique(sockets: list[dict[str, Any] | None]) -> list[dict[str, Any]]:
"""One entry per socket, the first process holding it, as ss reports it.""" """One entry per socket, the first process holding it, as ss reports it."""
seen: set[tuple] = set() seen: set[tuple[Any, ...]] = set()
result = [] result = []
for s in sockets: for s in sockets:
if s is None: if s is None:
View File
Whitespace-only changes.
+43 -6
View File
@@ -1,14 +1,14 @@
[build-system] [build-system]
requires = ["setuptools>=68", "wheel"] requires = ["setuptools>=77"]
build-backend = "setuptools.build_meta" build-backend = "setuptools.build_meta"
[project] [project]
name = "napalm-bsd" name = "napalm-bsd"
version = "0.4.0" version = "0.5.0"
description = "NAPALM drivers for FreeBSD and OpenBSD hosts via SSH" description = "NAPALM drivers for FreeBSD and OpenBSD hosts via SSH"
readme = "README.md" readme = "README.md"
requires-python = ">=3.10" requires-python = ">=3.10"
license = { text = "Apache-2.0" } license = "Apache-2.0"
authors = [ authors = [
{ name = "Christian Manivong", email = "christian@manivong.de" }, { name = "Christian Manivong", email = "christian@manivong.de" },
] ]
@@ -16,22 +16,23 @@ keywords = ["napalm", "network", "automation", "freebsd", "openbsd", "bsd", "ssh
classifiers = [ classifiers = [
"Development Status :: 3 - Alpha", "Development Status :: 3 - Alpha",
"Intended Audience :: System Administrators", "Intended Audience :: System Administrators",
"License :: OSI Approved :: Apache Software License",
"Operating System :: OS Independent", "Operating System :: OS Independent",
"Programming Language :: Python :: 3", "Programming Language :: Python :: 3",
"Programming Language :: Python :: 3.10", "Programming Language :: Python :: 3.10",
"Programming Language :: Python :: 3.11", "Programming Language :: Python :: 3.11",
"Programming Language :: Python :: 3.12", "Programming Language :: Python :: 3.12",
"Programming Language :: Python :: 3.13",
"Topic :: System :: Systems Administration", "Topic :: System :: Systems Administration",
"Typing :: Typed",
] ]
dependencies = [ dependencies = [
"napalm>=4.0", "napalm>=4.0",
"napalm-device-types>=4.19.0", # fixes: SnmpAgentMixin, EnsurePackageMixin "napalm-device-types>=4.21.1",
"paramiko>=5.0.0", # CVE-2026-44405 "paramiko>=5.0.0", # CVE-2026-44405
] ]
[project.optional-dependencies] [project.optional-dependencies]
dev = ["pytest", "pytest-cov", "ruff", "mypy"] dev = ["pytest>=8", "pytest-cov>=5", "mypy>=1.11", "ruff>=0.6"]
[project.entry-points."napalm.drivers"] [project.entry-points."napalm.drivers"]
freebsd = "napalm_bsd:FreeBSDDriver" freebsd = "napalm_bsd:FreeBSDDriver"
@@ -39,11 +40,47 @@ openbsd = "napalm_bsd:OpenBSDDriver"
[project.urls] [project.urls]
Repository = "https://git.netork.io/NAPALM/napalm-bsd" Repository = "https://git.netork.io/NAPALM/napalm-bsd"
Issues = "https://git.netork.io/NAPALM/napalm-bsd/issues"
Changelog = "https://git.netork.io/NAPALM/napalm-bsd/src/branch/main/CHANGELOG.md"
[tool.setuptools.packages.find] [tool.setuptools.packages.find]
where = ["."] where = ["."]
include = ["napalm_bsd*"] include = ["napalm_bsd*"]
[tool.setuptools.package-data]
napalm_bsd = ["py.typed"]
[tool.ruff] [tool.ruff]
line-length = 100 line-length = 100
target-version = "py310" target-version = "py310"
[tool.ruff.lint]
select = ["E", "F", "I", "UP", "B"]
[tool.ruff.lint.per-file-ignores]
# Recorded device output keeps its lines.
"tests/**" = ["E501"]
[tool.mypy]
python_version = "3.10"
strict = true
# napalm itself is untyped, so every driver subclasses Any.
disallow_subclassing_any = false
[[tool.mypy.overrides]]
module = ["napalm.*", "paramiko.*"]
ignore_missing_imports = true
[[tool.mypy.overrides]]
# Not yet under strict mypy (quality baseline); may only shrink.
module = ["napalm_bsd.base", "napalm_bsd.freebsd", "napalm_bsd.openbsd"]
ignore_errors = true
[tool.pytest.ini_options]
testpaths = ["tests"]
[tool.coverage.run]
source = ["napalm_bsd"]
[tool.coverage.report]
fail_under = 94
+16
View File
@@ -0,0 +1,16 @@
{
"coverage_floor": 94.0,
"function_complexity": {
"napalm_bsd/parse.py::ifconfig": 21,
"napalm_bsd/parse.py::routes": 21,
"napalm_bsd/parse.py::users": 11
},
"function_lines": {},
"mypy_relaxed": [
"napalm_bsd.base",
"napalm_bsd.freebsd",
"napalm_bsd.openbsd"
],
"python_file_lines": {},
"untyped_public": []
}
+3 -3
View File
@@ -8,8 +8,7 @@ import pytest
from napalm.base import NetworkDriver from napalm.base import NetworkDriver
from napalm_device_types import OSDriver from napalm_device_types import OSDriver
from napalm_bsd import FreeBSDDriver, OpenBSDDriver from napalm_bsd import FreeBSDDriver, OpenBSDDriver, base
from napalm_bsd import base
from napalm_bsd.base import BsdDriver from napalm_bsd.base import BsdDriver
FIXTURES = Path(__file__).parent / "fixtures" FIXTURES = Path(__file__).parent / "fixtures"
@@ -600,7 +599,8 @@ class TestPrivilegedStdin:
def _driver(self, monkeypatch, *, sudo_password, root=False): def _driver(self, monkeypatch, *, sudo_password, root=False):
sent = [] sent = []
monkeypatch.setattr( monkeypatch.setattr(
base, "run_on_transport", base,
"run_on_transport",
lambda transport, line, stdin=None, timeout=60: sent.append((line, stdin)) or _Result(), lambda transport, line, stdin=None, timeout=60: sent.append((line, stdin)) or _Result(),
) )
args = {"sudo_password": sudo_password} if sudo_password else {} args = {"sudo_password": sudo_password} if sudo_password else {}