diff --git a/test/test-exec-deserialization.py b/test/test-exec-deserialization.py index 9491e4081e..46627add91 100755 --- a/test/test-exec-deserialization.py +++ b/test/test-exec-deserialization.py @@ -1,20 +1,23 @@ #!/usr/bin/env python3 # SPDX-License-Identifier: LGPL-2.1-or-later +# pylint: disable=line-too-long,too-many-lines,too-many-branches,too-many-statements,too-many-arguments +# pylint: disable=too-many-public-methods,too-many-boolean-expressions,invalid-name,no-self-use +# pylint: disable=missing-function-docstring,missing-class-docstring,missing-module-docstring # # Copyright © 2017 Michal Sekletar # 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 - +import tempfile +import time +import unittest from enum import Enum + class UnitFileChange(Enum): NO_CHANGE = 0 LINES_SWAPPED = 1 @@ -26,59 +29,59 @@ class UnitFileChange(Enum): class ExecutionResumeTest(unittest.TestCase): def setUp(self): self.unit = 'test-issue-518.service' - self.unitfile_path = '/run/systemd/system/{0}'.format(self.unit) + self.unitfile_path = f'/run/systemd/system/{self.unit}' self.output_file = tempfile.mktemp() self.unit_files = {} - unit_file_content = ''' + unit_file_content = f''' [Service] Type=oneshot ExecStart=/bin/sleep 3 - ExecStart=/bin/bash -c "echo foo >> {0}" - '''.format(self.output_file) + ExecStart=/bin/bash -c "echo foo >> {self.output_file}" + ''' self.unit_files[UnitFileChange.NO_CHANGE] = unit_file_content - unit_file_content = ''' + unit_file_content = f''' [Service] Type=oneshot - ExecStart=/bin/bash -c "echo foo >> {0}" + ExecStart=/bin/bash -c "echo foo >> {self.output_file}" ExecStart=/bin/sleep 3 - '''.format(self.output_file) + ''' self.unit_files[UnitFileChange.LINES_SWAPPED] = unit_file_content - unit_file_content = ''' + unit_file_content = f''' [Service] Type=oneshot - ExecStart=/bin/bash -c "echo bar >> {0}" + ExecStart=/bin/bash -c "echo bar >> {self.output_file}" ExecStart=/bin/sleep 3 - ExecStart=/bin/bash -c "echo foo >> {0}" - '''.format(self.output_file) + ExecStart=/bin/bash -c "echo foo >> {self.output_file}" + ''' self.unit_files[UnitFileChange.COMMAND_ADDED_BEFORE] = unit_file_content - unit_file_content = ''' + unit_file_content = f''' [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) + ExecStart=/bin/bash -c "echo foo >> {self.output_file}" + ExecStart=/bin/bash -c "echo bar >> {self.output_file}" + ''' self.unit_files[UnitFileChange.COMMAND_ADDED_AFTER] = unit_file_content - unit_file_content = ''' + unit_file_content = f''' [Service] Type=oneshot - ExecStart=/bin/bash -c "echo baz >> {0}" + ExecStart=/bin/bash -c "echo baz >> {self.output_file}" ExecStart=/bin/sleep 3 - ExecStart=/bin/bash -c "echo foo >> {0}" - ExecStart=/bin/bash -c "echo bar >> {0}" - '''.format(self.output_file) + ExecStart=/bin/bash -c "echo foo >> {self.output_file}" + ExecStart=/bin/bash -c "echo bar >> {self.output_file}" + ''' self.unit_files[UnitFileChange.COMMAND_INTERLEAVED] = unit_file_content - unit_file_content = ''' + unit_file_content = f''' [Service] Type=oneshot - ExecStart=/bin/bash -c "echo bar >> {0}" - ExecStart=/bin/bash -c "echo baz >> {0}" + ExecStart=/bin/bash -c "echo bar >> {self.output_file}" + ExecStart=/bin/bash -c "echo baz >> {self.output_file}" '''.format(self.output_file) self.unit_files[UnitFileChange.REMOVAL] = unit_file_content @@ -91,19 +94,24 @@ class ExecutionResumeTest(unittest.TestCase): content = self.unit_files[unit_file_change] - with open(self.unitfile_path, 'w') as f: + with open(self.unitfile_path, 'w', encoding='utf-8') 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() + for _ in range(15): + try: + with open(self.output_file, 'r', encoding='utf-8') as log: + output = log.read() + self.assertEqual(output, expected_output) + return + except IOError: + pass - self.assertEqual(output, expected_output) + 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(UnitFileChange.NO_CHANGE) @@ -115,17 +123,13 @@ class ExecutionResumeTest(unittest.TestCase): 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)) @@ -135,7 +139,6 @@ class ExecutionResumeTest(unittest.TestCase): self.setup_unit() self.write_unit_file(UnitFileChange.COMMAND_ADDED_BEFORE) self.reload() - time.sleep(4) self.check_output(expected_output) @@ -145,7 +148,6 @@ class ExecutionResumeTest(unittest.TestCase): self.setup_unit() self.write_unit_file(UnitFileChange.COMMAND_ADDED_AFTER) self.reload() - time.sleep(4) self.check_output(expected_output) @@ -155,7 +157,6 @@ class ExecutionResumeTest(unittest.TestCase): self.setup_unit() self.write_unit_file(UnitFileChange.COMMAND_INTERLEAVED) self.reload() - time.sleep(4) self.check_output(expected_output) @@ -163,20 +164,19 @@ class ExecutionResumeTest(unittest.TestCase): 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) + unitfile_path = f"/run/systemd/system/{unit}" content = ''' [Service] ExecStart=/bin/sleep 5 ''' - with open(unitfile_path, 'w') as f: + with open(unitfile_path, 'w', encoding='utf-8') as f: f.write(content) self.reload() @@ -190,13 +190,13 @@ class ExecutionResumeTest(unittest.TestCase): ExecStart=/bin/true ''' - with open(unitfile_path, 'w') as f: + with open(unitfile_path, 'w', encoding='utf-8') 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) + 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]: