A script to test the difference between Requires, PartOf, and BindsTo of a systemd.unit

requires_partof_bindsto_diff.py

#! /usr/bin/env python
# -*- coding: utf-8 -*-
# vim:fenc=utf-8
#
# Copyright © 2021 Pi-Yueh Chuang <pychuang@pm.me>
#
# Distributed under terms of the BSD 3-Clause license.

"""Test the difference between Requires=, PartOf=, and BindsTo= of a systemd.unit.
"""
import os
import time
import atexit
import subprocess
import pandas


def kernel(command):
    """Execute a command and return the status of the two services.

    Arguments
    ---------
    command : str;
        The target systemd command.

    Returns
    -------
        A string of either V/V, V/X, X/V. or X/X. The first character indicates the status of a.service, and the last
        character indicated the status of b.service. "V" means the service is running, while "X" means otherwise.
    """
    _ = subprocess.run("systemctl --user daemon-reload", shell=True, check=True)

    _ = subprocess.run(command, shell=True, check=True)
    time.sleep(1)

    respond_a = subprocess.run("systemctl --user is-active a.service", shell=True, capture_output=True, check=False)
    respond_b = subprocess.run("systemctl --user is-active b.service", shell=True, capture_output=True, check=False)
    respond_a = respond_a.stdout.decode("utf-8").strip()
    respond_b = respond_b.stdout.decode("utf-8").strip()

    _ = subprocess.run("systemctl --user reset-failed a.service", shell=True, capture_output=True, check=False)
    _ = subprocess.run("systemctl --user reset-failed b.service", shell=True, capture_output=True, check=False)
    _ = subprocess.run("systemctl --user stop a.service b.service", shell=True, check=True)

    convert = {"active": "V", "inactive": "X", "failed": "X"}
    return "{}/{}".format(convert[respond_a], convert[respond_b])


def main(html=False):
    """Main function.

    Arguments
    ---------
    html : bool
        Whether to output html or just string.

    Returns
    -------
    If html is True, return a html representation of the result table. Otherwise, return a string representation
    of the result table that can be used in `print`.
    """

    # path to a.service
    a_service_path = os.path.join(os.path.expanduser("~"), ".config", "systemd", "user", "a.service")

    # path to b.service
    b_service_path = os.path.join(os.path.expanduser("~"), ".config", "systemd", "user", "b.service")

    # let the program automatically remove service files when the program exits, no matter how
    atexit.register(os.remove, a_service_path)
    atexit.register(os.remove, b_service_path)

    # commands to be tested
    test_cmds = [
        "systemctl --user start a.service",
        "systemctl --user start b.service",
        "systemctl --user start a.service b.service;" + "systemctl --user stop a.service",
        "systemctl --user start a.service b.service;" + "systemctl --user stop b.service",
        "systemctl --user start a.service b.service;" +
        "kill -9 $(systemctl --user show a.service | grep -oP '(?<=ExecMainPID=)\\d*')",
        "systemctl --user start a.service b.service;" +
        "kill -9 $(systemctl --user show b.service | grep -oP '(?<=ExecMainPID=)\\d*')",
        "systemctl --user start a.service b.service;" + "systemctl --user restart a.service",
        "systemctl --user start a.service b.service;" + "systemctl --user restart b.service",
    ]

    # result holder
    result = pandas.DataFrame(
        index=["Test {}".format(i+1) for i in range(len(test_cmds))],
        columns=["Requires", "PartOf", "BindsTo"],
        dtype=str
    )

    # loop through different experiment groups
    for tag in ["Requires", "PartOf", "BindsTo"]:

        # write to a.service
        with open(a_service_path, "w") as fobj:
            fobj.writelines([
                "[Unit]\n",
                "Description=Test target A\n",
                "[Service]\n",
                "Type=simple\n",
                "ExecStart=sh -c 'while true; do echo A is alive; sleep 3; done'\n",
            ])

        # write to a.service
        with open(b_service_path, "w") as fobj:
            fobj.writelines([
                "[Unit]\n",
                "Description=Test service B.\n",
                "{tag}=a.service\n".format(tag=tag),
                "[Service]\n",
                "Type=simple\n",
                "ExecStart=sh -c 'while true; do echo B is alive; sleep 3; done'\n",
            ])

        for i, cmd in enumerate(test_cmds):
            print("Running experiment group of {} test {}".format(tag, i+1))
            result.loc["Test {}".format(i+1), tag] = kernel(cmd)

    return result.to_html() if html else result.to_string()


if __name__ == "__main__":
    print(main())