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('--firmware', required=True)
  52     parser.add_argument('mkosi_args', nargs="*")
  53     args = parser.parse_args()
  54
  55     name = args.test_name + (f"-{i}" if (i := os.getenv("MESON_TEST_ITERATION")) else "")
  56     test_unit = f"testsuite-{args.test_number}.service"
  57
  58     dropin = textwrap.dedent(
  59         """\
  60         [Unit]
  61         After=multi-user.target network.target
  62         Requires=multi-user.target
  63         SuccessAction=exit
  64         SuccessActionExitStatus=123
  65
  66         [Service]
  67         StandardOutput=journal+console
  68         """
  69     )
  70
  71     if os.getenv("TEST_MATCH_SUBTEST"):
  72         dropin += textwrap.dedent(
  73             f"""
  74             [Service]
  75             Environment=TEST_MATCH_SUBTEST={os.environ["TEST_MATCH_SUBTEST"]}
  76             """
  77         )
  78
  79     if os.getenv("TEST_MATCH_TESTCASE"):
  80         dropin += textwrap.dedent(
  81             f"""
  82             [Service]
  83             Environment=TEST_MATCH_TESTCASE={os.environ["TEST_MATCH_TESTCASE"]}
  84             """
  85         )
  86
  87     if not sys.stderr.isatty():
  88         dropin += textwrap.dedent(
  89             """
  90             [Unit]
  91             FailureAction=exit
  92             """
  93         )
  94
  95         journal_file = (args.meson_build_dir / (f"test/journal/{name}.journal")).absolute()
  96         journal_file.unlink(missing_ok=True)
  97     else:
  98         journal_file = None
  99
 100     cmd = [
 101         'mkosi',
 102         '--directory', os.fspath(args.meson_source_dir),
 103         '--output-dir', os.fspath(args.meson_build_dir / 'mkosi.output'),
 104         '--extra-search-path', os.fspath(args.meson_build_dir),
 105         '--machine', name,
 106         '--ephemeral',
 107         *(['--forward-journal', journal_file] if journal_file else []),
 108         *(
 109             [
 110                 '--credential',
 111                 f"systemd.extra-unit.emergency-exit.service={shlex.quote(EMERGENCY_EXIT_SERVICE)}",
 112                 '--credential',
 113                 f"systemd.unit-dropin.emergency.target={shlex.quote(EMERGENCY_EXIT_DROPIN)}",
 114             ]
 115             if not sys.stderr.isatty()
 116             else []
 117         ),
 118         '--credential',
 119         f"systemd.unit-dropin.{test_unit}={shlex.quote(dropin)}",
 120         '--runtime-network=none',
 121         '--runtime-scratch=no',
 122         '--append',
 123         '--qemu-firmware', args.firmware,
 124         '--kernel-command-line-extra',
 125         ' '.join([
 126             'systemd.hostname=H',
 127             f"SYSTEMD_UNIT_PATH=/usr/lib/systemd/tests/testdata/testsuite-{args.test_number}.units:/usr/lib/systemd/tests/testdata/units:",
 128             f"systemd.unit={test_unit}",
 129             'systemd.mask=systemd-networkd-wait-online.service',
 130             *(
 131                 [
 132                     "systemd.mask=serial-getty@.service",
 133                     "systemd.show_status=no",
 134                     "systemd.crash_shell=0",
 135                     "systemd.crash_action=poweroff",
 136                 ]
 137                 if not sys.stderr.isatty()
 138                 else []
 139             ),
 140         ]),
 141         '--credential', f"journal.storage={'persistent' if sys.stderr.isatty() else args.storage}",
 142         *args.mkosi_args,
 143         'qemu',
 144     ]
 145
 146     result = subprocess.run(cmd)
 147
 148     # Return code 123 is the expected success code
 149     if result.returncode in (123, 77):
 150         # Do not keep journal files for tests that don't fail.
 151         if journal_file:
 152             journal_file.unlink(missing_ok=True)
 153
 154         exit(0 if result.returncode == 123 else 77)
 155
 156     if journal_file:
 157         cmd = [
 158             'journalctl',
 159             '--no-hostname',
 160             '-o', 'short-monotonic',
 161             '--file', journal_file,
 162             '-u', test_unit,
 163             '-p', 'info',
 164         ]
 165         print("Test failed, relevant logs can be viewed with: \n\n"
 166               f"{shlex.join(str(a) for a in cmd)}\n", file=sys.stderr)
 167
 168     # 0 also means we failed so translate that to a non-zero exit code to mark the test as failed.
 169     exit(result.returncode or 1)
 170
 171
 172 if __name__ == '__main__':
 173     main()