Adding upstream version 247.3.upstream/247.3upstream

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

1 files changed, 212 insertions, 0 deletions

diff --git a/test/test-exec-deserialization.py b/test/test-exec-deserialization.py
new file mode 100755
index 0000000..9491e40
--- /dev/null
+++ b/test/test-exec-deserialization.py
@@ -0,0 +1,212 @@
+#!/usr/bin/env python3
+# SPDX-License-Identifier: LGPL-2.1-or-later
+#
+#  Copyright © 2017 Michal Sekletar <msekleta@redhat.com>
+
+# ATTENTION: This uses the *installed* systemd, not the one from the built
+# source tree.
+
+import unittest
+import time
+import os
+import tempfile
+import subprocess
+import sys
+
+from enum import Enum
+
+class UnitFileChange(Enum):
+    NO_CHANGE = 0
+    LINES_SWAPPED = 1
+    COMMAND_ADDED_BEFORE = 2
+    COMMAND_ADDED_AFTER = 3
+    COMMAND_INTERLEAVED = 4
+    REMOVAL = 5
+
+class ExecutionResumeTest(unittest.TestCase):
+    def setUp(self):
+        self.unit = 'test-issue-518.service'
+        self.unitfile_path = '/run/systemd/system/{0}'.format(self.unit)
+        self.output_file = tempfile.mktemp()
+        self.unit_files = {}
+
+        unit_file_content = '''
+        [Service]
+        Type=oneshot
+        ExecStart=/bin/sleep 3
+        ExecStart=/bin/bash -c "echo foo >> {0}"
+        '''.format(self.output_file)
+        self.unit_files[UnitFileChange.NO_CHANGE] = unit_file_content
+
+        unit_file_content = '''
+        [Service]
+        Type=oneshot
+        ExecStart=/bin/bash -c "echo foo >> {0}"
+        ExecStart=/bin/sleep 3
+        '''.format(self.output_file)
+        self.unit_files[UnitFileChange.LINES_SWAPPED] = unit_file_content
+
+        unit_file_content = '''
+        [Service]
+        Type=oneshot
+        ExecStart=/bin/bash -c "echo bar >> {0}"
+        ExecStart=/bin/sleep 3
+        ExecStart=/bin/bash -c "echo foo >> {0}"
+        '''.format(self.output_file)
+        self.unit_files[UnitFileChange.COMMAND_ADDED_BEFORE] = unit_file_content
+
+        unit_file_content = '''
+        [Service]
+        Type=oneshot
+        ExecStart=/bin/sleep 3
+        ExecStart=/bin/bash -c "echo foo >> {0}"
+        ExecStart=/bin/bash -c "echo bar >> {0}"
+        '''.format(self.output_file)
+        self.unit_files[UnitFileChange.COMMAND_ADDED_AFTER] = unit_file_content
+
+        unit_file_content = '''
+        [Service]
+        Type=oneshot
+        ExecStart=/bin/bash -c "echo baz >> {0}"
+        ExecStart=/bin/sleep 3
+        ExecStart=/bin/bash -c "echo foo >> {0}"
+        ExecStart=/bin/bash -c "echo bar >> {0}"
+        '''.format(self.output_file)
+        self.unit_files[UnitFileChange.COMMAND_INTERLEAVED] = unit_file_content
+
+        unit_file_content = '''
+        [Service]
+        Type=oneshot
+        ExecStart=/bin/bash -c "echo bar >> {0}"
+        ExecStart=/bin/bash -c "echo baz >> {0}"
+        '''.format(self.output_file)
+        self.unit_files[UnitFileChange.REMOVAL] = unit_file_content
+
+    def reload(self):
+        subprocess.check_call(['systemctl', 'daemon-reload'])
+
+    def write_unit_file(self, unit_file_change):
+        if not isinstance(unit_file_change, UnitFileChange):
+            raise ValueError('Unknown unit file change')
+
+        content = self.unit_files[unit_file_change]
+
+        with open(self.unitfile_path, 'w') as f:
+            f.write(content)
+
+        self.reload()
+
+    def check_output(self, expected_output):
+        try:
+            with open(self.output_file, 'r') as log:
+                output = log.read()
+        except IOError:
+            self.fail()
+
+        self.assertEqual(output, expected_output)
+
+    def setup_unit(self):
+        self.write_unit_file(UnitFileChange.NO_CHANGE)
+        subprocess.check_call(['systemctl', '--job-mode=replace', '--no-block', 'start', self.unit])
+        time.sleep(1)
+
+    def test_no_change(self):
+        expected_output = 'foo\n'
+
+        self.setup_unit()
+        self.reload()
+        time.sleep(4)
+
+        self.check_output(expected_output)
+
+    def test_swapped(self):
+        expected_output = ''
+
+        self.setup_unit()
+        self.write_unit_file(UnitFileChange.LINES_SWAPPED)
+        self.reload()
+        time.sleep(4)
+
+        self.assertTrue(not os.path.exists(self.output_file))
+
+    def test_added_before(self):
+        expected_output = 'foo\n'
+
+        self.setup_unit()
+        self.write_unit_file(UnitFileChange.COMMAND_ADDED_BEFORE)
+        self.reload()
+        time.sleep(4)
+
+        self.check_output(expected_output)
+
+    def test_added_after(self):
+        expected_output = 'foo\nbar\n'
+
+        self.setup_unit()
+        self.write_unit_file(UnitFileChange.COMMAND_ADDED_AFTER)
+        self.reload()
+        time.sleep(4)
+
+        self.check_output(expected_output)
+
+    def test_interleaved(self):
+        expected_output = 'foo\nbar\n'
+
+        self.setup_unit()
+        self.write_unit_file(UnitFileChange.COMMAND_INTERLEAVED)
+        self.reload()
+        time.sleep(4)
+
+        self.check_output(expected_output)
+
+    def test_removal(self):
+        self.setup_unit()
+        self.write_unit_file(UnitFileChange.REMOVAL)
+        self.reload()
+        time.sleep(4)
+
+        self.assertTrue(not os.path.exists(self.output_file))
+
+    def test_issue_6533(self):
+        unit = "test-issue-6533.service"
+        unitfile_path = "/run/systemd/system/{}".format(unit)
+
+        content = '''
+        [Service]
+        ExecStart=/bin/sleep 5
+        '''
+
+        with open(unitfile_path, 'w') as f:
+            f.write(content)
+
+        self.reload()
+
+        subprocess.check_call(['systemctl', '--job-mode=replace', '--no-block', 'start', unit])
+        time.sleep(2)
+
+        content = '''
+        [Service]
+        ExecStart=/bin/sleep 5
+        ExecStart=/bin/true
+        '''
+
+        with open(unitfile_path, 'w') as f:
+            f.write(content)
+
+        self.reload()
+        time.sleep(5)
+
+        self.assertTrue(subprocess.call("journalctl -b _PID=1  | grep -q 'Freezing execution'", shell=True) != 0)
+
+    def tearDown(self):
+        for f in [self.output_file, self.unitfile_path]:
+            try:
+                os.remove(f)
+            except OSError:
+                # ignore error if log file doesn't exist
+                pass
+
+        self.reload()
+
+if __name__ == '__main__':
+    unittest.main(testRunner=unittest.TextTestRunner(stream=sys.stdout, verbosity=3))