test/integration-test-wrapper.py

   1 #!/usr/bin/python3
   2 # SPDX-License-Identifier: LGPL-2.1-or-later
   3
   4 '''Test wrapper command for driving integration tests.
   5
   6 Note: This is deliberately rough and only intended to drive existing tests
   7 with the expectation that as part of formally defining the API it will be tidy.
   8
   9 '''
  10
  11 import argparse
  12 import os
  13 import shlex
  14 import subprocess
  15 import sys
  16 import textwrap
  17 from pathlib import Path
  18
  19
  20 EMERGENCY_EXIT_DROPIN = """\
  21 [Unit]
  22 Wants=emergency-exit.service
  23 """
  24
  25
  26 EMERGENCY_EXIT_SERVICE = """\
  27 [Unit]
  28 DefaultDependencies=no
  29 Conflicts=shutdown.target
  30 Conflicts=rescue.service
  31 Before=shutdown.target
  32 Before=rescue.service
  33 FailureAction=exit
  34
  35 [Service]
  36 ExecStart=false
  37 """
  38
  39
  40 def main():
  41     if not bool(int(os.getenv("SYSTEMD_INTEGRATION_TESTS", "0"))):
  42         print("SYSTEMD_INTEGRATION_TESTS=1 not found in environment, skipping", file=sys.stderr)
  43         exit(77)
  44
  45     parser = argparse.ArgumentParser(description=__doc__)
  46     parser.add_argument('--meson-source-dir', required=True, type=Path)
  47     parser.add_argument('--meson-build-dir', required=True, type=Path)
  48     parser.add_argument('--test-name', required=True)
  49     parser.add_argument('--test-number', required=True)
  50     parser.add_argument('--storage', required=True)
  51     parser.add_argument('mkosi_args', nargs="*")
  52     args = parser.parse_args()
  53
  54     test_unit = f"testsuite-{args.test_number}.service"
  55
  56     dropin = textwrap.dedent(
  57         """\
  58         [Unit]
  59         After=multi-user.target network.target
  60         Requires=multi-user.target
  61
  62         [Service]
  63         StandardOutput=journal+console
  64         """
  65     )
  66
  67     if not sys.stderr.isatty():
  68         dropin += textwrap.dedent(
  69             """
  70             [Unit]
  71             SuccessAction=exit
  72             SuccessActionExitStatus=123
  73             FailureAction=exit
  74             """
  75         )
  76
  77         journal_file = (args.meson_build_dir / (f"test/journal/{args.test_name}.journal")).absolute()
  78         journal_file.unlink(missing_ok=True)
  79     else:
  80         journal_file = None
  81
  82     cmd = [
  83         'mkosi',
  84         '--debug',
  85         '--directory', os.fspath(args.meson_source_dir),
  86         '--output-dir', os.fspath(args.meson_build_dir / 'mkosi.output'),
  87         '--extra-search-path', os.fspath(args.meson_build_dir),
  88         '--machine', args.test_name,
  89         '--ephemeral',
  90         *(['--forward-journal', journal_file] if journal_file else []),
  91         *(
  92             [
  93                 '--credential',
  94                 f"systemd.extra-unit.emergency-exit.service={shlex.quote(EMERGENCY_EXIT_SERVICE)}",
  95                 '--credential',
  96                 f"systemd.unit-dropin.emergency.target={shlex.quote(EMERGENCY_EXIT_DROPIN)}",
  97             ]
  98             if not sys.stderr.isatty()
  99             else []
 100         ),
 101         '--credential',
 102         f"systemd.unit-dropin.{test_unit}={shlex.quote(dropin)}",
 103         '--runtime-network=none',
 104         '--runtime-scratch=no',
 105         '--append',
 106         '--kernel-command-line-extra',
 107         ' '.join([
 108             'systemd.hostname=H',
 109             f"SYSTEMD_UNIT_PATH=/usr/lib/systemd/tests/testdata/testsuite-{args.test_number}.units:/usr/lib/systemd/tests/testdata/units:",
 110             f"systemd.unit={test_unit}",
 111             'systemd.mask=systemd-networkd-wait-online.service',
 112             *(
 113                 [
 114                     "systemd.mask=serial-getty@.service",
 115                     "systemd.show_status=no",
 116                     "systemd.crash_shell=0",
 117                     "systemd.crash_action=poweroff",
 118                 ]
 119                 if not sys.stderr.isatty()
 120                 else []
 121             ),
 122         ]),
 123         '--credential', f"journal.storage={'persistent' if sys.stderr.isatty() else args.storage}" ,
 124         *args.mkosi_args,
 125         'qemu',
 126     ]
 127
 128     result = subprocess.run(cmd)
 129     # Return code 123 is the expected success code
 130     if result.returncode != (0 if sys.stderr.isatty() else 123):
 131         if result.returncode != 77 and journal_file:
 132             cmd = [
 133                 'journalctl',
 134                 '--no-hostname',
 135                 '-o', 'short-monotonic',
 136                 '--file', journal_file,
 137                 '-u', test_unit,
 138                 '-p', 'info',
 139             ]
 140             print("Test failed, relevant logs can be viewed with: \n\n"
 141                   f"{shlex.join(str(a) for a in cmd)}\n", file=sys.stderr)
 142         exit(result.returncode or 1)
 143
 144     # Do not keep journal files for tests that don't fail.
 145     if journal_file:
 146         journal_file.unlink(missing_ok=True)
 147
 148
 149 if __name__ == '__main__':
 150     main()