test/test-exec-deserialization.py

   1 #!/usr/bin/env python3
   2 # SPDX-License-Identifier: LGPL-2.1+
   3 #
   4 #  Copyright 2017 Michal Sekletar <msekleta@redhat.com>
   5 #
   6 #  systemd is free software; you can redistribute it and/or modify it
   7 #  under the terms of the GNU Lesser General Public License as published by
   8 #  the Free Software Foundation; either version 2.1 of the License, or
   9 #  (at your option) any later version.
  10 #
  11 #  systemd is distributed in the hope that it will be useful, but
  12 #  WITHOUT ANY WARRANTY; without even the implied warranty of
  13 #  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
  14 #  Lesser General Public License for more details.
  15 #
  16 #  You should have received a copy of the GNU Lesser General Public License
  17 #  along with systemd; If not, see <http://www.gnu.org/licenses/>.
  18
  19 # ATTENTION: This uses the *installed* systemd, not the one from the built
  20 # source tree.
  21
  22 import unittest
  23 import time
  24 import os
  25 import tempfile
  26 import subprocess
  27
  28 from enum import Enum
  29
  30 class UnitFileChange(Enum):
  31     NO_CHANGE = 0
  32     LINES_SWAPPED = 1
  33     COMMAND_ADDED_BEFORE = 2
  34     COMMAND_ADDED_AFTER = 3
  35     COMMAND_INTERLEAVED = 4
  36     REMOVAL = 5
  37
  38 class ExecutionResumeTest(unittest.TestCase):
  39     def setUp(self):
  40         self.unit = 'test-issue-518.service'
  41         self.unitfile_path = '/run/systemd/system/{0}'.format(self.unit)
  42         self.output_file = tempfile.mktemp()
  43         self.unit_files = {}
  44
  45         unit_file_content = '''
  46         [Service]
  47         Type=oneshot
  48         ExecStart=/bin/sleep 2
  49         ExecStart=/bin/bash -c "echo foo >> {0}"
  50         '''.format(self.output_file)
  51         self.unit_files[UnitFileChange.NO_CHANGE] = unit_file_content
  52
  53         unit_file_content = '''
  54         [Service]
  55         Type=oneshot
  56         ExecStart=/bin/bash -c "echo foo >> {0}"
  57         ExecStart=/bin/sleep 2
  58         '''.format(self.output_file)
  59         self.unit_files[UnitFileChange.LINES_SWAPPED] = unit_file_content
  60
  61         unit_file_content = '''
  62         [Service]
  63         Type=oneshot
  64         ExecStart=/bin/bash -c "echo bar >> {0}"
  65         ExecStart=/bin/sleep 2
  66         ExecStart=/bin/bash -c "echo foo >> {0}"
  67         '''.format(self.output_file)
  68         self.unit_files[UnitFileChange.COMMAND_ADDED_BEFORE] = unit_file_content
  69
  70         unit_file_content = '''
  71         [Service]
  72         Type=oneshot
  73         ExecStart=/bin/sleep 2
  74         ExecStart=/bin/bash -c "echo foo >> {0}"
  75         ExecStart=/bin/bash -c "echo bar >> {0}"
  76         '''.format(self.output_file)
  77         self.unit_files[UnitFileChange.COMMAND_ADDED_AFTER] = unit_file_content
  78
  79         unit_file_content = '''
  80         [Service]
  81         Type=oneshot
  82         ExecStart=/bin/bash -c "echo baz >> {0}"
  83         ExecStart=/bin/sleep 2
  84         ExecStart=/bin/bash -c "echo foo >> {0}"
  85         ExecStart=/bin/bash -c "echo bar >> {0}"
  86         '''.format(self.output_file)
  87         self.unit_files[UnitFileChange.COMMAND_INTERLEAVED] = unit_file_content
  88
  89         unit_file_content = '''
  90         [Service]
  91         Type=oneshot
  92         ExecStart=/bin/bash -c "echo bar >> {0}"
  93         ExecStart=/bin/bash -c "echo baz >> {0}"
  94         '''.format(self.output_file)
  95         self.unit_files[UnitFileChange.REMOVAL] = unit_file_content
  96
  97     def reload(self):
  98         subprocess.check_call(['systemctl', 'daemon-reload'])
  99
 100     def write_unit_file(self, unit_file_change):
 101         if not isinstance(unit_file_change, UnitFileChange):
 102             raise ValueError('Unknown unit file change')
 103
 104         content = self.unit_files[unit_file_change]
 105
 106         with open(self.unitfile_path, 'w') as f:
 107             f.write(content)
 108
 109         self.reload()
 110
 111     def check_output(self, expected_output):
 112         try:
 113             with open(self.output_file, 'r') as log:
 114                 output = log.read()
 115         except IOError:
 116             self.fail()
 117
 118         self.assertEqual(output, expected_output)
 119
 120     def setup_unit(self):
 121         self.write_unit_file(UnitFileChange.NO_CHANGE)
 122         subprocess.check_call(['systemctl', '--job-mode=replace', '--no-block', 'start', self.unit])
 123
 124     def test_no_change(self):
 125         expected_output = 'foo\n'
 126
 127         self.setup_unit()
 128         self.reload()
 129         time.sleep(4)
 130
 131         self.check_output(expected_output)
 132
 133     def test_swapped(self):
 134         expected_output = ''
 135
 136         self.setup_unit()
 137         self.write_unit_file(UnitFileChange.LINES_SWAPPED)
 138         self.reload()
 139         time.sleep(4)
 140
 141         self.assertTrue(not os.path.exists(self.output_file))
 142
 143     def test_added_before(self):
 144         expected_output = 'foo\n'
 145
 146         self.setup_unit()
 147         self.write_unit_file(UnitFileChange.COMMAND_ADDED_BEFORE)
 148         self.reload()
 149         time.sleep(4)
 150
 151         self.check_output(expected_output)
 152
 153     def test_added_after(self):
 154         expected_output = 'foo\nbar\n'
 155
 156         self.setup_unit()
 157         self.write_unit_file(UnitFileChange.COMMAND_ADDED_AFTER)
 158         self.reload()
 159         time.sleep(4)
 160
 161         self.check_output(expected_output)
 162
 163     def test_interleaved(self):
 164         expected_output = 'foo\nbar\n'
 165
 166         self.setup_unit()
 167         self.write_unit_file(UnitFileChange.COMMAND_INTERLEAVED)
 168         self.reload()
 169         time.sleep(4)
 170
 171         self.check_output(expected_output)
 172
 173     def test_removal(self):
 174         self.setup_unit()
 175         self.write_unit_file(UnitFileChange.REMOVAL)
 176         self.reload()
 177         time.sleep(4)
 178
 179         self.assertTrue(not os.path.exists(self.output_file))
 180
 181     def test_issue_6533(self):
 182         unit = "test-issue-6533.service"
 183         unitfile_path = "/run/systemd/system/{}".format(unit)
 184
 185         content = '''
 186         [Service]
 187         ExecStart=/bin/sleep 5
 188         '''
 189
 190         with open(unitfile_path, 'w') as f:
 191             f.write(content)
 192
 193         self.reload()
 194
 195         subprocess.check_call(['systemctl', '--job-mode=replace', '--no-block', 'start', unit])
 196         time.sleep(2)
 197
 198         content = '''
 199         [Service]
 200         ExecStart=/bin/sleep 5
 201         ExecStart=/bin/true
 202         '''
 203
 204         with open(unitfile_path, 'w') as f:
 205             f.write(content)
 206
 207         self.reload()
 208         time.sleep(5)
 209
 210         self.assertTrue(subprocess.call("journalctl -b _PID=1  | grep -q 'Freezing execution'", shell=True) != 0)
 211
 212     def tearDown(self):
 213         for f in [self.output_file, self.unitfile_path]:
 214             try:
 215                 os.remove(f)
 216             except OSError:
 217                 # ignore error if log file doesn't exist
 218                 pass
 219
 220         self.reload()
 221
 222 if __name__ == '__main__':
 223     unittest.main()