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