test_systemd_service_detection.py #2

#!/usr/bin/env python3
"""Service selection regressions from real disposable container provisioning."""
import os
from pathlib import Path
import shutil
import subprocess
import tempfile
import unittest

BIN = Path(__file__).resolve().parents[1] / "Server/Unix/p4/common/bin"
BASH = os.environ.get("SDP_TEST_BASH", shutil.which("bash"))


class ServiceDetection(unittest.TestCase):
    def test_all_service_callers_distinguish_absent_disabled_and_masked_units(self):
        with tempfile.TemporaryDirectory() as directory:
            root = Path(directory)
            mock = root / "systemctl"
            # Behave like systemd 219 (RHEL/CentOS 7): no "show --value", and
            # properties print as Name=value, as newer releases also do without it.
            mock.write_text("#!/bin/sh\nif [ \"$1\" = show ]; then\n  for arg; do [ \"$arg\" = --value ] && { echo \"systemctl: unrecognized option '--value'\" >&2; exit 1; }; done\n  printf 'LoadState=%s\\n' \"$TEST_LOAD_STATE\"; exit \"$TEST_QUERY_STATUS\"\nfi\nprintf 'systemd %s\\n' \"$*\" >> \"$TEST_ACTIONS\"\n[ \"$TEST_LOAD_STATE\" != masked ]\n")
            mock.chmod(0o755)
            init = root / "init"
            init.write_text('#!/bin/sh\nprintf "init %s\\n" "$*" >> "$TEST_ACTIONS"\nexit "${TEST_INIT_STATUS:-0}"\n')
            init.chmod(0o755)
            actions = root / "actions"
            for service in ("p4d", "p4broker", "p4p"):
                for action in ("start", "stop"):
                    for state, query_status, systemd in (("not-found", "0", False), ("not-found", "1", False), ("loaded", "0", True), ("masked", "0", True), ("", "1", False)):
                        with self.subTest(service=service, action=action, state=state, query_status=query_status):
                            actions.unlink(missing_ok=True)
                            script = '''source "$BACKUP_FUNCTIONS"
P4DInitScript=$TEST_INIT
P4BrokerInitScript=$TEST_INIT
P4ProxyInitScript=$TEST_INIT
log () { :; }
die () { exit 97; }
sudo () { "$@"; }
is_server_up () { [[ "$TEST_ACTION" == start ]]; }
"${TEST_ACTION}_${TEST_SERVICE}"
'''
                            env = dict(os.environ, BACKUP_FUNCTIONS=str(BIN / "backup_functions.sh"),
                                       P4CBIN=str(root), LOGFILE=str(root / "log"), P4LOG=str(root / "log"),
                                       PATH=str(root) + ":" + os.environ["PATH"],
                                       TEST_ACTIONS=str(actions), TEST_LOAD_STATE=state, TEST_QUERY_STATUS=query_status,
                                       TEST_SERVICE=service, TEST_ACTION=action,
                                       P4DBIN="p4d_1", P4BROKERBIN="p4broker_1", P4PBIN="p4p_1",
                                       TEST_INIT=str(init))
                            result = subprocess.run([BASH, "-c", script], env=env, capture_output=True, text=True, timeout=5)
                            self.assertEqual(result.returncode, 97 if state == "masked" else 0, result.stderr)
                            self.assertTrue(actions.read_text().startswith("systemd " if systemd else "init "))
                    with self.subTest(service=service, action=action, init_failure=True):
                        env.update(TEST_LOAD_STATE="not-found", TEST_QUERY_STATUS="0", TEST_INIT_STATUS="7")
                        result = subprocess.run([BASH, "-c", script], env=env, capture_output=True, text=True, timeout=5)
                        self.assertEqual(result.returncode, 97, result.stderr)


if __name__ == "__main__":
    unittest.main()