summaryrefslogtreecommitdiffstats
path: root/test/test-exec-deserialization.py
diff options
context:
space:
mode:
authorDaniel Baumann <daniel.baumann@progress-linux.org>2024-04-10 20:49:52 +0000
committerDaniel Baumann <daniel.baumann@progress-linux.org>2024-04-10 20:49:52 +0000
commit55944e5e40b1be2afc4855d8d2baf4b73d1876b5 (patch)
tree33f869f55a1b149e9b7c2b7e201867ca5dd52992 /test/test-exec-deserialization.py
parentInitial commit. (diff)
downloadsystemd-55944e5e40b1be2afc4855d8d2baf4b73d1876b5.tar.xz
systemd-55944e5e40b1be2afc4855d8d2baf4b73d1876b5.zip
Adding upstream version 255.4.upstream/255.4
Signed-off-by: Daniel Baumann <daniel.baumann@progress-linux.org>
Diffstat (limited to 'test/test-exec-deserialization.py')
-rwxr-xr-xtest/test-exec-deserialization.py214
1 files changed, 214 insertions, 0 deletions
diff --git a/test/test-exec-deserialization.py b/test/test-exec-deserialization.py
new file mode 100755
index 0000000..f8f3a6d
--- /dev/null
+++ b/test/test-exec-deserialization.py
@@ -0,0 +1,214 @@
+#!/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 os
+import subprocess
+import sys
+import time
+import unittest
+import uuid
+from enum import Enum
+
+class InstallChange(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 = f'/run/systemd/system/{self.unit}'
+ self.output_file = f"/tmp/test-issue-518-{uuid.uuid4()}"
+ self.unit_files = {}
+
+ unit_file_content = f'''
+ [Service]
+ Type=oneshot
+ ExecStart=/bin/sleep 3
+ ExecStart=/bin/bash -c "echo foo >>{self.output_file}"
+ '''
+ self.unit_files[InstallChange.NO_CHANGE] = unit_file_content
+
+ unit_file_content = f'''
+ [Service]
+ Type=oneshot
+ ExecStart=/bin/bash -c "echo foo >>{self.output_file}"
+ ExecStart=/bin/sleep 3
+ '''
+ self.unit_files[InstallChange.LINES_SWAPPED] = unit_file_content
+
+ unit_file_content = f'''
+ [Service]
+ Type=oneshot
+ ExecStart=/bin/bash -c "echo bar >>{self.output_file}"
+ ExecStart=/bin/sleep 3
+ ExecStart=/bin/bash -c "echo foo >>{self.output_file}"
+ '''
+ self.unit_files[InstallChange.COMMAND_ADDED_BEFORE] = unit_file_content
+
+ unit_file_content = f'''
+ [Service]
+ Type=oneshot
+ ExecStart=/bin/sleep 3
+ ExecStart=/bin/bash -c "echo foo >>{self.output_file}"
+ ExecStart=/bin/bash -c "echo bar >>{self.output_file}"
+ '''
+ self.unit_files[InstallChange.COMMAND_ADDED_AFTER] = unit_file_content
+
+ unit_file_content = f'''
+ [Service]
+ Type=oneshot
+ ExecStart=/bin/bash -c "echo baz >>{self.output_file}"
+ ExecStart=/bin/sleep 3
+ ExecStart=/bin/bash -c "echo foo >>{self.output_file}"
+ ExecStart=/bin/bash -c "echo bar >>{self.output_file}"
+ '''
+ self.unit_files[InstallChange.COMMAND_INTERLEAVED] = unit_file_content
+
+ unit_file_content = f'''
+ [Service]
+ Type=oneshot
+ ExecStart=/bin/bash -c "echo bar >>{self.output_file}"
+ ExecStart=/bin/bash -c "echo baz >>{self.output_file}"
+ '''
+ self.unit_files[InstallChange.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, InstallChange):
+ raise ValueError('Unknown unit file change')
+
+ content = self.unit_files[unit_file_change]
+
+ with open(self.unitfile_path, 'w', encoding='utf-8') as f:
+ f.write(content)
+
+ self.reload()
+
+ def check_output(self, expected_output):
+ for _ in range(15):
+ # Wait until the unit finishes so we don't check an incomplete log
+ if subprocess.call(['systemctl', '-q', 'is-active', self.unit]) == 0:
+ continue
+
+ os.sync()
+
+ try:
+ with open(self.output_file, 'r', encoding='utf-8') as log:
+ output = log.read()
+ self.assertEqual(output, expected_output)
+ return
+ except IOError:
+ pass
+
+ time.sleep(1)
+
+ self.fail(f'Timed out while waiting for the output file {self.output_file} to appear')
+
+ def setup_unit(self):
+ self.write_unit_file(InstallChange.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()
+
+ self.check_output(expected_output)
+
+ def test_swapped(self):
+ self.setup_unit()
+ self.write_unit_file(InstallChange.LINES_SWAPPED)
+ self.reload()
+
+ 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(InstallChange.COMMAND_ADDED_BEFORE)
+ self.reload()
+
+ self.check_output(expected_output)
+
+ def test_added_after(self):
+ expected_output = 'foo\nbar\n'
+
+ self.setup_unit()
+ self.write_unit_file(InstallChange.COMMAND_ADDED_AFTER)
+ self.reload()
+
+ self.check_output(expected_output)
+
+ def test_interleaved(self):
+ expected_output = 'foo\nbar\n'
+
+ self.setup_unit()
+ self.write_unit_file(InstallChange.COMMAND_INTERLEAVED)
+ self.reload()
+
+ self.check_output(expected_output)
+
+ def test_removal(self):
+ self.setup_unit()
+ self.write_unit_file(InstallChange.REMOVAL)
+ self.reload()
+
+ self.assertTrue(not os.path.exists(self.output_file))
+
+ def test_issue_6533(self):
+ unit = "test-issue-6533.service"
+ unitfile_path = f"/run/systemd/system/{unit}"
+
+ content = '''
+ [Service]
+ ExecStart=/bin/sleep 5
+ '''
+
+ with open(unitfile_path, 'w', encoding='utf-8') 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', encoding='utf-8') as f:
+ f.write(content)
+
+ self.reload()
+ time.sleep(5)
+
+ self.assertNotEqual(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))