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