test/test-exec-deserialization.py

   1 #!/usr/bin/env python3
   2 # SPDX-License-Identifier: LGPL-2.1-or-later
   3 # pylint: disable=line-too-long,too-many-lines,too-many-branches,too-many-statements,too-many-arguments
   4 # pylint: disable=too-many-public-methods,too-many-boolean-expressions,invalid-name,no-self-use
   5 # pylint: disable=missing-function-docstring,missing-class-docstring,missing-module-docstring
   6 #
   7 #  Copyright © 2017 Michal Sekletar <msekleta@redhat.com>
   8
   9 # ATTENTION: This uses the *installed* systemd, not the one from the built
  10 # source tree.
  11
  12 import os
  13 import subprocess
  14 import sys
  15 import time
  16 import unittest
  17 import uuid
  18 from enum import Enum
  19
  20 class InstallChange(Enum):
  21     NO_CHANGE = 0
  22     LINES_SWAPPED = 1
  23     COMMAND_ADDED_BEFORE = 2
  24     COMMAND_ADDED_AFTER = 3
  25     COMMAND_INTERLEAVED = 4
  26     REMOVAL = 5
  27
  28 class ExecutionResumeTest(unittest.TestCase):
  29     def setUp(self):
  30         self.unit = 'test-issue-518.service'
  31         self.unitfile_path = f'/run/systemd/system/{self.unit}'
  32         self.output_file = f"/tmp/test-issue-518-{uuid.uuid4()}"
  33         self.unit_files = {}
  34
  35         unit_file_content = f'''
  36         [Service]
  37         Type=oneshot
  38         ExecStart=/bin/sleep 3
  39         ExecStart=/bin/bash -c "echo foo >>{self.output_file}"
  40         '''
  41         self.unit_files[InstallChange.NO_CHANGE] = unit_file_content
  42
  43         unit_file_content = f'''
  44         [Service]
  45         Type=oneshot
  46         ExecStart=/bin/bash -c "echo foo >>{self.output_file}"
  47         ExecStart=/bin/sleep 3
  48         '''
  49         self.unit_files[InstallChange.LINES_SWAPPED] = unit_file_content
  50
  51         unit_file_content = f'''
  52         [Service]
  53         Type=oneshot
  54         ExecStart=/bin/bash -c "echo bar >>{self.output_file}"
  55         ExecStart=/bin/sleep 3
  56         ExecStart=/bin/bash -c "echo foo >>{self.output_file}"
  57         '''
  58         self.unit_files[InstallChange.COMMAND_ADDED_BEFORE] = unit_file_content
  59
  60         unit_file_content = f'''
  61         [Service]
  62         Type=oneshot
  63         ExecStart=/bin/sleep 3
  64         ExecStart=/bin/bash -c "echo foo >>{self.output_file}"
  65         ExecStart=/bin/bash -c "echo bar >>{self.output_file}"
  66         '''
  67         self.unit_files[InstallChange.COMMAND_ADDED_AFTER] = unit_file_content
  68
  69         unit_file_content = f'''
  70         [Service]
  71         Type=oneshot
  72         ExecStart=/bin/bash -c "echo baz >>{self.output_file}"
  73         ExecStart=/bin/sleep 3
  74         ExecStart=/bin/bash -c "echo foo >>{self.output_file}"
  75         ExecStart=/bin/bash -c "echo bar >>{self.output_file}"
  76         '''
  77         self.unit_files[InstallChange.COMMAND_INTERLEAVED] = unit_file_content
  78
  79         unit_file_content = f'''
  80         [Service]
  81         Type=oneshot
  82         ExecStart=/bin/bash -c "echo bar >>{self.output_file}"
  83         ExecStart=/bin/bash -c "echo baz >>{self.output_file}"
  84         '''
  85         self.unit_files[InstallChange.REMOVAL] = unit_file_content
  86
  87     def reload(self):
  88         subprocess.check_call(['systemctl', 'daemon-reload'])
  89
  90     def write_unit_file(self, unit_file_change):
  91         if not isinstance(unit_file_change, InstallChange):
  92             raise ValueError('Unknown unit file change')
  93
  94         content = self.unit_files[unit_file_change]
  95
  96         with open(self.unitfile_path, 'w', encoding='utf-8') as f:
  97             f.write(content)
  98
  99         self.reload()
 100
 101     def check_output(self, expected_output):
 102         for _ in range(15):
 103             # Wait until the unit finishes so we don't check an incomplete log
 104             if subprocess.call(['systemctl', '-q', 'is-active', self.unit]) == 0:
 105                 continue
 106
 107             os.sync()
 108
 109             try:
 110                 with open(self.output_file, 'r', encoding='utf-8') as log:
 111                     output = log.read()
 112                     self.assertEqual(output, expected_output)
 113                     return
 114             except IOError:
 115                 pass
 116
 117             time.sleep(1)
 118
 119         self.fail(f'Timed out while waiting for the output file {self.output_file} to appear')
 120
 121     def setup_unit(self):
 122         self.write_unit_file(InstallChange.NO_CHANGE)
 123         subprocess.check_call(['systemctl', '--job-mode=replace', '--no-block', 'start', self.unit])
 124         time.sleep(1)
 125
 126     def test_no_change(self):
 127         expected_output = 'foo\n'
 128
 129         self.setup_unit()
 130         self.reload()
 131
 132         self.check_output(expected_output)
 133
 134     def test_swapped(self):
 135         self.setup_unit()
 136         self.write_unit_file(InstallChange.LINES_SWAPPED)
 137         self.reload()
 138
 139         self.assertTrue(not os.path.exists(self.output_file))
 140
 141     def test_added_before(self):
 142         expected_output = 'foo\n'
 143
 144         self.setup_unit()
 145         self.write_unit_file(InstallChange.COMMAND_ADDED_BEFORE)
 146         self.reload()
 147
 148         self.check_output(expected_output)
 149
 150     def test_added_after(self):
 151         expected_output = 'foo\nbar\n'
 152
 153         self.setup_unit()
 154         self.write_unit_file(InstallChange.COMMAND_ADDED_AFTER)
 155         self.reload()
 156
 157         self.check_output(expected_output)
 158
 159     def test_interleaved(self):
 160         expected_output = 'foo\nbar\n'
 161
 162         self.setup_unit()
 163         self.write_unit_file(InstallChange.COMMAND_INTERLEAVED)
 164         self.reload()
 165
 166         self.check_output(expected_output)
 167
 168     def test_removal(self):
 169         self.setup_unit()
 170         self.write_unit_file(InstallChange.REMOVAL)
 171         self.reload()
 172
 173         self.assertTrue(not os.path.exists(self.output_file))
 174
 175     def test_issue_6533(self):
 176         unit = "test-issue-6533.service"
 177         unitfile_path = f"/run/systemd/system/{unit}"
 178
 179         content = '''
 180         [Service]
 181         ExecStart=/bin/sleep 5
 182         '''
 183
 184         with open(unitfile_path, 'w', encoding='utf-8') as f:
 185             f.write(content)
 186
 187         self.reload()
 188
 189         subprocess.check_call(['systemctl', '--job-mode=replace', '--no-block', 'start', unit])
 190         time.sleep(2)
 191
 192         content = '''
 193         [Service]
 194         ExecStart=/bin/sleep 5
 195         ExecStart=/bin/true
 196         '''
 197
 198         with open(unitfile_path, 'w', encoding='utf-8') as f:
 199             f.write(content)
 200
 201         self.reload()
 202         time.sleep(5)
 203
 204         self.assertNotEqual(subprocess.call("journalctl -b _PID=1 | grep -q 'Freezing execution'", shell=True), 0)
 205
 206     def tearDown(self):
 207         for f in [self.output_file, self.unitfile_path]:
 208             try:
 209                 os.remove(f)
 210             except OSError:
 211                 # ignore error if log file doesn't exist
 212                 pass
 213
 214         self.reload()
 215
 216 if __name__ == '__main__':
 217     unittest.main(testRunner=unittest.TextTestRunner(stream=sys.stdout, verbosity=3))