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 json
  13 import os
  14 import shlex
  15 import subprocess
  16 import sys
  17 import textwrap
  18 from pathlib import Path
  19
  20
  21 EMERGENCY_EXIT_DROPIN = """\
  22 [Unit]
  23 Wants=emergency-exit.service
  24 """
  25
  26
  27 EMERGENCY_EXIT_SERVICE = """\
  28 [Unit]
  29 DefaultDependencies=no
  30 Conflicts=shutdown.target
  31 Conflicts=rescue.service
  32 Before=shutdown.target
  33 Before=rescue.service
  34 FailureAction=exit
  35
  36 [Service]
  37 ExecStart=false
  38 """
  39
  40
  41 def main():
  42     parser = argparse.ArgumentParser(description=__doc__)
  43     parser.add_argument('--meson-source-dir', required=True, type=Path)
  44     parser.add_argument('--meson-build-dir', required=True, type=Path)
  45     parser.add_argument('--name', required=True)
  46     parser.add_argument('--unit', required=True)
  47     parser.add_argument('--storage', required=True)
  48     parser.add_argument('--firmware', required=True)
  49     parser.add_argument('--slow', action=argparse.BooleanOptionalAction)
  50     parser.add_argument('mkosi_args', nargs="*")
  51     args = parser.parse_args()
  52
  53     if not bool(int(os.getenv("SYSTEMD_INTEGRATION_TESTS", "0"))):
  54         print(f"SYSTEMD_INTEGRATION_TESTS=1 not found in environment, skipping {args.name}", file=sys.stderr)
  55         exit(77)
  56
  57     if args.slow and not bool(int(os.getenv("SYSTEMD_SLOW_TESTS", "0"))):
  58         print(f"SYSTEMD_SLOW_TESTS=1 not found in environment, skipping {args.name}", file=sys.stderr)
  59         exit(77)
  60
  61     name = args.name + (f"-{i}" if (i := os.getenv("MESON_TEST_ITERATION")) else "")
  62
  63     dropin = textwrap.dedent(
  64         """\
  65         [Unit]
  66         SuccessAction=exit
  67         SuccessActionExitStatus=123
  68
  69         [Service]
  70         StandardOutput=journal+console
  71         """
  72     )
  73
  74     if os.getenv("TEST_MATCH_SUBTEST"):
  75         dropin += textwrap.dedent(
  76             f"""
  77             [Service]
  78             Environment=TEST_MATCH_SUBTEST={os.environ["TEST_MATCH_SUBTEST"]}
  79             """
  80         )
  81
  82     if os.getenv("TEST_MATCH_TESTCASE"):
  83         dropin += textwrap.dedent(
  84             f"""
  85             [Service]
  86             Environment=TEST_MATCH_TESTCASE={os.environ["TEST_MATCH_TESTCASE"]}
  87             """
  88         )
  89
  90     if not sys.stderr.isatty():
  91         dropin += textwrap.dedent(
  92             """
  93             [Unit]
  94             FailureAction=exit
  95             """
  96         )
  97
  98         journal_file = (args.meson_build_dir / (f"test/journal/{name}.journal")).absolute()
  99         journal_file.unlink(missing_ok=True)
 100     else:
 101         journal_file = None
 102
 103     cmd = [
 104         'mkosi',
 105         '--directory', os.fspath(args.meson_source_dir),
 106         '--output-dir', os.fspath(args.meson_build_dir / 'mkosi.output'),
 107         '--extra-search-path', os.fspath(args.meson_build_dir),
 108         '--machine', name,
 109         '--ephemeral',
 110         *(['--forward-journal', journal_file] if journal_file else []),
 111         *(
 112             [
 113                 '--credential',
 114                 f"systemd.extra-unit.emergency-exit.service={shlex.quote(EMERGENCY_EXIT_SERVICE)}",
 115                 '--credential',
 116                 f"systemd.unit-dropin.emergency.target={shlex.quote(EMERGENCY_EXIT_DROPIN)}",
 117             ]
 118             if not sys.stderr.isatty()
 119             else []
 120         ),
 121         '--credential',
 122         f"systemd.unit-dropin.{args.unit}={shlex.quote(dropin)}",
 123         '--runtime-network=none',
 124         '--runtime-scratch=no',
 125         *args.mkosi_args,
 126         '--append',
 127         '--qemu-firmware', args.firmware,
 128         '--kernel-command-line-extra',
 129         ' '.join([
 130             'systemd.hostname=H',
 131             f"SYSTEMD_UNIT_PATH=/usr/lib/systemd/tests/testdata/{args.name}.units:/usr/lib/systemd/tests/testdata/units:",
 132             f"systemd.unit={args.unit}",
 133             'systemd.mask=systemd-networkd-wait-online.service',
 134             *(
 135                 [
 136                     "systemd.mask=serial-getty@.service",
 137                     "systemd.show_status=no",
 138                     "systemd.crash_shell=0",
 139                     "systemd.crash_action=poweroff",
 140                 ]
 141                 if not sys.stderr.isatty()
 142                 else []
 143             ),
 144         ]),
 145         '--credential', f"journal.storage={'persistent' if sys.stderr.isatty() else args.storage}",
 146         'qemu',
 147     ]
 148
 149     result = subprocess.run(cmd)
 150
 151     # Return code 123 is the expected success code
 152     if result.returncode in (123, 77):
 153         # Do not keep journal files for tests that don't fail.
 154         if journal_file:
 155             journal_file.unlink(missing_ok=True)
 156
 157         exit(0 if result.returncode == 123 else 77)
 158
 159     if journal_file:
 160         ops = []
 161
 162         if os.getenv("GITHUB_ACTIONS"):
 163             id = os.environ["GITHUB_RUN_ID"]
 164             iteration = os.environ["GITHUB_RUN_ATTEMPT"]
 165             j = json.loads(
 166                 subprocess.run(
 167                     [
 168                         "mkosi",
 169                         "--directory", os.fspath(args.meson_source_dir),
 170                         "--json",
 171                         "summary",
 172                     ],
 173                     stdout=subprocess.PIPE,
 174                     text=True,
 175                 ).stdout
 176             )
 177             images = {image["Image"]: image for image in j["Images"]}
 178             distribution = images["system"]["Distribution"]
 179             release = images["system"]["Release"]
 180             artifact = f"ci-mkosi-{id}-{iteration}-{distribution}-{release}-failed-test-journals"
 181             ops += [f"gh run download {id} --name {artifact} -D ci/{artifact}"]
 182             journal_file = Path(f"ci/{artifact}/test/journal/{name}.journal")
 183
 184         ops += [f"journalctl --file {journal_file} --no-hostname -o short-monotonic -u {args.unit} -p info"]
 185
 186         print("Test failed, relevant logs can be viewed with: \n\n"
 187               f"{(' && '.join(ops))}\n", file=sys.stderr)
 188
 189     # 0 also means we failed so translate that to a non-zero exit code to mark the test as failed.
 190     exit(result.returncode or 1)
 191
 192
 193 if __name__ == '__main__':
 194     main()