nordic-dev.net/nixpkgs
nordic-dev.net/nixpkgs
2
0
Fork 0
mirror of https://github.com/NixOS/nixpkgs.git synced 2024-11-22 23:13:19 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #146905 from Synthetica9/failure_mode

Browse Source 
nixos/test-driver: add polling_condition
This commit is contained in:
Jacek Galowicz 2022-01-08 18:28:25 +01:00 committed by GitHub
commit 048fd95f10
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
8 changed files with 673 additions and 419 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -88,6 +88,8 @@ starting them in parallel:
start_all()
```
## Machine objects {#ssec-machine-objects}
The following methods are available on machine objects:
`start`
@ -313,3 +315,52 @@ repository):
# fmt: on
'';
```
## Failing tests early {#ssec-failing-tests-early}
To fail tests early when certain invariables are no longer met (instead of waiting for the build to time out), the decorator `polling_condition` is provided. For example, if we are testing a program `foo` that should not quit after being started, we might write the following:
```py
@polling_condition
def foo_running():
machine.succeed("pgrep -x foo")
machine.succeed("foo --start")
machine.wait_until_succeeds("pgrep -x foo")
with foo_running:
... # Put `foo` through its paces
```
`polling_condition` takes the following (optional) arguments:
`seconds_interval`
:
specifies how often the condition should be polled:
```py
@polling_condition(seconds_interval=10)
def foo_running():
machine.succeed("pgrep -x foo")
```
`description`
:
is used in the log when the condition is checked. If this is not provided, the description is pulled from the docstring of the function. These two are therefore equivalent:
```py
@polling_condition
def foo_running():
"check that foo is running"
machine.succeed("pgrep -x foo")
```
```py
@polling_condition(description="check that foo is running")
def foo_running():
machine.succeed("pgrep -x foo")
```

138
nixos/doc/manual/from_md/development/writing-nixos-tests.section.xml
View File

