There is a whole lot of noise in a NixOS test log due to journal, commands, and test driver messages all being mixing together. With this commit the test driver messages are highlighted so you don't have to squint too much to see where a subtest starts and ends or what was the last command being run. Here's an excerpt from a highlighted log: [2mmachine # [ 6.647826] dhcpcd[668]: eth0: adding default route via 10.0.2.2[0m [2mmachine # I've had a great day![0m (finished: must succeed: curl -sfL https://example.com/how-are-you? | tee /dev/stderr, in 0.19 seconds) (finished: subtest: Privoxy can filter https requests, in 0.22 seconds) [1m[32msubtest: Blocks are working[0m machine: [1m[32mwaiting for TCP port 443[0m [2mmachine # Connection to localhost (::1) 443 port [tcp/https] succeeded![0m (finished: waiting for TCP port 443, in 0.03 seconds) machine: [1m[32mmust fail: curl -f https://example.com/ads 1>&2[0m [2mmachine # [ 6.730608] nscd[742]: 742 monitored file `/etc/resolv.conf` was written to[0m [2mmachine # [ 6.733071] privoxy[814]: 2022-04-14 09:41:01.695 7fa995c39640 Actions: +block{Fake ads} ... [2mmachine # curl: (22) The requested URL returned error: 403[0m [2mmachine # [ 6.753098] systemd[1]: Stopping Name Service Cache Daemon...[0m [2mmachine # [ 6.755632] systemd[1]: nscd.service: Deactivated successfully.[0m [2mmachine # [ 6.757069] systemd[1]: Stopped Name Service Cache Daemon.[0m (finished: must fail: curl -f https://example.com/ads 1>&2, in 0.07 seconds) machine: [1m[32mmust succeed: curl -f https://example.com/PRIVOXY-FORCE/ads 1>&2[0m [2mmachine # [ 6.765099] systemd[1]: Starting Name Service Cache Daemon...[0m [2mmachine # [ 6.781519] nscd[925]: 925 monitoring file `/etc/passwd` (1)[0m [2mmachine # [ 6.844105] systemd[1]: Reached target Network is Online.[0m [2mmachine # [ 6.848860] privoxy[814]: 2022-04-14 09:41:01.805 7fa995c39640 Actions: +block{Fake ads} ... [2mmachine # [ 6.856068] nscd[925]: 925 monitoring file `/etc/nsswitch.conf` (8)[0m [2mmachine # [ 6.858229] systemd[1]: Reached target Multi-User System.[0m [2mmachine # [ 6.858934] nscd[925]: 925 monitoring directory `/etc` (2)[0m [2mmachine # [ 6.861315] systemd[1]: Startup finished in 2.374s (kernel) + 4.451s (userspace) = 6.826s.[0m [2mmachine # Hot Nixpkgs PRs in your area. Click here![0m [2mmachine # [ 6.873055] nscd[925]: 925 monitoring directory `/etc` (2)[0m [2mmachine # [ 6.874489] nscd[925]: 925 monitoring file `/etc/nsswitch.conf` (8)[0m [2mmachine # [ 6.875599] nscd[925]: 925 monitoring directory `/etc` (2)[0m (finished: must succeed: curl -f https://example.com/PRIVOXY-FORCE/ads 1>&2, in 0.12 seconds) (finished: subtest: Blocks are working, in 0.22 seconds) [1m[32msubtest: Temporary certificates are cleaned[0m machine: [1m[32mmust succeed: test $(ls /run/privoxy/certs | wc -l) -gt 0[0m (finished: must succeed: test $(ls /run/privoxy/certs | wc -l) -gt 0, in 0.02 seconds) machine: [1m[32mmust succeed: date -s "$(date --date '12 days')"[0m (finished: must succeed: date -s "$(date --date '12 days')", in 0.02 seconds) [2mmachine # [ 6.959589] systemd[1]: Started Logrotate Service.[0m [2mmachine # [ 6.966685] systemd[1]: Starting Cleanup of Temporary Directories...[0m [2mmachine # [ 6.974783] systemd[1]: logrotate.service: Deactivated successfully.[0m [2mmachine # [ 7.004493] systemd[1]: systemd-tmpfiles-clean.service: Deactivated successfully.[0m [2mmachine # [ 7.005457] systemd[1]: Finished Cleanup of Temporary Directories.[0m machine: [1m[32mmust succeed: test $(ls /run/privoxy/certs | wc -l) -eq 0[0m (finished: must succeed: test $(ls /run/privoxy/certs | wc -l) -eq 0, in 0.01 seconds) (finished: subtest: Temporary certificates are cleaned, in 0.15 seconds) (finished: run the VM test script, in 7.44 seconds) test script finished in 7.48s [1m[32mcleanup[0m kill machine (pid 8)
227 lines
7.2 KiB
Python
227 lines
7.2 KiB
Python
from contextlib import contextmanager
|
|
from pathlib import Path
|
|
from typing import Any, Dict, Iterator, List, Union, Optional, Callable, ContextManager
|
|
import os
|
|
import tempfile
|
|
|
|
from test_driver.logger import rootlog
|
|
from test_driver.machine import Machine, NixStartScript, retry
|
|
from test_driver.vlan import VLan
|
|
from test_driver.polling_condition import PollingCondition
|
|
|
|
|
|
def get_tmp_dir() -> Path:
|
|
"""Returns a temporary directory that is defined by TMPDIR, TEMP, TMP or CWD
|
|
Raises an exception in case the retrieved temporary directory is not writeable
|
|
See https://docs.python.org/3/library/tempfile.html#tempfile.gettempdir
|
|
"""
|
|
tmp_dir = Path(tempfile.gettempdir())
|
|
tmp_dir.mkdir(mode=0o700, exist_ok=True)
|
|
if not tmp_dir.is_dir():
|
|
raise NotADirectoryError(
|
|
"The directory defined by TMPDIR, TEMP, TMP or CWD: {0} is not a directory".format(
|
|
tmp_dir
|
|
)
|
|
)
|
|
if not os.access(tmp_dir, os.W_OK):
|
|
raise PermissionError(
|
|
"The directory defined by TMPDIR, TEMP, TMP, or CWD: {0} is not writeable".format(
|
|
tmp_dir
|
|
)
|
|
)
|
|
return tmp_dir
|
|
|
|
|
|
class Driver:
|
|
"""A handle to the driver that sets up the environment
|
|
and runs the tests"""
|
|
|
|
tests: str
|
|
vlans: List[VLan]
|
|
machines: List[Machine]
|
|
polling_conditions: List[PollingCondition]
|
|
|
|
def __init__(
|
|
self,
|
|
start_scripts: List[str],
|
|
vlans: List[int],
|
|
tests: str,
|
|
out_dir: Path,
|
|
keep_vm_state: bool = False,
|
|
):
|
|
self.tests = tests
|
|
self.out_dir = out_dir
|
|
|
|
tmp_dir = get_tmp_dir()
|
|
|
|
with rootlog.nested("start all VLans"):
|
|
vlans = list(set(vlans))
|
|
self.vlans = [VLan(nr, tmp_dir) for nr in vlans]
|
|
|
|
def cmd(scripts: List[str]) -> Iterator[NixStartScript]:
|
|
for s in scripts:
|
|
yield NixStartScript(s)
|
|
|
|
self.polling_conditions = []
|
|
|
|
self.machines = [
|
|
Machine(
|
|
start_command=cmd,
|
|
keep_vm_state=keep_vm_state,
|
|
name=cmd.machine_name,
|
|
tmp_dir=tmp_dir,
|
|
callbacks=[self.check_polling_conditions],
|
|
out_dir=self.out_dir,
|
|
)
|
|
for cmd in cmd(start_scripts)
|
|
]
|
|
|
|
def __enter__(self) -> "Driver":
|
|
return self
|
|
|
|
def __exit__(self, *_: Any) -> None:
|
|
with rootlog.nested("cleanup"):
|
|
for machine in self.machines:
|
|
machine.release()
|
|
|
|
def subtest(self, name: str) -> Iterator[None]:
|
|
"""Group logs under a given test name"""
|
|
with rootlog.nested("subtest: " + name):
|
|
try:
|
|
yield
|
|
return True
|
|
except Exception as e:
|
|
rootlog.error(f'Test "{name}" failed with error: "{e}"')
|
|
raise e
|
|
|
|
def test_symbols(self) -> Dict[str, Any]:
|
|
@contextmanager
|
|
def subtest(name: str) -> Iterator[None]:
|
|
return self.subtest(name)
|
|
|
|
general_symbols = dict(
|
|
start_all=self.start_all,
|
|
test_script=self.test_script,
|
|
machines=self.machines,
|
|
vlans=self.vlans,
|
|
driver=self,
|
|
log=rootlog,
|
|
os=os,
|
|
create_machine=self.create_machine,
|
|
subtest=subtest,
|
|
run_tests=self.run_tests,
|
|
join_all=self.join_all,
|
|
retry=retry,
|
|
serial_stdout_off=self.serial_stdout_off,
|
|
serial_stdout_on=self.serial_stdout_on,
|
|
polling_condition=self.polling_condition,
|
|
Machine=Machine, # for typing
|
|
)
|
|
machine_symbols = {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
|
|
vlan_symbols = {
|
|
f"vlan{v.nr}": self.vlans[idx] for idx, v in enumerate(self.vlans)
|
|
}
|
|
print(
|
|
"additionally exposed symbols:\n "
|
|
+ ", ".join(map(lambda m: m.name, self.machines))
|
|
+ ",\n "
|
|
+ ", ".join(map(lambda v: f"vlan{v.nr}", self.vlans))
|
|
+ ",\n "
|
|
+ ", ".join(list(general_symbols.keys()))
|
|
)
|
|
return {**general_symbols, **machine_symbols, **vlan_symbols}
|
|
|
|
def test_script(self) -> None:
|
|
"""Run the test script"""
|
|
with rootlog.nested("run the VM test script"):
|
|
symbols = self.test_symbols() # call eagerly
|
|
exec(self.tests, symbols, None)
|
|
|
|
def run_tests(self) -> None:
|
|
"""Run the test script (for non-interactive test runs)"""
|
|
self.test_script()
|
|
# TODO: Collect coverage data
|
|
for machine in self.machines:
|
|
if machine.is_up():
|
|
machine.execute("sync")
|
|
|
|
def start_all(self) -> None:
|
|
"""Start all machines"""
|
|
with rootlog.nested("start all VMs"):
|
|
for machine in self.machines:
|
|
machine.start()
|
|
|
|
def join_all(self) -> None:
|
|
"""Wait for all machines to shut down"""
|
|
with rootlog.nested("wait for all VMs to finish"):
|
|
for machine in self.machines:
|
|
machine.wait_for_shutdown()
|
|
|
|
def create_machine(self, args: Dict[str, Any]) -> Machine:
|
|
rootlog.warning(
|
|
"Using legacy create_machine(), please instantiate the"
|
|
"Machine class directly, instead"
|
|
)
|
|
|
|
tmp_dir = get_tmp_dir()
|
|
|
|
if args.get("startCommand"):
|
|
start_command: str = args.get("startCommand", "")
|
|
cmd = NixStartScript(start_command)
|
|
name = args.get("name", cmd.machine_name)
|
|
else:
|
|
cmd = Machine.create_startcommand(args) # type: ignore
|
|
name = args.get("name", "machine")
|
|
|
|
return Machine(
|
|
tmp_dir=tmp_dir,
|
|
out_dir=self.out_dir,
|
|
start_command=cmd,
|
|
name=name,
|
|
keep_vm_state=args.get("keep_vm_state", False),
|
|
allow_reboot=args.get("allow_reboot", False),
|
|
)
|
|
|
|
def serial_stdout_on(self) -> None:
|
|
rootlog._print_serial_logs = True
|
|
|
|
def serial_stdout_off(self) -> None:
|
|
rootlog._print_serial_logs = False
|
|
|
|
def check_polling_conditions(self) -> None:
|
|
for condition in self.polling_conditions:
|
|
condition.maybe_raise()
|
|
|
|
def polling_condition(
|
|
self,
|
|
fun_: Optional[Callable] = None,
|
|
*,
|
|
seconds_interval: float = 2.0,
|
|
description: Optional[str] = None,
|
|
) -> Union[Callable[[Callable], ContextManager], ContextManager]:
|
|
driver = self
|
|
|
|
class Poll:
|
|
def __init__(self, fun: Callable):
|
|
self.condition = PollingCondition(
|
|
fun,
|
|
seconds_interval,
|
|
description,
|
|
)
|
|
|
|
def __enter__(self) -> None:
|
|
driver.polling_conditions.append(self.condition)
|
|
|
|
def __exit__(self, a, b, c) -> None: # type: ignore
|
|
res = driver.polling_conditions.pop()
|
|
assert res is self.condition
|
|
|
|
if fun_ is None:
|
|
return Poll
|
|
else:
|
|
return Poll(fun_)
|