nathan/nixpkgs
1
0
Fork 0
mirror of https://github.com/NixOS/nixpkgs.git synced 2026-10-30 16:10:36 +00:00
Code Issues Packages Projects Releases Wiki Activity

staging-nixos (#577672)

Browse Source
This commit is contained in:
Vladimír Čunát
2026-10-23 05:40:56 +00:00
committed by GitHub
parent 0a494f9b31 0939bf0bf2
commit 2cf4061217
24 changed files with 813 additions and 186 deletions
Download Patch File Download Diff File Expand all files Collapse all files

26
nixos/doc/manual/development/running-nixos-tests-interactively.section.md
View File

@@ -41,6 +41,32 @@ back into the test driver command line upon its completion. This allows
you to inspect the state of the VMs after the test (e.g. to debug the
test script).
## Graphical tests {#sec-nixos-test-interactive-graphical}
For `systemd-nspawn` containers with X11 enabled, `.driverInteractive` opens
TigerVNC once the X server is ready. It connects to the container's default
X11 display through an x11vnc Unix socket in the shared directory and accepts
keyboard and mouse input.
Closing the viewer leaves the container running.
Because the nspawn driver runs as root, preserve the host graphical-session
variables when starting it. For X11, use:
```ShellSession
$ sudo --preserve-env=DISPLAY,XAUTHORITY \
./result/bin/nixos-test-driver
```
For a Wayland host session, use:
```ShellSession
$ sudo --preserve-env=WAYLAND_DISPLAY,XDG_RUNTIME_DIR \
./result/bin/nixos-test-driver
```
If neither `DISPLAY` nor `WAYLAND_DISPLAY` is set, the driver skips the viewer.
Ordinary test runs do not start a VNC server or viewer.
## Shell access to VMs in interactive mode {#sec-nixos-test-shell-access}
::: {.warning}

6
nixos/doc/manual/development/writing-nixos-tests.section.md
View File

@@ -151,12 +151,16 @@ Some advantages of virtual machines over containers are:
- Virtual machines run a separate kernel, which allows testing kernel features
(kernel modules, etc.).
- Virtual machines support testing graphical applications on X11.
- Virtual machines allow testing NixOS modules that use systemd's namespacing options (such as `ProtectSystem=` or `MountAPIVFS=`).
- Virtual machines allow testing [`specialisation`](options.html#opt-specialisation).
(Switching to a specialisation requires the creation of SUID/SGID wrappers, which is disallowed in `systemd-nspawn` within the Nix sandbox.)
- Virtual machines allow the execution of `setuid` binaries.
Both backends support testing graphical applications on X11. Virtual machines
provide emulated display hardware, while containers use a headless X server.
See [running tests interactively](#sec-nixos-test-interactive-graphical) for how
to view container displays during debugging.
Refer to the sections on [QEMU virtual machines](#ssec-nixos-test-qemu-vms)
and [systemd-nspawn containers](#ssec-nixos-test-nspawn-containers) below
for more details on configuring each type of machine.

3
nixos/doc/manual/redirects.json
View File

@@ -2459,6 +2459,9 @@
"sec-running-nixos-tests-interactively": [
"index.html#sec-running-nixos-tests-interactively"
],
"sec-nixos-test-interactive-graphical": [
"index.html#sec-nixos-test-interactive-graphical"
],
"sec-nixos-test-shell-access": [
"index.html#sec-nixos-test-shell-access"
],

2
nixos/doc/manual/release-notes/rl-2611.section.md
View File

@@ -471,6 +471,8 @@
- The `shell_interact()` function on interactive runs of NixOS VM tests has been deprecated. Use the SSH backdoor instead.
- NixOS tests using `systemd-nspawn` containers now support graphical X11 applications, including keyboard input, screenshots, and OCR. The [interactive test driver](#sec-nixos-test-interactive-graphical) opens each container's default X11 display in TigerVNC for observation and input.
- The {option}`programs.fish.shellFunctions` option can now be used to create custom fish functions in a structured manner, as opposed to concatenating strings with {option}`program.fish.interactiveShellInit`.
:::{.example}
# Migrating fish functions to `programs.fish.shellFunctions`

10
nixos/lib/test-driver/default.nix
View File

@@ -71,8 +71,11 @@ buildPythonApplication {
tesseract4
];
# containers test requires extra nix features that are not available in ofborg.
passthru.tests = removeAttrs nixosTests.nixos-test-driver [ "containers" ];
# containers tests require extra nix features that are not available in ofborg.
passthru.tests = removeAttrs nixosTests.nixos-test-driver [
"containers"
"nspawn-interactive"
];
doCheck = true;
@@ -83,7 +86,8 @@ buildPythonApplication {
checkPhase = ''
echo -e "\x1b[32m## run ty\x1b[0m"
ty check --error-on-warning test_driver extract-docstrings.py
# Installation adds the driver to PYTHONPATH; keep ty imports in the source tree.
PYTHONPATH=".:$PYTHONPATH" ty check --error-on-warning test_driver extract-docstrings.py
echo -e "\x1b[32m## run ruff check\x1b[0m"
ruff check .
echo -e "\x1b[32m## run ruff format\x1b[0m"

1
nixos/lib/test-driver/src/test_driver/__init__.py
View File

@@ -178,6 +178,7 @@ def main() -> None:
logger=logger,
keep_machine_state=args.keep_machine_state,
debug=debugger,
interactive=bool(args.interactive),
) as driver:
if driver.config.enable_ssh_backdoor:
driver.dump_machine_ssh()

5
nixos/lib/test-driver/src/test_driver/driver.py
View File

@@ -30,6 +30,7 @@ from test_driver.machine import (
QemuMachine,
retry,
)
from test_driver.machine.nspawn_display import NspawnVncConfiguration
from test_driver.polling_condition import PollingCondition
from test_driver.vlan import VLan
@@ -37,6 +38,7 @@ from test_driver.vlan import VLan
class NodeConfiguration(BaseModel):
name: str
start_script: Path
vnc: NspawnVncConfiguration | None = None
class DriverConfiguration(BaseModel):
@@ -158,6 +160,7 @@ class Driver:
logger: AbstractLogger,
keep_machine_state: bool = False,
debug: DebugAbstract = DebugNop(),
interactive: bool = False,
):
self.config = config
self.tests = config.test_script.read_text()
@@ -166,6 +169,7 @@ class Driver:
self.debug = debug
self.polling_conditions = []
self.keep_machine_state = keep_machine_state
self.interactive = interactive
def __enter__(self) -> "Driver":
self.race_timer = threading.Timer(
@@ -214,6 +218,7 @@ class Driver:
NspawnMachine(
name=name,
start_command=container_config.start_script.as_posix(),
vnc=container_config.vnc if self.interactive else None,
tmp_dir=tmp_dir,
logger=self.logger,
keep_machine_state=self.keep_machine_state,

431
nixos/lib/test-driver/src/test_driver/machine/__init__.py
View File

@@ -32,6 +32,7 @@ from test_driver.duration import (
from test_driver.efi import EfiVariable, EfiVars
from test_driver.errors import MachineError, RequestedAssertionFailed
from test_driver.logger import AbstractLogger
from test_driver.machine.nspawn_display import NspawnVnc, NspawnVncConfiguration
from test_driver.machine.ocr import (
perform_ocr_on_screenshot,
perform_ocr_variants_on_screenshot,
@@ -101,6 +102,56 @@ CHAR_TO_KEY = {
")": "shift-0x0B",
}
# send_key uses QEMU key names, with "\n" as a driver alias for Return.
# QEMU names: https://www.qemu.org/docs/master/interop/qemu-qmp-ref.html#enum-QKeyCode
# xdotool names: https://github.com/jordansissel/xdotool/blob/main/xdotool.pod#keyboard-commands
# QEMU key name (or "\n") -> X11 keysym name.
XDOTOOL_KEY_ALIASES = {
"\n": "Return",
"alt": "Alt_L",
"alt_r": "Alt_R",
"backspace": "BackSpace",
"delete": "Delete",
"down": "Down",
"esc": "Escape",
"kp_enter": "KP_Enter",
"left": "Left",
"meta_l": "Super_L",
"meta_r": "Super_R",
"ret": "Return",
"right": "Right",
"shift": "Shift_L",
"spc": "space",
"tab": "Tab",
"up": "Up",
}
# QEMU modifier name -> xdotool modifier name.
XDOTOOL_MODIFIER_ALIASES = {
"alt": "alt",
"ctrl": "ctrl",
"meta_l": "super",
"meta_r": "super",
"shift": "shift",
}
def xdotool_key_sequence(key: str) -> str:
"""Translate a QEMU-style key chord to xdotool syntax.
For example, "ctrl-alt-ret" becomes "ctrl+alt+Return".
"""
parts = key.split("-")
modifiers = []
while len(parts) > 1 and parts[0] in XDOTOOL_MODIFIER_ALIASES:
modifiers.append(XDOTOOL_MODIFIER_ALIASES[parts.pop(0)])
base = "-".join(parts)
base = XDOTOOL_KEY_ALIASES.get(base, base)
if re.fullmatch(r"f\d+", base):
base = base.upper()
return "+".join([*modifiers, base])
def make_command(args: list) -> str:
return " ".join(map(shlex.quote, (map(str, args))))
@@ -596,6 +647,160 @@ class BaseMachine(ABC):
with self.nested(f"waiting for file '{filename}'"):
retry(check_file, as_timedelta(timeout))
def get_window_names(self) -> list[str]:
return self.succeed(
r"xwininfo -root -tree | sed 's/.*0x[0-9a-f]* \"\([^\"]*\)\".*/\1/; t; d'"
).splitlines()
def wait_for_window(
self, regexp: str, timeout: Duration = dt.timedelta(minutes=15)
) -> None:
"""
Wait until an X11 window has appeared whose name matches the given
regular expression, e.g., `wait_for_window("Terminal")`.
"""
_warn_if_numeric_duration(timeout, "wait_for_window")
pattern = re.compile(regexp)
def window_is_visible(last_try: bool) -> bool:
names = self.get_window_names()
if last_try:
self.log(
f"Last chance to match {regexp} on the window list,"
+ " which currently contains: "
+ ", ".join(names)
)
return any(pattern.search(name) for name in names)
with self.nested("waiting for a window to appear"):
retry(window_is_visible, as_timedelta(timeout))
@abstractmethod
def send_key(
self,
key: str,
delay: Duration | None = dt.timedelta(milliseconds=10),
log: bool | None = True,
) -> None:
"""
Simulate pressing a key or key chord, e.g.,
`send_key("ctrl-alt-delete")`.
Key names follow QEMU's sendkey vocabulary: `ret` means Enter and
`spc` means Space. Named keys and modifiers are case-sensitive;
use lowercase names, e.g. `f1` or `ctrl-alt-f1`. Literal characters
preserve case: `F` types uppercase F, while `f` types lowercase f.
See https://www.qemu.org/docs/master/interop/qemu-qmp-ref.html#enum-QKeyCode.
Portable key names include printable ASCII characters, function keys,
`tab`, `ret`, `esc`, `spc`, `backspace`, `delete`, `left`, `right`,
`up`, `down`, and `kp_enter`. Chords may use the `ctrl`, `alt`,
`shift`, `meta_l`, and `meta_r` modifiers. Machine backends may accept
additional key names.
"""
...
def send_chars(
self, chars: str, delay: Duration | None = dt.timedelta(milliseconds=10)
) -> None:
r"""
Simulate typing a sequence of characters on the virtual keyboard,
e.g., `send_chars("foobar\n")` will type the string `foobar`
followed by the Enter key.
"""
_warn_if_numeric_duration(delay, "send_chars")
with self.nested(f"sending keys {repr(chars)}"):
for char in chars:
self.send_key(char, delay, log=False)
@contextmanager
def _managed_screenshot(self) -> Generator[Path]:
"""
Take a screenshot and yield the path to its PPM file.
The file will be deleted when leaving the generator.
"""
raise MachineError(f"Screenshots are not supported by {type(self).__name__}")
yield Path()
def screenshot(self, filename: str) -> None:
"""
Take a picture of the display of the machine, in PNG format.
The screenshot will be available in the derivation output.
"""
if "." not in filename:
filename += ".png"
if "/" not in filename:
filename = os.path.join(self.out_dir, filename)
with self.nested(
f"making screenshot {filename}",
{"image": os.path.basename(filename)},
):
with self._managed_screenshot() as screenshot_path:
ret = subprocess.run(
f"pnmtopng '{screenshot_path}' > '{filename}'", shell=True
)
if ret.returncode != 0:
raise MachineError(
f"Cannot convert screenshot (pnmtopng returned code {ret.returncode})"
)
def get_screen_text_variants(self) -> list[str]:
"""
Return a list of different interpretations of what is currently
visible on the machine's screen using optical character
recognition. The number and order of the interpretations is not
specified and is subject to change, but if no exception is raised at
least one will be returned.
::: {.note}
This requires [`enableOCR`](#test-opt-enableOCR) to be set to `true`.
:::
"""
with self._managed_screenshot() as screenshot_path:
return perform_ocr_variants_on_screenshot(screenshot_path)
def get_screen_text(self) -> str:
"""
Return a textual representation of what is currently visible on the
machine's screen using optical character recognition.
::: {.note}
This requires [`enableOCR`](#test-opt-enableOCR) to be set to `true`.
:::
"""
with self._managed_screenshot() as screenshot_path:
return perform_ocr_on_screenshot(screenshot_path)
def wait_for_text(
self, regex: str, timeout: Duration = dt.timedelta(minutes=15)
) -> None:
"""
Wait until the supplied regular expressions matches the textual
contents of the screen by using optical character recognition (see
`get_screen_text` and `get_screen_text_variants`).
::: {.note}
This requires [`enableOCR`](#test-opt-enableOCR) to be set to `true`.
:::
"""
_warn_if_numeric_duration(timeout, "wait_for_text")
def screen_matches(last_try: bool) -> bool:
variants = self.get_screen_text_variants()
for text in variants:
if re.search(regex, text) is not None:
return True
if last_try:
self.log(f"Last OCR attempt failed. Text was: {variants}")
return False
with self.nested(f"waiting for {regex} to appear on screen"):
retry(screen_matches, as_timedelta(timeout))
def wait_for_open_port(
self,
port: int,
@@ -1110,19 +1315,6 @@ class QemuMachine(BaseMachine):
if elapsed >= timeout:
raise TimeoutError
def send_chars(
self, chars: str, delay: Duration | None = dt.timedelta(milliseconds=10)
) -> None:
r"""
Simulate typing a sequence of characters on the virtual keyboard,
e.g., `send_chars("foobar\n")` will type the string `foobar`
followed by the Enter key.
"""
_warn_if_numeric_duration(delay, "send_chars")
with self.nested(f"sending keys {repr(chars)}"):
for char in chars:
self.send_key(char, delay, log=False)
def wait_for_file(
self, filename: str, timeout: Duration = dt.timedelta(minutes=15)
) -> None:
@@ -1199,84 +1391,6 @@ class QemuMachine(BaseMachine):
self.send_monitor_command(f"screendump {screenshot_path}")
yield screenshot_path
def screenshot(self, filename: str) -> None:
"""
Take a picture of the display of the virtual machine, in PNG format.
The screenshot will be available in the derivation output.
"""
if "." not in filename:
filename += ".png"
if "/" not in filename:
filename = os.path.join(self.out_dir, filename)
with self.nested(
f"making screenshot {filename}",
{"image": os.path.basename(filename)},
):
with self._managed_screenshot() as screenshot_path:
ret = subprocess.run(
f"pnmtopng '{screenshot_path}' > '{filename}'", shell=True
)
if ret.returncode != 0:
raise MachineError(
f"Cannot convert screenshot (pnmtopng returned code {ret.returncode})"
)
def get_screen_text_variants(self) -> list[str]:
"""
Return a list of different interpretations of what is currently
visible on the machine's screen using optical character
recognition. The number and order of the interpretations is not
specified and is subject to change, but if no exception is raised at
least one will be returned.
::: {.note}
This requires [`enableOCR`](#test-opt-enableOCR) to be set to `true`.
:::
"""
with self._managed_screenshot() as screenshot_path:
return perform_ocr_variants_on_screenshot(screenshot_path)
def get_screen_text(self) -> str:
"""
Return a textual representation of what is currently visible on the
machine's screen using optical character recognition.
::: {.note}
This requires [`enableOCR`](#test-opt-enableOCR) to be set to `true`.
:::
"""
with self._managed_screenshot() as screenshot_path:
return perform_ocr_on_screenshot(screenshot_path)
def wait_for_text(
self, regex: str, timeout: Duration = dt.timedelta(minutes=15)
) -> None:
"""
Wait until the supplied regular expressions matches the textual
contents of the screen by using optical character recognition (see
`get_screen_text` and `get_screen_text_variants`).
::: {.note}
This requires [`enableOCR`](#test-opt-enableOCR) to be set to `true`.
:::
"""
_warn_if_numeric_duration(timeout, "wait_for_text")
def screen_matches(last_try: bool) -> bool:
variants = self.get_screen_text_variants()
for text in variants:
if re.search(regex, text) is not None:
return True
if last_try:
self.log(f"Last OCR attempt failed. Text was: {variants}")
return False
with self.nested(f"waiting for {regex} to appear on screen"):
retry(screen_matches, as_timedelta(timeout))
def wait_for_console_text(
self, regex: str, timeout: Duration | None = None
) -> None:
@@ -1325,13 +1439,6 @@ class QemuMachine(BaseMachine):
delay: Duration | None = dt.timedelta(milliseconds=10),
log: bool | None = True,
) -> None:
"""
Simulate pressing keys on the virtual keyboard, e.g.,
`send_key("ctrl-alt-delete")`.
Please also refer to the QEMU documentation for more information on the
input syntax: https://en.wikibooks.org/wiki/QEMU/Monitor#sendkey_keys
"""
_warn_if_numeric_duration(delay, "send_key")
key = CHAR_TO_KEY.get(key, key)
context = self.nested(f"sending key {repr(key)}") if log else nullcontext()
@@ -1473,6 +1580,8 @@ class QemuMachine(BaseMachine):
"""
_warn_if_numeric_duration(timeout, "wait_for_x")
# Keep this separate from nspawn's authenticated X probe: QEMU tests may
# install the session user's Xauthority cookie only after this returns.
def check_x(_last_try: bool) -> bool:
cmd = (
"journalctl -b SYSLOG_IDENTIFIER=systemd | "
@@ -1487,34 +1596,6 @@ class QemuMachine(BaseMachine):
with self.nested("waiting for the X11 server"):
retry(check_x, as_timedelta(timeout))
def get_window_names(self) -> list[str]:
return self.succeed(
r"xwininfo -root -tree | sed 's/.*0x[0-9a-f]* \"\([^\"]*\)\".*/\1/; t; d'"
).splitlines()
def wait_for_window(
self, regexp: str, timeout: Duration = dt.timedelta(minutes=15)
) -> None:
"""
Wait until an X11 window has appeared whose name matches the given
regular expression, e.g., `wait_for_window("Terminal")`.
"""
_warn_if_numeric_duration(timeout, "wait_for_window")
pattern = re.compile(regexp)
def window_is_visible(last_try: bool) -> bool:
names = self.get_window_names()
if last_try:
self.log(
f"Last chance to match {regexp} on the window list,"
+ " which currently contains: "
+ ", ".join(names)
)
return any(pattern.search(name) for name in names)
with self.nested("waiting for a window to appear"):
retry(window_is_visible, as_timedelta(timeout))
def forward_port(self, host_port: int = 8080, guest_port: int = 80) -> None:
"""
Forward a TCP port on the host to a TCP port on the guest.
@@ -1606,6 +1687,7 @@ class NspawnMachine(BaseMachine):
machine_sock_path: Path
machine_sock: socket.socket | None
notify_thread: threading.Thread | None
vnc: NspawnVnc | None
@staticmethod
def machine_name_from_start_command(start_command: str) -> str:
@@ -1622,6 +1704,7 @@ class NspawnMachine(BaseMachine):
logger: AbstractLogger,
callbacks: list[Callable] | None = None,
keep_machine_state: bool = False,
vnc: NspawnVncConfiguration | None = None,
):
# TODO: don't compute `name` from `start_command` path, instead thread it down explicitly.
# See analogous TODO in `QemuStartCommand::machine_name`.
@@ -1635,6 +1718,8 @@ class NspawnMachine(BaseMachine):
)
self.start_command = start_command
self.vnc_configuration = vnc
self.vnc = None
self.process = None
self.notify_thread = None
# State maintained by the notify-socket drainer thread (see
@@ -1645,6 +1730,79 @@ class NspawnMachine(BaseMachine):
self.machine_sock_path = self.tmp_dir / f"{self.name}-nspawn.sock"
def wait_for_x(self, timeout: Duration = dt.timedelta(minutes=15)) -> None:
"""
Wait until it is possible to connect to the X server.
"""
_warn_if_numeric_duration(timeout, "wait_for_x")
def check_x(_last_try: bool) -> bool:
status, _ = self.execute("xwininfo -root >/dev/null 2>&1")
return status == 0
with self.nested("waiting for the X11 server"):
retry(check_x, as_timedelta(timeout))
def send_key(
self,
key: str,
delay: Duration | None = dt.timedelta(milliseconds=10),
log: bool | None = True,
) -> None:
_warn_if_numeric_duration(delay, "send_key")
context = self.nested(f"sending key {repr(key)}") if log else nullcontext()
with context:
# Type literal characters to avoid key syntax, e.g. "+" as a chord separator.
if len(key) == 1 and key.isprintable():
command = [
"xdotool",
"type",
"--clearmodifiers",
"--delay",
0,
"--",
key,
]
else:
command = [
"xdotool",
"key",
"--clearmodifiers",
xdotool_key_sequence(key),
]
self.succeed(make_command(command))
if delay is not None:
time.sleep(as_seconds(delay))
@contextmanager
def _managed_screenshot(self) -> Generator[Path]:
# xwd writes inside the container and xwdtopnm reads on the host, so
# the intermediate files must live in their shared directory.
with tempfile.TemporaryDirectory(dir=self.shared_dir) as shared_td:
shared_path = Path(shared_td)
xwd_path = shared_path / "screen.xwd"
ppm_path = shared_path / "screen.ppm"
machine_xwd_path = Path("/tmp/shared") / shared_path.name / xwd_path.name
self.succeed(
make_command(
[
"xwd",
"-root",
"-silent",
"-out",
machine_xwd_path,
]
)
)
with ppm_path.open("wb") as ppm:
ret = subprocess.run(["xwdtopnm", xwd_path], stdout=ppm)
if ret.returncode != 0:
raise MachineError(
f"Cannot convert screenshot (xwdtopnm returned code {ret.returncode})"
)
yield ppm_path
def ssh_backdoor_command(self) -> str:
# documented in systemd-ssh-generator(8) and https://systemd.io/CONTAINER_INTERFACE/
socket_path = f"/run/systemd/nspawn/unix-export/{self.name}/ssh"
@@ -1652,6 +1810,8 @@ class NspawnMachine(BaseMachine):
return f'ssh -o User=root -o ProxyCommand="{proxy_cmd}" bash'
def release(self) -> None:
self._stop_vnc()
if self.process is None:
return
@@ -1776,7 +1936,11 @@ class NspawnMachine(BaseMachine):
# NOTE If the test calls switch-to-configuration (with a differently configured specialization)
# this will use the /etc/profile of the new specialisation while `QemuMachine` nodes
# will continue to use the original /etc/profile.
command = f"set -eo pipefail; USER=root HOME=/root source /etc/profile; set -u; {command}"
command = (
"set -eo pipefail; "
"export USER=root HOME=/root DISPLAY=:0.0; "
f"source /etc/profile; set -u; {command}"
)
cp = subprocess.run(
[
@@ -1895,6 +2059,18 @@ class NspawnMachine(BaseMachine):
journal_thread = threading.Thread(target=self._stream_journal, daemon=True)
journal_thread.start()
if self.vnc_configuration is not None:
if any(name in os.environ for name in ("DISPLAY", "WAYLAND_DISPLAY")):
self.vnc = NspawnVnc(self, self.vnc_configuration)
self.vnc.start()
else:
self.log("no graphical host display available; VNC viewer disabled")
def _stop_vnc(self) -> None:
if self.vnc is not None:
self.vnc.stop()
self.vnc = None
def shutdown(self) -> None:
"""
Shut down the container, waiting for it to exit.
@@ -1917,6 +2093,7 @@ class NspawnMachine(BaseMachine):
timeout=timeout.total_seconds() if timeout is not None else None
)
self.process = None
self._stop_vnc()
class MachineDeprecationWrapper:

201
nixos/lib/test-driver/src/test_driver/machine/nspawn_display.py Normal file
View File

@@ -0,0 +1,201 @@
import datetime as dt
import shlex
import socket
import subprocess
import tempfile
import threading
from pathlib import Path
from typing import TYPE_CHECKING, Self
from pydantic import BaseModel
if TYPE_CHECKING:
from test_driver.machine import NspawnMachine
class NspawnVncConfiguration(BaseModel):
x11vnc: Path
tigervnc: Path
def _container_running(machine: "NspawnMachine") -> bool:
process = machine.process
return process is not None and process.poll() is None
class VncServer:
unit = "nixos-test-vnc"
def __init__(
self,
machine: "NspawnMachine",
x11vnc_executable: Path,
socket_path: Path,
stop_event: threading.Event,
) -> None:
self.machine = machine
self.x11vnc_executable = x11vnc_executable
self.socket_path = socket_path
self.stop_event = stop_event
def _command(self) -> list[str]:
container_socket = Path("/tmp/shared") / self.socket_path.relative_to(
self.machine.shared_dir
)
return [
str(self.x11vnc_executable),
"-display",
":0",
"-auth",
"/root/.Xauthority",
"-unixsock",
str(container_socket),
"-desktop",
f"{self.machine.name} (nspawn)",
# Suppress the password warning for our private Unix socket.
"-nopw",
# Disable IPv4 and IPv6 TCP listeners; use only the Unix socket.
"-rfbport",
"0",
"-rfbportv6",
"0",
]
def __enter__(self) -> Self:
command = shlex.join(
[
"systemd-run",
"--quiet",
"--collect",
"--service-type=exec",
f"--unit={self.unit}",
*self._command(),
]
)
try:
self.machine.succeed(command, timeout=dt.timedelta(seconds=5))
except BaseException:
self._stop()
raise
return self
def wait_ready(self) -> bool:
while not self.stop_event.is_set() and _container_running(self.machine):
with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as connection:
connection.settimeout(1)
try:
connection.connect(str(self.socket_path))
return True
except (FileNotFoundError, ConnectionRefusedError, TimeoutError):
pass
status, _ = self.machine.execute(
f"systemctl is-active --quiet {self.unit}.service",
timeout=dt.timedelta(seconds=5),
)
if status != 0:
raise RuntimeError("VNC server exited before accepting connections")
self.stop_event.wait(0.5)
return False
def _stop(self) -> None:
if _container_running(self.machine):
try:
self.machine.execute(
f"systemctl stop {self.unit}.service",
timeout=dt.timedelta(seconds=5),
)
except Exception as error:
self.machine.log(f"failed to stop the VNC server: {error}")
def __exit__(self, *_: object) -> None:
self._stop()
class VncViewer:
def __init__(
self,
machine: "NspawnMachine",
tigervnc_executable: Path,
socket_path: Path,
stop_event: threading.Event,
) -> None:
self.machine = machine
self.tigervnc_executable = tigervnc_executable
self.socket_path = socket_path
self.stop_event = stop_event
def __enter__(self) -> Self:
self.process = subprocess.Popen(
[str(self.tigervnc_executable), str(self.socket_path)],
# Keep terminal Ctrl-C from reaching the viewer; driver cleanup stops it.
start_new_session=True,
)
self.machine.log(f"VNC viewer running (pid {self.process.pid})")
return self
def wait(self) -> None:
while not self.stop_event.is_set() and _container_running(self.machine):
if self.process.poll() is not None:
self.machine.log(
f"VNC viewer exited (status {self.process.returncode})"
)
return
self.stop_event.wait(0.5)
def __exit__(self, *_: object) -> None:
self.process.terminate()
try:
self.process.wait(timeout=5)
except subprocess.TimeoutExpired:
self.process.kill()
self.process.wait()
class NspawnVnc:
"""Show a container's default X11 display in TigerVNC."""
def __init__(
self, machine: "NspawnMachine", configuration: NspawnVncConfiguration
) -> None:
self.machine = machine
self.configuration = configuration
self.stop_event = threading.Event()
self.thread = threading.Thread(target=self._run, daemon=True)
def start(self) -> None:
self.thread.start()
def stop(self) -> None:
self.stop_event.set()
self.thread.join(timeout=15)
if self.thread.is_alive():
self.machine.log("VNC worker did not stop within 15 seconds")
def _run(self) -> None:
try:
self.machine.wait_for_x()
if self.stop_event.is_set():
return
with tempfile.TemporaryDirectory(dir=self.machine.shared_dir) as directory:
host_socket = Path(directory) / "vnc.sock"
with VncServer(
self.machine,
self.configuration.x11vnc,
host_socket,
self.stop_event,
) as server:
if not server.wait_ready() or self.stop_event.is_set():
return
with VncViewer(
self.machine,
self.configuration.tigervnc,
host_socket,
self.stop_event,
) as viewer:
viewer.wait()
except Exception as error:
if not self.stop_event.is_set():
self.machine.log(f"failed to open the VNC viewer: {error}")

60
nixos/lib/testing/driver-configuration.nix
View File

@@ -7,23 +7,26 @@
let
inherit (lib) types;
nodeConfigurationAttrs = lib.mkOption {
internal = true;
type = types.attrsOf (
types.submodule {
options = {
name = lib.mkOption {
internal = true;
type = types.str;
};
start_script = lib.mkOption {
internal = true;
type = types.path;
};
};
}
);
};
nodeConfigurationAttrs =
extraOptions:
lib.mkOption {
internal = true;
type = types.attrsOf (
types.submodule {
options = {
name = lib.mkOption {
internal = true;
type = types.str;
};
start_script = lib.mkOption {
internal = true;
type = types.path;
};
}
// extraOptions;
}
);
};
in
{
options = {
@@ -32,8 +35,27 @@ in
internal = true;
type = types.submodule {
options = {
vms = nodeConfigurationAttrs;
containers = nodeConfigurationAttrs;
vms = nodeConfigurationAttrs { };
containers = nodeConfigurationAttrs {
vnc = lib.mkOption {
internal = true;
default = null;
type = types.nullOr (
types.submodule {
options = {
x11vnc = lib.mkOption {
internal = true;
type = types.path;
};
tigervnc = lib.mkOption {
internal = true;
type = types.path;
};
};
}
);
};
};
vlans = lib.mkOption {
internal = true;
type = types.listOf types.ints.unsigned;

6
nixos/lib/testing/interactive.nix
View File

@@ -46,6 +46,12 @@ in
config = {
interactive.qemu.package = hostPkgs.qemu;
interactive.extraDriverArgs = [ "--interactive" ];
interactive.driverConfiguration.containers = lib.mapAttrs (_: machine: {
vnc = lib.mkIf machine.services.xserver.enable {
x11vnc = lib.getExe hostPkgs.x11vnc;
tigervnc = lib.getExe' hostPkgs.tigervnc "vncviewer";
};
}) config.containers;
passthru.driverInteractive = config.interactive.driver;
};
}

7
nixos/modules/installer/tools/nixos-generate-config.pl
View File

@@ -24,8 +24,7 @@ sub uniq {
}
sub runCommand {
my ($cmd) = @_;
open FILE, "$cmd 2>&1 |" or die "Failed to execute: $cmd\n";
open FILE, "-|", @_ or die "Failed to execute: @_\n";
my @ret = <FILE>;
close FILE;
return ($?, @ret);
@@ -458,7 +457,7 @@ EOF
# Is this a btrfs filesystem?
if ($fsType eq "btrfs") {
my ($status, @info) = runCommand("@btrfs@ subvol show $rootDir$mountPoint");
my ($status, @info) = runCommand("@btrfs@", "subvol", "show", "$rootDir$mountPoint");
if ($status != 0 || join("", @info) =~ /ERROR:/) {
die "Failed to retrieve subvolume info for $mountPoint\n";
}
@@ -505,7 +504,7 @@ EOF
# This should work for single and multi-device systems.
# still needs subvolume support
if ($fsType eq "bcachefs") {
my ($status, @info) = runCommand("@bcachefs@ fs usage $rootDir$mountPoint");
my ($status, @info) = runCommand("@bcachefs@", "fs", "usage", "$rootDir$mountPoint");
my $UUID = $info[0];
if ($status == 0 && $UUID =~ /^Filesystem:[ \t\n]*([0-9a-z-]+)/) {

6
nixos/modules/services/hardware/udev.nix
View File

@@ -87,7 +87,7 @@ let
--replace-quiet /usr/bin/cat ${pkgs.coreutils}/bin/cat \
--replace-quiet /usr/bin/basename ${pkgs.coreutils}/bin/basename 2>/dev/null
${lib.optionalString (initrdBin != null) ''
substituteInPlace $i --replace-quiet '/run/current-system/systemd' "${lib.removeSuffix "/bin" initrdBin}"
substituteInPlace $i --replace-quiet '/run/current-system/systemd' "${initrdBin}"
''}
done
@@ -475,9 +475,9 @@ in
boot.initrd.systemd.contents = {
"/etc/udev/rules.d".source = udevRulesFor {
name = "initrd-udev-rules";
initrdBin = config.boot.initrd.systemd.contents."/bin".source;
initrdBin = lib.removeSuffix "/bin" config.boot.initrd.systemd.contents."/bin".source;
udevPackages = config.boot.initrd.services.udev.packages;
udevPath = config.boot.initrd.systemd.contents."/bin".source;
udevPath = lib.removeSuffix "/bin" config.boot.initrd.systemd.contents."/bin".source;
udev = config.boot.initrd.systemd.package;
systemd = config.boot.initrd.systemd.package;
binPackages = config.boot.initrd.services.udev.binPackages ++ [

11
nixos/modules/testing/test-instrumentation.nix
View File

@@ -224,7 +224,16 @@ in
];
# `xwininfo' is used by the test driver to query open windows.
environment.systemPackages = [ pkgs.xwininfo ];
environment.systemPackages = [
pkgs.xwininfo
]
++ lib.optionals config.boot.isNspawnContainer [
# Unlike a QEMU machine, an nspawn container has no monitor that can
# capture its display or inject keyboard input, so the test driver uses
# `xwd' and `xdotool' inside the container instead.
pkgs.xdotool
pkgs.xwd
];
# Log everything to the serial console.
services.journald.settings.Journal = {

1
nixos/tests/all-tests.nix
View File

@@ -198,6 +198,7 @@ in
ssh-backdoor = runTestOn [ "x86_64-linux" ] ./nixos-test-driver/ssh-backdoor.nix;
console-log = runTest ./nixos-test-driver/console-log.nix;
containers = runTest ./nixos-test-driver/containers.nix;
nspawn-interactive = runTest ./nixos-test-driver/nspawn-interactive.nix;
nspawn-daemon-reexec-dbus = runTest ./nspawn-daemon-reexec-dbus.nix;
multi-arch-test = runTest ./nixos-test-driver/multi-arch-test.nix;
skip-typecheck = runTest ./nixos-test-driver/skip-typecheck.nix;

33
nixos/tests/common/x11.nix
View File

@@ -1,19 +1,44 @@
{ lib, ... }:
{ config, lib, ... }:
let
isNspawn = config.boot.isNspawnContainer;
in
{
imports = [
./auto.nix
];
services.xserver.enable = true;
services.xserver = lib.mkMerge [
{
enable = true;
# Use IceWM as the window manager.
windowManager.icewm.enable = true;
}
(lib.mkIf isNspawn {
# nspawn containers have no physical display device, so use a headless
# X server for graphical tests.
videoDrivers = [ "dummy" ];
resolutions = [
{
x = 1024;
y = 768;
}
];
# logind correctly marks an nspawn seat without display hardware as
# non-graphical. The dummy X server does not depend on that hardware.
displayManager.lightdm.extraConfig = ''
logind-check-graphical = false
'';
})
];
# Automatically log in.
test-support.displayManager.auto.enable = true;
# Use IceWM as the window manager.
# Don't use a desktop manager.
services.displayManager.defaultSession = lib.mkDefault "none+icewm";
services.xserver.windowManager.icewm.enable = true;
environment.etc = {
# Help with OCR

39
nixos/tests/firefox.nix
View File

@@ -11,12 +11,11 @@
maintainers = [ shlevy ];
};
nodes.machine =
{ pkgs, ... }:
containers.machine =
{ ... }:
{
imports = [ ./common/x11.nix ];
environment.systemPackages = [ pkgs.xdotool ];
programs.firefox = {
enable = true;
@@ -24,15 +23,35 @@
package = firefoxPackage;
};
hardware.alsa = {
enable = true;
enableRecorder = true;
defaultDevice.playback = "pcm.recorder";
services.pipewire = {
enable = false;
alsa.enable = false;
pulse.enable = false;
};
services.pulseaudio = {
enable = true;
systemWide = true;
extraConfig = ''
load-module module-null-sink sink_name=recorder
set-default-sink recorder
'';
};
users.users.root.extraGroups = [ "pulse-access" ];
systemd.services.pulseaudio.wantedBy = [ "multi-user.target" ];
systemd.services.audio-recorder = {
description = "Record NixOS test audio to /tmp/record.wav";
script = "${pkgs.alsa-utils}/bin/arecord -Drecorder -fS16_LE -r48000 -c2 /tmp/record.wav";
after = [ "pulseaudio.service" ];
requires = [ "pulseaudio.service" ];
script = ''
${pkgs.pulseaudio}/bin/parec \
--device=recorder.monitor \
--file-format=wav \
/tmp/record.wav
'';
};
};
@@ -94,10 +113,10 @@
machine.copy_from_machine("/tmp/record.wav")
with subtest("Close sound test tab"):
machine.execute("xdotool key ctrl+w")
machine.send_key("ctrl-w")
with subtest("Close default browser prompt"):
machine.execute("xdotool key space")
machine.send_key("spc")
with subtest("Wait until Firefox draws the developer tool panel"):
machine.sleep(10)

2
nixos/tests/firefox_decrypt.nix
View File

@@ -6,7 +6,7 @@
maintainers = with lib.maintainers; [ schnusch ];
};
nodes.machine =
containers.machine =
{ pkgs, ... }:
{
imports = [ ./common/x11.nix ];

4
nixos/tests/firefoxpwa.nix
View File

@@ -4,7 +4,7 @@
name = "firefoxpwa";
meta.maintainers = with lib.maintainers; [ camillemndn ];
nodes.machine =
containers.machine =
{ pkgs, ... }:
{
imports = [ ./common/x11.nix ];
@@ -19,8 +19,6 @@
};
services.jellyfin.enable = true;
# Jellyfin requires at least 2 GB of disk space
virtualisation.diskSize = 3 * 1024; # 3 GB
};
enableOCR = true;

125
nixos/tests/nixos-test-driver/nspawn-interactive.nix Normal file
View File

@@ -0,0 +1,125 @@
{
config,
hostPkgs,
lib,
...
}:
let
# This test script loads the nested driver's configuration.
# Give that driver an empty script to avoid a circular store reference.
interactiveConfig = hostPkgs.writers.writeJSON "interactive-driver-config.json" (
config.interactive.driverConfiguration // { test_script = hostPkgs.emptyFile; }
);
in
{
name = "nspawn-interactive";
meta.teams = [ lib.teams.test-driver ];
containers.machine = {
imports = [ ../common/x11.nix ];
virtualisation.vlans = [ ];
};
# Run a nested interactive driver under Xvfb, type into the container through
# TigerVNC, and check that closing the viewer leaves the container running.
testScript = ''
import datetime as dt
import os
import subprocess
from collections.abc import Iterator
from contextlib import contextmanager
from test_driver.driver import Driver, load_driver_configuration
def xdotool(*args: str) -> str:
return subprocess.check_output(
["${lib.getExe hostPkgs.xdotool}", *args], text=True, timeout=30
).strip()
def wait_for_host_display() -> None:
def can_connect(_last_try: bool) -> bool:
return subprocess.call(
["${lib.getExe hostPkgs.xdotool}", "getdisplaygeometry"],
stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
) == 0
retry(can_connect, timeout=dt.timedelta(seconds=30))
def wait_for_viewer() -> str:
return xdotool(
"search", "--sync", "--onlyvisible", "--name",
r"machine \(nspawn\)",
)
def open_guest_terminal(machine: NspawnMachine) -> None:
# Keep the shell prompt from replacing the terminal's window title.
machine.execute("xterm -title input-test -e bash --noprofile --norc >&2 &")
machine.wait_for_window("input-test", timeout=dt.timedelta(seconds=30))
def focus_terminal_in_viewer(machine: NspawnMachine, window: str) -> None:
xdotool("windowraise", window)
xdotool("windowfocus", "--sync", window)
xdotool("mousemove", "--window", window, "150", "200")
xdotool("click", "1")
machine.wait_until_succeeds(
'test "$(xdotool getwindowfocus getwindowname)" = input-test',
timeout=dt.timedelta(seconds=30),
)
def run_command_in_terminal(
machine: NspawnMachine, window: str, command: str
) -> None:
focus_terminal_in_viewer(machine, window)
xdotool("type", "--delay", "50", command)
xdotool("key", "Return")
@contextmanager
def test_display() -> Iterator[None]:
# Keep the viewer visible while debugging the outer test interactively.
if driver.interactive and os.environ.get("DISPLAY"):
yield
return
previous_display = os.environ.get("DISPLAY")
with subprocess.Popen([
"${lib.getExe' hostPkgs.xorg-server "Xvfb"}", ":99", "-screen", "0", "1280x1024x24", "-ac",
]) as display:
os.environ["DISPLAY"] = ":99"
try:
yield
finally:
display.terminate()
if previous_display is None:
os.environ.pop("DISPLAY", None)
else:
os.environ["DISPLAY"] = previous_display
with test_display():
wait_for_host_display()
with Driver(
config=load_driver_configuration("${interactiveConfig}"),
out_dir=driver.out_dir,
logger=driver.logger,
interactive=True,
) as interactive:
machine = interactive.machines_nspawn[0]
machine.start()
with subtest("type through the viewer"):
window = wait_for_viewer()
machine.fail("ss -H -lntp | grep -w x11vnc")
open_guest_terminal(machine)
run_command_in_terminal(machine, window, "echo viewer-input > /tmp/viewer-input")
machine.wait_until_succeeds(
"grep -qx viewer-input /tmp/viewer-input",
timeout=dt.timedelta(seconds=30),
)
with subtest("closing the viewer leaves the container running"):
xdotool("windowclose", window)
machine.wait_until_fails("pgrep -x x11vnc")
machine.succeed("test -s /tmp/viewer-input")
'';
}

4
pkgs/by-name/bi/bind/package.nix
View File

@@ -30,11 +30,11 @@
stdenv.mkDerivation (finalAttrs: {
pname = "bind";
version = "9.20.29";
version = "9.20.30";
src = fetchurl {
url = "https://downloads.isc.org/isc/bind9/${finalAttrs.version}/bind-${finalAttrs.version}.tar.xz";
hash = "sha256-WHApUIs7G0Min65BbJflVDq6RYCc76ypilAEoCpXNsE=";
hash = "sha256-nhRDpApyowJyy8MZbQIKay4mk3ERyD5Z4QHoLh47T+8=";
};
outputs = [

4
pkgs/by-name/na/nano/package.nix
View File

@@ -31,11 +31,11 @@ let
in
stdenv.mkDerivation rec {
pname = "nano";
version = "9.2";
version = "9.3";
src = fetchurl {
url = "mirror://gnu/nano/nano-${version}.tar.xz";
hash = "sha256-Bey5kke3guils6Je1BAd0DSwI2kC90SbyXlbcXZC9+k=";
hash = "sha256-qa/8UslZbuBoK83UzrPAsjXrIh2D1zPf2bDf9FiXyw0=";
};
nativeBuildInputs = [ texinfo ] ++ lib.optional enableNls gettext;

6
pkgs/by-name/ru/ruff/package.nix
View File

@@ -16,7 +16,7 @@
rustPlatform.buildRustPackage (finalAttrs: {
pname = "ruff";
version = "0.17.1";
version = "0.17.2";
__structuredAttrs = true;
@@ -24,12 +24,12 @@ rustPlatform.buildRustPackage (finalAttrs: {
owner = "astral-sh";
repo = "ruff";
tag = finalAttrs.version;
hash = "sha256-jpdEdVpp5PWIxn64End1duumm6YJ9EaesBKEBgWoU7Q=";
hash = "sha256-pnEBvVElyBKTc4tsZIMD48H3RQESlZaHR+8TLE6bCDs=";
};
cargoBuildFlags = [ "--package=ruff" ];
cargoHash = "sha256-JrkjRbZ5z0DoUdAWtns6xfKiQLxEhRQSuzYmEngMCBI=";
cargoHash = "sha256-CqLZr1VHgwF4a8d5zQku5Xrl/K/ARYnsSlQriYIieWk=";
nativeBuildInputs = [ installShellFiles ];

6
pkgs/by-name/ty/ty/package.nix
View File

@@ -18,7 +18,7 @@
rustPlatform.buildRustPackage (finalAttrs: {
pname = "ty";
version = "0.0.90";
version = "0.0.91";
__structuredAttrs = true;
src = fetchFromGitHub {
@@ -26,7 +26,7 @@ rustPlatform.buildRustPackage (finalAttrs: {
repo = "ty";
tag = finalAttrs.version;
fetchSubmodules = true;
hash = "sha256-IvyIOqaQZWuIu9+NJwDGf3gQBwZa50s/gZNMJockxH8=";
hash = "sha256-HI3POcbJYDFJpgTg6Y1q8amUNDUrCcrz43MNwzxUb/A=";
};
# For Darwin platforms, remove the integration test for file notifications,
@@ -40,7 +40,7 @@ rustPlatform.buildRustPackage (finalAttrs: {
cargoBuildFlags = [ "--package=ty" ];
cargoHash = "sha256-80Xt1nL3/yw4RNsVFIyayc1MxzX+Ybs8YGgGGTIodhQ=";
cargoHash = "sha256-lzy/H9l6liudRY41zSpAF8I1QWoooASsWFoxvOso1Zo=";
nativeBuildInputs = [ installShellFiles ];
buildInputs = [ rust-jemalloc-sys ];