gador/nixpkgs
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

nixos/test-driver: deprecate referring to the only node via variable machine (#511859)

Browse Source
This commit is contained in:
Jacek Galowicz
2026-05-02 11:49:33 +00:00
committed by GitHub
parent 2e6583c04d ee5465f41a
commit 7a50142dfd
16 changed files with 102 additions and 51 deletions
Download Patch File Download Diff File Expand all files Collapse all files
nixos/lib/test-driver/src/test_driver/__init__.py
+24
View File
@@ -6,6 +6,7 @@ import warnings
from pathlib import Path
import ptpython.ipython
from colorama import Fore, Style
from test_driver.debug import Debug, DebugAbstract, DebugNop
from test_driver.driver import Driver, DriverConfiguration, load_driver_configuration
@@ -55,6 +56,29 @@ def writeable_dir(arg: str) -> Path:
return path
def formatwarning(
message: Warning | str,
category: type[Warning],
filename: str,
lineno: int,
line: str | None = None,
) -> str:
return (
Style.BRIGHT
+ Fore.YELLOW
+ f"??? Warning ({category.__name__}): " # ty: ignore[unsupported-operator]
+ Style.NORMAL
+ str(message)
+ "\n"
+ f' File "{filename}", line {lineno}\n'
+ (f" {line}\n" if line is not None else "")
+ Style.RESET_ALL
)
warnings.formatwarning = formatwarning # ty:ignore[invalid-assignment]
def main() -> None:
arg_parser = argparse.ArgumentParser(prog="nixos-test-driver")
arg_parser.add_argument(
nixos/lib/test-driver/src/test_driver/driver.py
+10 -3
View File
@@ -22,6 +22,7 @@ from test_driver.errors import MachineError, RequestedAssertionFailed
from test_driver.logger import AbstractLogger
from test_driver.machine import (
BaseMachine,
MachineDeprecationWrapper,
NspawnMachine,
QemuMachine,
retry,
@@ -338,11 +339,17 @@ class Driver:
debug=self.debug,
dump_machine_ssh=self.dump_machine_ssh,
)
machine_symbols = {pythonize_name(m.name): m for m in self.machines}
machine_symbols: dict[
str, QemuMachine | NspawnMachine | MachineDeprecationWrapper
] = {pythonize_name(m.name): m for m in self.machines}
# If there's exactly one machine, make it available under the name
# "machine", even if it's not called that.
if len(self.machines) == 1:
(machine_symbols["machine"],) = self.machines
if len(self.machines) == 1 and "machine" not in machine_symbols:
only_machine_name = next(iter(machine_symbols))
machine_symbols["machine"] = MachineDeprecationWrapper(
f"It's deprecated to use the `machine` variable when the only machine is called {only_machine_name}. This behavior will no longer work in NixOS 27.05.",
self.machines[0],
)
vlan_symbols = {
f"vlan{v.nr}": self.vlans[idx] for idx, v in enumerate(self.vlans)
}
nixos/lib/test-driver/src/test_driver/machine/__init__.py
+15
View File
@@ -1710,3 +1710,18 @@ class NspawnMachine(BaseMachine):
with self.nested("waiting for the container to power off"):
self.process.wait()
self.process = None
class MachineDeprecationWrapper:
def __init__(self, msg: str, machine: QemuMachine | NspawnMachine):
self.msg = msg
self.machine = machine
def __getattribute__(self, name: str):
if name in ("msg", "machine"):
return object.__getattribute__(self, name)
typename = self.machine.__class__.__name__
warnings.warn(
f"invoking '{typename}.{name}' is deprecated: {self.msg}",
)
return self.machine.__getattribute__(name)
nixos/tests/benchexec.nix
+1 -1
View File
@@ -32,7 +32,7 @@ in
in
''
start_all()
machine.wait_for_unit("multi-user.target")
benchexec.wait_for_unit("multi-user.target")
benchexec.succeed(''''\
systemd-run \
--property='StandardOutput=file:${stdout}' \
nixos/tests/drawterm.nix
+2
View File
@@ -49,6 +49,8 @@ let
enableOCR = true;
testScript = ''
machine = ${name}
@polling_condition
def drawterm_running():
machine.succeed("pgrep drawterm")
nixos/tests/freetube.nix
+1
View File
@@ -41,6 +41,7 @@ let
testScript = ''
start_all()
machine = ${name}
machine.wait_for_unit('graphical.target')
machine.wait_for_text('Your Subscription list is currently empty')
machine.send_key("ctrl-r")
nixos/tests/mympd.nix
+3 -3
View File
@@ -15,14 +15,14 @@
testScript = ''
start_all();
machine.wait_for_unit("mympd.service");
mympd.wait_for_unit("mympd.service");
# Ensure that mympd can connect to mpd
machine.wait_until_succeeds(
mympd.wait_until_succeeds(
"journalctl -eu mympd -o cat | grep 'Connected to MPD'"
)
# Ensure that the web server is working
machine.succeed("curl http://localhost:8081 --compressed | grep -o myMPD")
mympd.succeed("curl http://localhost:8081 --compressed | grep -o myMPD")
'';
}
nixos/tests/openresty-lua.nix
+1 -1
View File
@@ -96,7 +96,7 @@ in
webserver.succeed("mkdir -p /var/web")
webserver.succeed("chown nginx:nginx /var/web")
webserver.succeed("$(curl -vvv http://sandbox.test/test2-write)")
assert "404 Not Found" in machine.succeed(
assert "404 Not Found" in webserver.succeed(
"curl -vvv -s http://sandbox.test/test2-read/bar.txt"
)
'';
nixos/tests/paisa.nix
+2 -2
View File
@@ -14,8 +14,8 @@
testScript = ''
start_all()
machine.systemctl("start network-online.target")
machine.wait_for_unit("network-online.target")
serviceEmptyConf.systemctl("start network-online.target")
serviceEmptyConf.wait_for_unit("network-online.target")
with subtest("empty/default config test"):
serviceEmptyConf.wait_for_unit("paisa.service")
nixos/tests/quickwit.nix
+5 -5
View File
@@ -75,18 +75,18 @@ in
)
with subtest("verify UI installed"):
machine.succeed("curl -sSf http://127.0.0.1:7280/ui/")
server.succeed("curl -sSf http://127.0.0.1:7280/ui/")
with subtest("injest and query data"):
import json
# Test CLI ingestion
print(machine.succeed('${pkgs.quickwit}/bin/quickwit index create --index-config ${pkgs.writeText "index.yaml" index_yaml}'))
print(server.succeed('${pkgs.quickwit}/bin/quickwit index create --index-config ${pkgs.writeText "index.yaml" index_yaml}'))
# Important to use `--wait`, otherwise the queries below race with index processing.
print(machine.succeed('${pkgs.quickwit}/bin/quickwit index ingest --index example_server_logs --input-path ${pkgs.writeText "exampleDocs.json" exampleDocs} --wait'))
print(server.succeed('${pkgs.quickwit}/bin/quickwit index ingest --index example_server_logs --input-path ${pkgs.writeText "exampleDocs.json" exampleDocs} --wait'))
# Test CLI query
cli_query_output = machine.succeed('${pkgs.quickwit}/bin/quickwit index search --index example_server_logs --query "exception"')
cli_query_output = server.succeed('${pkgs.quickwit}/bin/quickwit index search --index example_server_logs --query "exception"')
print(cli_query_output)
# Assert query result is as expected.
@@ -94,7 +94,7 @@ in
assert num_hits == 3, f"cli_query_output contains unexpected number of results: {num_hits}"
# Test API query
api_query_output = machine.succeed('curl --fail http://127.0.0.1:7280/api/v1/example_server_logs/search?query=exception')
api_query_output = server.succeed('curl --fail http://127.0.0.1:7280/api/v1/example_server_logs/search?query=exception')
print(api_query_output)
quickwit.log(quickwit.succeed(
nixos/tests/snmpd.nix
+2 -2
View File
@@ -17,8 +17,8 @@
testScript = ''
start_all();
machine.wait_for_unit("snmpd.service")
machine.succeed("snmpwalk -v 2c -c public localhost | grep SNMPv2-MIB::sysName.0");
snmpd.wait_for_unit("snmpd.service")
snmpd.succeed("snmpwalk -v 2c -c public localhost | grep SNMPv2-MIB::sysName.0");
'';
}
nixos/tests/systemd-journal-gateway.nix
+3 -3
View File
@@ -45,11 +45,11 @@
client.wait_for_unit("multi-user.target")
def copy_pem(file: str):
machine.copy_from_host(source=f"{tmpdir}/{file}", target=f"/run/secrets/{file}")
machine.succeed(f"chmod 600 /run/secrets/{file} && chown systemd-journal-gateway /run/secrets/{file}")
client.copy_from_host(source=f"{tmpdir}/{file}", target=f"/run/secrets/{file}")
client.succeed(f"chmod 600 /run/secrets/{file} && chown systemd-journal-gateway /run/secrets/{file}")
with subtest("Copying keys and certificates"):
machine.succeed("mkdir -p /run/secrets/{client,server}")
client.succeed("mkdir -p /run/secrets/{client,server}")
copy_pem("server/cert.pem")
copy_pem("server/key.pem")
copy_pem("client/cert.pem")
nixos/tests/tang.nix
+23 -23
View File
@@ -48,44 +48,44 @@
};
testScript = ''
start_all()
machine.wait_for_unit("sockets.target")
server.wait_for_unit("sockets.target")
with subtest("Check keys are generated"):
machine.wait_until_succeeds("curl -v http://127.0.0.1:7654/adv")
key = machine.wait_until_succeeds("tang-show-keys 7654")
server.wait_until_succeeds("curl -v http://127.0.0.1:7654/adv")
key = server.wait_until_succeeds("tang-show-keys 7654")
with subtest("Check systemd access list"):
machine.succeed("ping -c 3 192.168.0.1")
machine.fail("curl -v --connect-timeout 3 http://192.168.0.1:7654/adv")
server.succeed("ping -c 3 192.168.0.1")
server.fail("curl -v --connect-timeout 3 http://192.168.0.1:7654/adv")
with subtest("Check basic encrypt and decrypt message"):
machine.wait_until_succeeds(f"""echo 'Hello World' | clevis encrypt tang '{{ "url": "http://127.0.0.1:7654", "thp":"{key}"}}' > /tmp/encrypted""")
decrypted = machine.wait_until_succeeds("clevis decrypt < /tmp/encrypted")
server.wait_until_succeeds(f"""echo 'Hello World' | clevis encrypt tang '{{ "url": "http://127.0.0.1:7654", "thp":"{key}"}}' > /tmp/encrypted""")
decrypted = server.wait_until_succeeds("clevis decrypt < /tmp/encrypted")
assert decrypted.strip() == "Hello World"
machine.wait_until_succeeds("tang-show-keys 7654")
server.wait_until_succeeds("tang-show-keys 7654")
with subtest("Check encrypt and decrypt disk"):
machine.succeed("cryptsetup luksFormat --force-password --batch-mode /dev/vdb <<<'password'")
machine.succeed(f"""clevis luks bind -s1 -y -f -d /dev/vdb tang '{{ "url": "http://127.0.0.1:7654", "thp":"{key}" }}' <<< 'password' """)
clevis_luks = machine.succeed("clevis luks list -d /dev/vdb")
server.succeed("cryptsetup luksFormat --force-password --batch-mode /dev/vdb <<<'password'")
server.succeed(f"""clevis luks bind -s1 -y -f -d /dev/vdb tang '{{ "url": "http://127.0.0.1:7654", "thp":"{key}" }}' <<< 'password' """)
clevis_luks = server.succeed("clevis luks list -d /dev/vdb")
assert clevis_luks.strip() == """1: tang '{"url":"http://127.0.0.1:7654"}'"""
machine.succeed("clevis luks unlock -d /dev/vdb")
machine.succeed("find /dev/mapper -name 'luks*' -exec cryptsetup close {} +")
machine.succeed("clevis luks unlock -d /dev/vdb")
machine.succeed("find /dev/mapper -name 'luks*' -exec cryptsetup close {} +")
server.succeed("clevis luks unlock -d /dev/vdb")
server.succeed("find /dev/mapper -name 'luks*' -exec cryptsetup close {} +")
server.succeed("clevis luks unlock -d /dev/vdb")
server.succeed("find /dev/mapper -name 'luks*' -exec cryptsetup close {} +")
# without tang available, unlock should fail
machine.succeed("systemctl stop tangd.socket")
machine.fail("clevis luks unlock -d /dev/vdb")
machine.succeed("systemctl start tangd.socket")
server.succeed("systemctl stop tangd.socket")
server.fail("clevis luks unlock -d /dev/vdb")
server.succeed("systemctl start tangd.socket")
with subtest("Rotate server keys"):
machine.succeed("${pkgs.tang}/libexec/tangd-rotate-keys -d /var/lib/tang")
machine.succeed("clevis luks unlock -d /dev/vdb")
machine.succeed("find /dev/mapper -name 'luks*' -exec cryptsetup close {} +")
server.succeed("${pkgs.tang}/libexec/tangd-rotate-keys -d /var/lib/tang")
server.succeed("clevis luks unlock -d /dev/vdb")
server.succeed("find /dev/mapper -name 'luks*' -exec cryptsetup close {} +")
with subtest("Test systemd service security"):
output = machine.succeed("systemd-analyze security tangd@.service")
machine.log(output)
output = server.succeed("systemd-analyze security tangd@.service")
server.log(output)
assert output[-9:-1] == "SAFE :-}"
'';
}
nixos/tests/vscodium.nix
+2
View File
@@ -51,6 +51,8 @@ let
enableOCR = true;
testScript = ''
machine = ${name}
@polling_condition
def codium_running():
machine.succeed('pgrep -x codium')
nixos/tests/web-apps/strichliste.nix
+7 -7
View File
@@ -52,12 +52,12 @@
start_all()
def get_users():
response = machine.succeed("http --check-status http://strichliste.local/api/user")
response = server.succeed("http --check-status http://strichliste.local/api/user")
users = json.loads(response)["users"]
return users
def get_user(uid: int):
response = machine.succeed(f"http --check-status http://strichliste.local/api/user/{uid}")
response = server.succeed(f"http --check-status http://strichliste.local/api/user/{uid}")
user = json.loads(response)["user"]
return user
@@ -67,7 +67,7 @@
t.assertEqual(len(users), 0, "Strichliste must not have users.")
with subtest("Create user"):
machine.succeed("http --check-status post http://strichliste.local/api/user name=Alice")
server.succeed("http --check-status post http://strichliste.local/api/user name=Alice")
users = get_users()
t.assertEqual(len(users), 1, "Strichliste must have exactly one user.")
@@ -77,19 +77,19 @@
t.assertEqual(user["balance"], 0, "New users should have a balance of 0")
with subtest("Deposit money"):
machine.succeed("http --check-status post http://strichliste.local/api/user/1/transaction amount=500")
server.succeed("http --check-status post http://strichliste.local/api/user/1/transaction amount=500")
user = get_user(1)
t.assertEqual(user["balance"], 500, "Balance must be 500 after depositing 500")
with subtest("Dispense money"):
machine.succeed("http --check-status post http://strichliste.local/api/user/1/transaction amount=-1000")
server.succeed("http --check-status post http://strichliste.local/api/user/1/transaction amount=-1000")
user = get_user(1)
t.assertEqual(user["balance"], -500, "Balance must be -500 after dispensing 1000")
with subtest("Undo transaction"):
response = machine.succeed("http --check-status post http://strichliste.local/api/user/1/transaction amount=7500")
response = server.succeed("http --check-status post http://strichliste.local/api/user/1/transaction amount=7500")
transaction = json.loads(response)["transaction"]
machine.succeed(f"http --check-status delete http://strichliste.local/api/user/1/transaction/{transaction['id']}")
server.succeed(f"http --check-status delete http://strichliste.local/api/user/1/transaction/{transaction['id']}")
server.wait_for_unit("phpfpm-strichliste.service")
nixos/tests/windmill/api-integration.nix
+1 -1
View File
@@ -53,7 +53,7 @@
# NOTE; Wait a couple of seconds for all windmill components to finalise their database migration flow. This prevents race conditions on schema constraints.
time.sleep(10) # seconds
windmill.succeed("curl --silent --fail http://windmill:8001")
t.assertIn("v${pkgs.windmill.version}", machine.succeed("curl --silent --fail http://windmill:8001/api/version"), "Mismatched version response")
t.assertIn("v${pkgs.windmill.version}", windmill.succeed("curl --silent --fail http://windmill:8001/api/version"), "Mismatched version response")
with subtest("Validation"):
windmill.succeed("integration-test --language python3 --script ${./python3.script} --input ${./python3.input}")