Adding debian version 247.3-7+deb11u4.debian/247.3-7+deb11u4

Signed-off-by: Daniel Baumann <daniel.baumann@progress-linux.org>

1 files changed, 297 insertions, 0 deletions

diff --git a/debian/tests/systemd-fsckd b/debian/tests/systemd-fsckd
new file mode 100755
index 0000000..09d68f5
--- /dev/null
+++ b/debian/tests/systemd-fsckd
@@ -0,0 +1,297 @@
+#!/usr/bin/python3
+# autopkgtest check: Ensure that systemd-fsckd can report progress and cancel
+# (C) 2015 Canonical Ltd.
+# Author: Didier Roche <didrocks@ubuntu.com>
+
+from contextlib import suppress
+import inspect
+import fileinput
+import os
+import subprocess
+import shutil
+import stat
+import sys
+import unittest
+from time import sleep, time
+
+GRUB_AUTOPKGTEST_CONFIG_PATH = "/etc/default/grub.d/50-cloudimg-settings.cfg"
+TEST_AUTOPKGTEST_CONFIG_PATH = "/etc/default/grub.d/99-fsckdtest.cfg"
+
+SYSTEMD_ETC_SYSTEM_UNIT_DIR = "/etc/systemd/system/"
+SYSTEMD_PROCESS_KILLER_PATH = os.path.join(SYSTEMD_ETC_SYSTEM_UNIT_DIR, "process-killer.service")
+
+SYSTEMD_FSCK_ROOT_PATH = "/lib/systemd/system/systemd-fsck-root.service"
+SYSTEMD_FSCK_ROOT_ENABLE_PATH = os.path.join(SYSTEMD_ETC_SYSTEM_UNIT_DIR, 'local-fs.target.wants/systemd-fsck-root.service')
+
+SYSTEM_FSCK_PATH = '/sbin/fsck'
+PROCESS_KILLER_PATH = '/sbin/process-killer'
+SAVED_FSCK_PATH = "{}.real".format(SYSTEM_FSCK_PATH)
+
+FSCKD_TIMEOUT = 30
+
+
+class FsckdTest(unittest.TestCase):
+    '''Check that we run, report and can cancel fsck'''
+
+    def __init__(self, test_name, after_reboot, return_code):
+        super().__init__(test_name)
+        self._test_name = test_name
+        self._after_reboot = after_reboot
+        self._return_code = return_code
+
+    def setUp(self):
+        super().setUp()
+        # ensure we have our root fsck enabled by default (it detects it runs in a vm and doesn't pull the target)
+        # note that it can already exists in case of a reboot (as there was no tearDown as we wanted)
+        os.makedirs(os.path.dirname(SYSTEMD_FSCK_ROOT_ENABLE_PATH), exist_ok=True)
+        with suppress(FileExistsError):
+            os.symlink(SYSTEMD_FSCK_ROOT_PATH, SYSTEMD_FSCK_ROOT_ENABLE_PATH)
+        enable_plymouth()
+
+        # note that the saved real fsck can still exists in case of a reboot (as there was no tearDown as we wanted)
+        if not os.path.isfile(SAVED_FSCK_PATH):
+            os.rename(SYSTEM_FSCK_PATH, SAVED_FSCK_PATH)
+
+        # install mock fsck and killer
+        self.install_bin(os.path.join(os.path.dirname(os.path.realpath(__file__)), 'fsck'),
+                         SYSTEM_FSCK_PATH)
+        self.install_bin(os.path.join(os.path.dirname(os.path.realpath(__file__)), 'process-killer'),
+                         PROCESS_KILLER_PATH)
+
+        self.files_to_clean = [SYSTEMD_FSCK_ROOT_ENABLE_PATH, SYSTEM_FSCK_PATH, SYSTEMD_PROCESS_KILLER_PATH, PROCESS_KILLER_PATH]
+
+    def tearDown(self):
+        # tearDown is only called once the test really ended (not while rebooting during tests)
+        for f in self.files_to_clean:
+            with suppress(FileNotFoundError):
+                os.remove(f)
+        os.rename(SAVED_FSCK_PATH, SYSTEM_FSCK_PATH)
+        super().tearDown()
+
+    def test_fsckd_run(self):
+        '''Ensure we can reboot after a fsck was processed'''
+        if not self._after_reboot:
+            self.reboot()
+        else:
+            self.assertFsckdStop()
+            self.assertFsckProceeded()
+            self.assertSystemRunning()
+
+    def test_fsckd_run_without_plymouth(self):
+        '''Ensure we can reboot without plymouth after a fsck was processed'''
+        if not self._after_reboot:
+            enable_plymouth(enable=False)
+            self.reboot()
+        else:
+            self.assertFsckdStop()
+            self.assertFsckProceeded(with_plymouth=False)
+            self.assertSystemRunning()
+
+    def test_fsck_with_failure(self):
+        '''Ensure that a failing fsck doesn't prevent fsckd to stop'''
+        if not self._after_reboot:
+            self.install_process_killer_unit('fsck')
+            self.reboot()
+        else:
+            self.assertFsckdStop()
+            self.assertWasRunning('process-killer')
+            self.assertFalse(self.is_failed_unit('process-killer'))
+            self.assertFsckProceeded()
+            self.assertSystemRunning()
+
+    def test_systemd_fsck_with_failure(self):
+        '''Ensure that a failing systemd-fsck doesn't prevent fsckd to stop'''
+        if not self._after_reboot:
+            self.install_process_killer_unit('systemd-fsck', kill=True)
+            self.reboot()
+        else:
+            self.assertFsckdStop()
+            self.assertProcessKilled()
+            self.assertTrue(self.is_failed_unit('systemd-fsck-root'))
+            self.assertWasRunning('systemd-fsckd')
+            self.assertWasRunning('plymouth-start')
+            self.assertSystemRunning()
+
+    def test_systemd_fsckd_with_failure(self):
+        '''Ensure that a failing systemd-fsckd doesn't prevent system to boot'''
+        if not self._after_reboot:
+            self.install_process_killer_unit('systemd-fsckd', kill=True)
+            self.reboot()
+        else:
+            self.assertFsckdStop()
+            self.assertProcessKilled()
+            self.assertFalse(self.is_failed_unit('systemd-fsck-root'))
+            self.assertTrue(self.is_failed_unit('systemd-fsckd'))
+            self.assertWasRunning('plymouth-start')
+            self.assertSystemRunning()
+
+    def test_systemd_fsck_with_plymouth_failure(self):
+        '''Ensure that a failing plymouth doesn't prevent fsckd to reconnect/exit'''
+        if not self._after_reboot:
+            self.install_process_killer_unit('plymouthd', kill=True)
+            self.reboot()
+        else:
+            self.assertFsckdStop()
+            self.assertWasRunning('process-killer')
+            self.assertFsckProceeded()
+            self.assertFalse(self.is_active_unit('plymouth-start'))
+            self.assertSystemRunning()
+
+    def install_bin(self, source, dest):
+        '''install mock fsck'''
+        shutil.copy2(source, dest)
+        st = os.stat(dest)
+        os.chmod(dest, st.st_mode | stat.S_IEXEC)
+
+    def is_active_unit(self, unit):
+        '''Check that given unit is active'''
+
+        return subprocess.call(['systemctl', 'status', unit],
+                               stdout=subprocess.PIPE) == 0
+
+    def is_failed_unit(self, unit):
+        '''Check that given unit failed'''
+
+        p = subprocess.Popen(['systemctl', 'is-active', unit], stdout=subprocess.PIPE)
+        out, err = p.communicate()
+        if b'failed' in out:
+            return True
+        return False
+
+    def assertWasRunning(self, unit, expect_running=True):
+        '''Assert that a given unit has been running'''
+        p = subprocess.Popen(['systemctl', 'status', '--no-pager', unit],
+                             stdout=subprocess.PIPE, universal_newlines=True)
+        out = p.communicate()[0].strip()
+        if expect_running:
+            self.assertRegex(out, 'Active:.*since')
+        else:
+            self.assertNotRegex(out, 'Active:.*since')
+        self.assertIn(p.returncode, (0, 3))
+
+    def assertFsckdStop(self):
+        '''Ensure systemd-fsckd stops, which indicates no more fsck activity'''
+        timeout = time() + FSCKD_TIMEOUT
+        while time() < timeout:
+            if not self.is_active_unit('systemd-fsckd'):
+                return
+            sleep(1)
+        raise Exception("systemd-fsckd still active after {}s".format(FSCKD_TIMEOUT))
+
+    def assertFsckProceeded(self, with_plymouth=True):
+        '''Assert we executed most of the fsck-related services successfully'''
+        self.assertWasRunning('systemd-fsckd')
+        self.assertFalse(self.is_failed_unit('systemd-fsckd'))
+        self.assertTrue(self.is_active_unit('systemd-fsck-root'))  # remains active after exit
+        if with_plymouth:
+            self.assertWasRunning('plymouth-start')
+        else:
+            self.assertWasRunning('plymouth-start', expect_running=False)
+
+    def assertSystemRunning(self):
+        '''Assert that the system is running'''
+
+        self.assertTrue(self.is_active_unit('default.target'))
+
+    def assertProcessKilled(self):
+        '''Assert the targeted process was killed successfully'''
+        self.assertWasRunning('process-killer')
+        self.assertFalse(self.is_failed_unit('process-killer'))
+
+    def reboot(self):
+        '''Reboot the system with the current test marker'''
+        subprocess.check_call(['/tmp/autopkgtest-reboot', "{}:{}".format(self._test_name, self._return_code)])
+
+    def install_process_killer_unit(self, process_name, kill=False):
+        '''Create a systemd unit which will kill process_name'''
+        with open(SYSTEMD_PROCESS_KILLER_PATH, 'w') as f:
+            f.write('''[Unit]
+DefaultDependencies=no
+
+[Service]
+Type=simple
+ExecStart=/usr/bin/timeout 10 {} {}
+
+[Install]
+WantedBy=systemd-fsck-root.service'''.format(PROCESS_KILLER_PATH,
+                                             '--signal SIGKILL {}'.format(process_name) if kill else process_name))
+        subprocess.check_call(['systemctl', 'daemon-reload'])
+        subprocess.check_call(['systemctl', 'enable', 'process-killer'], stderr=subprocess.DEVNULL)
+
+
+def enable_plymouth(enable=True):
+    '''ensure plymouth is enabled in grub config (doesn't reboot)'''
+    plymouth_enabled = 'splash' in open('/boot/grub/grub.cfg').read()
+    if enable and not plymouth_enabled:
+        if os.path.exists(GRUB_AUTOPKGTEST_CONFIG_PATH):
+            shutil.copy2(GRUB_AUTOPKGTEST_CONFIG_PATH, TEST_AUTOPKGTEST_CONFIG_PATH)
+            for line in fileinput.input([TEST_AUTOPKGTEST_CONFIG_PATH], inplace=True):
+                if line.startswith("GRUB_CMDLINE_LINUX_DEFAULT"):
+                    print(line[:line.rfind('"')] + ' splash quiet"\n')
+        else:
+            os.makedirs(os.path.dirname(TEST_AUTOPKGTEST_CONFIG_PATH), exist_ok=True)
+            with open(TEST_AUTOPKGTEST_CONFIG_PATH, 'w') as f:
+                f.write('GRUB_CMDLINE_LINUX_DEFAULT="console=ttyS0 splash quiet"\n')
+    elif not enable and plymouth_enabled:
+        with suppress(FileNotFoundError):
+            os.remove(TEST_AUTOPKGTEST_CONFIG_PATH)
+    subprocess.check_call(['update-grub'], stderr=subprocess.DEVNULL)
+
+
+def boot_with_systemd_distro():
+    '''Reboot with systemd as init and distro setup for grub'''
+    enable_plymouth()
+    subprocess.check_call(['/tmp/autopkgtest-reboot', 'systemd-started'])
+
+
+def getAllTests(unitTestClass):
+    '''get all test names in predictable sorted order from unitTestClass'''
+    return sorted([test[0] for test in inspect.getmembers(unitTestClass, predicate=inspect.isfunction)
+                  if test[0].startswith('test_')])
+
+
+# AUTOPKGTEST_REBOOT_MARK contains the test name to pursue after reboot
+# (to check results and states after reboot, mostly).
+# we append the previous global return code (0 or 1) to it.
+# Example: AUTOPKGTEST_REBOOT_MARK=test_foo:0
+if __name__ == '__main__':
+    if os.path.exists('/run/initramfs/fsck-root'):
+        print('SKIP: root file system is being checked by initramfs already')
+        sys.exit(0)
+
+    all_tests = getAllTests(FsckdTest)
+    reboot_marker = os.getenv('AUTOPKGTEST_REBOOT_MARK')
+
+    current_test_after_reboot = ""
+    if not reboot_marker:
+        boot_with_systemd_distro()
+
+    # first test
+    if reboot_marker == "systemd-started":
+        current_test = all_tests[0]
+        return_code = 0
+    else:
+        (current_test_after_reboot, return_code) = reboot_marker.split(':')
+        current_test = current_test_after_reboot
+        return_code = int(return_code)
+
+    # loop on remaining tests to run
+    try:
+        remaining_tests = all_tests[all_tests.index(current_test):]
+    except ValueError:
+        print("Invalid value for AUTOPKGTEST_REBOOT_MARK, {} is not a valid test name".format(reboot_marker))
+        sys.exit(2)
+
+    # run all remaining tests
+    for test_name in remaining_tests:
+        after_reboot = False
+        # if this tests needed a reboot (and it has been performed), executes second part of it
+        if test_name == current_test_after_reboot:
+            after_reboot = True
+        suite = unittest.TestSuite()
+        suite.addTest(FsckdTest(test_name, after_reboot, return_code))
+        result = unittest.TextTestRunner(stream=sys.stdout, verbosity=2).run(suite)
+        if len(result.failures) != 0 or len(result.errors) != 0:
+            return_code = 1
+
+    sys.exit(return_code)