nixos/test-driver: use an argument instead of the out env-var

nixos/lib/test-driver/test_driver/__init__.py

@@ -33,6 +33,18 @@ class EnvDefault(argparse.Action):
         setattr(namespace, self.dest, values)
 
 
+class WriteableDir(argparse.Action):  # type: ignore
+    def __call__(self, parser, namespace, values, option_string=None):  # type: ignore
+        if not values.is_dir():
+            raise argparse.ArgumentTypeError("{0} is not a directory".format(values))
+        if os.access(values, os.W_OK):
+            setattr(namespace, self.dest, values)
+        else:
+            raise argparse.ArgumentTypeError(
+                "{0} is not a writeable directory".format(values)
+            )
+
+
 def main() -> None:
     arg_parser = argparse.ArgumentParser(prog="nixos-test-driver")
     arg_parser.add_argument(
@@ -63,6 +75,13 @@ def main() -> None:
         nargs="*",
         help="vlans to span by the driver",
     )
+    arg_parser.add_argument(
+        "output_directory",
+        action=WriteableDir,
+        help="""The path to the directory where outputs copied from the VM will be placed.
+                By e.g. Machine.copy_from_vm or Machine.screenshot""",
+        type=Path,
+    )
     arg_parser.add_argument(
         "testscript",
         action=EnvDefault,
@@ -77,7 +96,11 @@ def main() -> None:
         rootlog.info("Machine state will be reset. To keep it, pass --keep-vm-state")
 
     with Driver(
-        args.start_scripts, args.vlans, args.testscript.read_text(), args.keep_vm_state
+        args.start_scripts,
+        args.vlans,
+        args.testscript.read_text(),
+        args.output_directory,
+        args.keep_vm_state,
     ) as driver:
         if args.interactive:
             ptpython.repl.embed(driver.test_symbols(), {})
@@ -94,7 +117,7 @@ def generate_driver_symbols() -> None:
     in user's test scripts. That list is then used by pyflakes to lint those
     scripts.
     """
-    d = Driver([], [], "")
+    d = Driver([], [], "", Path())
     test_symbols = d.test_symbols()
     with open("driver-symbols", "w") as fp:
         fp.write(",".join(test_symbols.keys()))

nixos/lib/test-driver/test_driver/driver.py

@@ -24,9 +24,11 @@ class Driver:
         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 = Path(os.environ.get("TMPDIR", tempfile.gettempdir()))
         tmp_dir.mkdir(mode=0o700, exist_ok=True)
@@ -46,7 +48,11 @@ class Driver:
                 keep_vm_state=keep_vm_state,
                 name=cmd.machine_name,
                 tmp_dir=tmp_dir,
+<<<<<<< HEAD
                 callbacks=[self.check_polling_conditions],
+=======
+                out_dir=self.out_dir,
+>>>>>>> 68b2e235272 (nixos/test-driver: use an argument instead of the out env-var)
             )
             for cmd in cmd(start_scripts)
         ]
@@ -154,6 +160,7 @@ class Driver:
 
         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),

nixos/lib/test-driver/test_driver/machine.py

@@ -297,6 +297,7 @@ class Machine:
     the machine lifecycle with the help of a start script / command."""
 
     name: str
+    out_dir: Path
     tmp_dir: Path
     shared_dir: Path
     state_dir: Path
@@ -325,6 +326,7 @@ class Machine:
 
     def __init__(
         self,
+        out_dir: Path,
         tmp_dir: Path,
         start_command: StartCommand,
         name: str = "machine",
@@ -332,6 +334,7 @@ class Machine:
         allow_reboot: bool = False,
         callbacks: Optional[List[Callable]] = None,
     ) -> None:
+        self.out_dir = out_dir
         self.tmp_dir = tmp_dir
         self.keep_vm_state = keep_vm_state
         self.allow_reboot = allow_reboot
@@ -702,10 +705,9 @@ class Machine:
             self.connected = True
 
     def screenshot(self, filename: str) -> None:
-        out_dir = os.environ.get("out", os.getcwd())
         word_pattern = re.compile(r"^\w+$")
         if word_pattern.match(filename):
-            filename = os.path.join(out_dir, "{}.png".format(filename))
+            filename = os.path.join(self.out_dir, "{}.png".format(filename))
         tmp = "{}.ppm".format(filename)
 
         with self.nested(
@@ -756,7 +758,6 @@ class Machine:
         all the VMs (using a temporary directory).
         """
         # Compute the source, target, and intermediate shared file names
-        out_dir = Path(os.environ.get("out", os.getcwd()))
         vm_src = Path(source)
         with tempfile.TemporaryDirectory(dir=self.shared_dir) as shared_td:
             shared_temp = Path(shared_td)
@@ -766,7 +767,7 @@ class Machine:
             # Copy the file to the shared directory inside VM
             self.succeed(make_command(["mkdir", "-p", vm_shared_temp]))
             self.succeed(make_command(["cp", "-r", vm_src, vm_intermediate]))
-            abs_target = out_dir / target_dir / vm_src.name
+            abs_target = self.out_dir / target_dir / vm_src.name
             abs_target.parent.mkdir(exist_ok=True, parents=True)
             # Copy the file from the shared directory outside VM
             if intermediate.is_dir():

nixos/lib/testing-python.nix

@@ -30,7 +30,7 @@ rec {
           # effectively mute the XMLLogger
           export LOGFILE=/dev/null
 
-          ${driver}/bin/nixos-test-driver
+          ${driver}/bin/nixos-test-driver $out
         '';
 
       passthru = driver.passthru // {