@ -117,6 +117,8 @@ if not "Linux" in machine.succeed("uname"):
<programlisting language="python">
start_all()
</programlisting>
<section xml:id="ssec-machine-objects">
<title>Machine objects</title>
<para>
The following methods are available on machine objects:
</para>
@ -194,8 +196,8 @@ start_all()
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.
interpretations is not specified and is subject to change,
but if no exception is raised at least one will be returned.
</para>
<note>
<para>
@ -253,9 +255,9 @@ start_all()
<para>
Simulate typing a sequence of characters on the virtual
keyboard, e.g.,
<literal>send_chars(&quot;foobar\n&quot;)</literal> will type
the string <literal>foobar</literal> followed by the Enter
key.
<literal>send_chars(&quot;foobar\n&quot;)</literal> will
type the string <literal>foobar</literal> followed by the
Enter key.
</para>
</listitem>
</varlistentry>
@ -266,23 +268,24 @@ start_all()
<listitem>
<para>
Execute a shell command, returning a list
<literal>(status, stdout)</literal>. If the command detaches,
it must close stdout, as <literal>execute</literal> will wait
for this to consume all output reliably. This can be achieved
by redirecting stdout to stderr <literal>&gt;&amp;2</literal>,
to <literal>/dev/console</literal>,
<literal>/dev/null</literal> or a file. Examples of detaching
commands are <literal>sleep 365d &amp;</literal>, where the
shell forks a new process that can write to stdout and
<literal>xclip -i</literal>, where the
<literal>xclip</literal> command itself forks without closing
stdout. Takes an optional parameter
<literal>(status, stdout)</literal>. If the command
detaches, it must close stdout, as
<literal>execute</literal> will wait for this to consume all
output reliably. This can be achieved by redirecting stdout
to stderr <literal>&gt;&amp;2</literal>, to
<literal>/dev/console</literal>,
<literal>/dev/null</literal> or a file. Examples of
detaching commands are <literal>sleep 365d &amp;</literal>,
where the shell forks a new process that can write to stdout
and <literal>xclip -i</literal>, where the
<literal>xclip</literal> command itself forks without
closing stdout. Takes an optional parameter
<literal>check_return</literal> that defaults to
<literal>True</literal>. Setting this parameter to
<literal>False</literal> will not check for the return code
and return -1 instead. This can be used for commands that shut
down the VM and would therefore break the pipe that would be
used for retrieving the return code.
and return -1 instead. This can be used for commands that
shut down the VM and would therefore break the pipe that
would be used for retrieving the return code.
</para>
</listitem>
</varlistentry>
@ -300,8 +303,9 @@ start_all()
<itemizedlist>
<listitem>
<para>
If several commands are separated by <literal>;</literal>
and one fails, the command as a whole will fail.
If several commands are separated by
<literal>;</literal> and one fails, the command as a
whole will fail.
</para>
</listitem>
<listitem>
@ -353,7 +357,8 @@ start_all()
</term>
<listitem>
<para>
Repeat a shell command with 1-second intervals until it fails.
Repeat a shell command with 1-second intervals until it
fails.
</para>
</listitem>
</varlistentry>
@ -462,9 +467,9 @@ start_all()
<literal>copy_from_host(&quot;myfile&quot;, &quot;/etc/my/important/file&quot;)</literal>.
</para>
<para>
The first argument is the file on the host. The file needs to
be accessible while building the nix derivation. The second
argument is the location of the file on the machine.
The first argument is the file on the host. The file needs
to be accessible while building the nix derivation. The
second argument is the location of the file on the machine.
</para>
</listitem>
</varlistentry>
@ -490,8 +495,8 @@ machine.systemctl(&quot;list-jobs --no-pager&quot;, &quot;any-user&quot;) # spaw
<listitem>
<para>
Allows you to directly interact with the guest shell. This
should only be used during test development, not in production
tests. Killing the interactive session with
should only be used during test development, not in
production tests. Killing the interactive session with
<literal>Ctrl-d</literal> or <literal>Ctrl-c</literal> also
ends the guest session.
</para>
@ -510,12 +515,13 @@ machine.wait_for_unit(&quot;xautolock.service&quot;, &quot;x-session-user&quot;)
</programlisting>
<para>
This applies to <literal>systemctl</literal>,
<literal>get_unit_info</literal>, <literal>wait_for_unit</literal>,
<literal>start_job</literal> and <literal>stop_job</literal>.
<literal>get_unit_info</literal>,
<literal>wait_for_unit</literal>, <literal>start_job</literal> and
<literal>stop_job</literal>.
</para>
<para>
For faster dev cycles it's also possible to disable the code-linters
(this shouldn't be commited though):
For faster dev cycles it's also possible to disable the
code-linters (this shouldn't be commited though):
</para>
<programlisting language="bash">
import ./make-test-python.nix {
@ -532,10 +538,10 @@ import ./make-test-python.nix {
}
</programlisting>
<para>
This will produce a Nix warning at evaluation time. To fully disable
the linter, wrap the test script in comment directives to disable
the Black linter directly (again, don't commit this within the
Nixpkgs repository):
This will produce a Nix warning at evaluation time. To fully
disable the linter, wrap the test script in comment directives to
disable the Black linter directly (again, don't commit this within
the Nixpkgs repository):
</para>
<programlisting language="bash">
testScript =
@ -546,3 +552,65 @@ import ./make-test-python.nix {
'';
</programlisting>
</section>
<section xml:id="ssec-failing-tests-early">
<title>Failing tests early</title>
<para>
To fail tests early when certain invariables are no longer met
(instead of waiting for the build to time out), the decorator
<literal>polling_condition</literal> is provided. For example, if
we are testing a program <literal>foo</literal> that should not
quit after being started, we might write the following:
</para>
<programlisting language="python">
@polling_condition
def foo_running():
machine.succeed(&quot;pgrep -x foo&quot;)
machine.succeed(&quot;foo --start&quot;)
machine.wait_until_succeeds(&quot;pgrep -x foo&quot;)
with foo_running:
... # Put `foo` through its paces
</programlisting>
<para>
<literal>polling_condition</literal> takes the following
(optional) arguments:
</para>
<para>
<literal>seconds_interval</literal>
</para>
<para>
: specifies how often the condition should be polled:
</para>
<programlisting>
```py
@polling_condition(seconds_interval=10)
def foo_running():
machine.succeed(&quot;pgrep -x foo&quot;)
```
</programlisting>
<para>
<literal>description</literal>
</para>
<para>
: is used in the log when the condition is checked. If this is not
provided, the description is pulled from the docstring of the
function. These two are therefore equivalent:
</para>
<programlisting>
```py
@polling_condition
def foo_running():
&quot;check that foo is running&quot;
machine.succeed(&quot;pgrep -x foo&quot;)
```
```py
@polling_condition(description=&quot;check that foo is running&quot;)
def foo_running():
machine.succeed(&quot;pgrep -x foo&quot;)
```
</programlisting>
</section>
</section>

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

@ -14,7 +14,7 @@
python3Packages.buildPythonApplication rec {
pname = "nixos-test-driver";
version = "1.0";
version = "1.1";
src = ./.;
propagatedBuildInputs = [ coreutils netpbm python3Packages.colorama python3Packages.ptpython qemu_pkg socat vde2 ]
@ -26,7 +26,7 @@ python3Packages.buildPythonApplication rec {
mypy --disallow-untyped-defs \
--no-implicit-optional \
--ignore-missing-imports ${src}/test_driver
pylint --errors-only ${src}/test_driver
pylint --errors-only --enable=unused-import ${src}/test_driver
black --check --diff ${src}/test_driver
'';
}

2
nixos/lib/test-driver/setup.py
View File

@ -2,7 +2,7 @@ from setuptools import setup, find_packages
setup(
name="nixos-test-driver",
version='1.0',
version='1.1',
packages=find_packages(),
entry_points={
"console_scripts": [

41
nixos/lib/test-driver/test_driver/driver.py
View File

@ -1,12 +1,13 @@
from contextlib import contextmanager
from pathlib import Path
from typing import Any, Dict, Iterator, List
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
class Driver:
@ -16,6 +17,7 @@ class Driver:
tests: str
vlans: List[VLan]
machines: List[Machine]
polling_conditions: List[PollingCondition]
def __init__(
self,
@ -36,12 +38,15 @@ class Driver:
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],
)
for cmd in cmd(start_scripts)
]
@ -84,6 +89,7 @@ class Driver:
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}
@ -159,3 +165,36 @@ class Driver:
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_)

9
nixos/lib/test-driver/test_driver/machine.py
View File

@ -318,6 +318,7 @@ class Machine:
# Store last serial console lines for use
# of wait_for_console_text
last_lines: Queue = Queue()
callbacks: List[Callable]
def __repr__(self) -> str:
return f"<Machine '{self.name}'>"
@ -329,12 +330,14 @@ class Machine:
name: str = "machine",
keep_vm_state: bool = False,
allow_reboot: bool = False,
callbacks: Optional[List[Callable]] = None,
) -> None:
self.tmp_dir = tmp_dir
self.keep_vm_state = keep_vm_state
self.allow_reboot = allow_reboot
self.name = name
self.start_command = start_command
self.callbacks = callbacks if callbacks is not None else []
# set up directories
self.shared_dir = self.tmp_dir / "shared-xchg"
@ -406,6 +409,7 @@ class Machine:
return answer
def send_monitor_command(self, command: str) -> str:
self.run_callbacks()
with self.nested("sending monitor command: {}".format(command)):
message = ("{}\n".format(command)).encode()
assert self.monitor is not None
@ -509,6 +513,7 @@ class Machine:
def execute(
self, command: str, check_return: bool = True, timeout: Optional[int] = 900
) -> Tuple[int, str]:
self.run_callbacks()
self.connect()
if timeout is not None:
@ -969,3 +974,7 @@ class Machine:
self.shell.close()
self.monitor.close()
self.serial_thread.join()
def run_callbacks(self) -> None:
for callback in self.callbacks:
callback()

77
nixos/lib/test-driver/test_driver/polling_condition.py Normal file
View File

@ -0,0 +1,77 @@
from typing import Callable, Optional
import time
from .logger import rootlog
class PollingConditionFailed(Exception):
pass
class PollingCondition:
condition: Callable[[], bool]
seconds_interval: float
description: Optional[str]
last_called: float
entered: bool
def __init__(
self,
condition: Callable[[], Optional[bool]],
seconds_interval: float = 2.0,
description: Optional[str] = None,
):
self.condition = condition # type: ignore
self.seconds_interval = seconds_interval
if description is None:
if condition.__doc__:
self.description = condition.__doc__
else:
self.description = condition.__name__
else:
self.description = str(description)
self.last_called = float("-inf")
self.entered = False
def check(self) -> bool:
if self.entered or not self.overdue:
return True
with self, rootlog.nested(self.nested_message):
rootlog.info(f"Time since last: {time.monotonic() - self.last_called:.2f}s")
try:
res = self.condition() # type: ignore
except Exception:
res = False
res = res is None or res
rootlog.info(self.status_message(res))
return res
def maybe_raise(self) -> None:
if not self.check():
raise PollingConditionFailed(self.status_message(False))
def status_message(self, status: bool) -> str:
return f"Polling condition {'succeeded' if status else 'failed'}: {self.description}"
@property
def nested_message(self) -> str:
nested_message = ["Checking polling condition"]
if self.description is not None:
nested_message.append(repr(self.description))
return " ".join(nested_message)
@property
def overdue(self) -> bool:
return self.last_called + self.seconds_interval < time.monotonic()
def __enter__(self) -> None:
self.entered = True
def __exit__(self, exc_type, exc_value, traceback) -> None: # type: ignore
self.entered = False
self.last_called = time.monotonic()

14
nixos/tests/vscodium.nix
View File

@ -34,13 +34,19 @@ let
};
enableOCR = true;
testScript = ''
@polling_condition
def codium_running():
machine.succeed('pgrep -x codium')
start_all()
machine.wait_for_unit('graphical.target')
machine.wait_until_succeeds('pgrep -x codium')
with codium_running:
# Wait until vscodium is visible. "File" is in the menu bar.
machine.wait_for_text('File')
machine.wait_for_text('Get Started')
machine.screenshot('start_screen')
test_string = 'testfile'
@ -63,7 +69,11 @@ let
# (the default filename is the first line of the file)
machine.wait_for_file(f'/home/alice/{test_string}')
machine.send_key('ctrl-q')
machine.wait_until_fails('pgrep -x codium')
'';
});
in builtins.mapAttrs (k: v: mkTest k v { }) tests
in
builtins.mapAttrs (k: v: mkTest k v { }) tests