1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117 | #!/usr/bin/python3
# autopkgtest check: Ensure that systemd-fsckd can report progress and cancel
# (C) 2015 Canonical Ltd.
# Author: Didier Roche <didrocks@ubuntu.com>
from contextlib import suppress
import fileinput
import os
import subprocess
import shutil
import sys
import unittest
from time import sleep
SYSTEMD_SYSTEM_UNIT_DIR = "/lib/systemd/system"
PREFSCK_UNIT_PATH = os.path.join(SYSTEMD_SYSTEM_UNIT_DIR, 'pre-fsck.service')
MOCK_FSCK_PATH = os.path.join(os.path.dirname(os.path.realpath(__file__), '..', '..', 'test', 'mocks', 'fsck'))
FSCK_SYSTEM_PATH = '/sbin/fsck'
FSCKD_TIMEOUT = 30
class FsckdTest(unittest.TestCase):
'''Check that we run, report and can cancel fsck'''
def __init__(self, test_name, result_check, return_code):
super().__init__(test_name)
self._test_name = test_name
self._result_check = result_check
self._return_code = return_code
def setUp(self):
super().setUp()
self.files_to_clean = [PREFSCK_UNIT_PATH]
os.rename(FSCK_SYSTEM_PATH, "{}.real".format(FSCK_SYSTEM_PATH))
def tearDown(self):
for f in self.files_to_clean:
with suppress(FileNotFoundError):
os.remove(f)
# tearDown is only call once the test really ended (not on reboot during tests)
os.rename("{}.real".format(FSCK_SYSTEM_PATH), PREFSCK_UNIT_PATH)
shutil.copy2("{}.real".format(FSCK_SYSTEM_PATH), FSCK_SYSTEM_PATH)
super().tearDown()
def test_fsckd_run(self):
'''Ensure we can reboot after a fsck was processed'''
if not result_check:
self.install_mock_fsck()
self.reboot()
else:
sleep(FSCKD_TIMEOUT) # ensure we let fsckd exit
self.assertFalse(self.is_active_unit('systemd-fsck-root'))
self.assertFalse(self.is_active_unit('systemd-fsckd'))
def test_fsck_with_failure(self):
'''ensure that we can reboot after a failed fsck was processed'''
pass
def is_active_unit(self, unit):
'''Check that given unit is active'''
if subprocess.call(['systemctl', '-q', 'is-active', unit]) != 0:
return False
return True
def reboot(self):
subprocess.check_call(['autopkgtest-reboot', "{}:{}".format(self._test_name, self._return_code)])
def boot_with_systemd(marker):
'''Reboot with systemd as init
In case something else is currently running in the testbed
'''
if subprocess.call(['systemctl', 'status'], stdout=subprocess.PIPE,
stderr=subprocess.PIPE) != 0:
print('Installing systemd-sysv and rebooting...')
subprocess.check_call('apt-get -y install systemd-sysv 2>&1',
shell=True)
subprocess.check_call(['autopkgtest-reboot', marker])
def getAllTests(unitTestClass):
'''get all test names in predictable sorted order from unitTestClass'''
return sorted([test[0] for test in inspect.getmembers(unitTestClass, predicate=inspect.isfunction)
if test[0].startswith('test_')])
# ADT_REBOOT_MARK contains the test name to continue end in the result
# mode (to check states after reboot, mostly).
# this one is appended with the current global return code (0 or 1)
# Example: ADT_REBOOT_MARK=test_foo:0
if __name__ == '__main__':
all_tests = getAllTests(FsckdTest)
current_test_after_reboot = os.getenv('ADT_REBOOT_MARK')
if not current_test_after_reboot:
boot_with_systemd("{}:0".format(all_tests[0]))
(current_test, return_code) = current_test_after_reboot.split(':')
try:
remaining_tests = all_tests[all_tests.index(current_test_after_reboot):]
except ValueError:
print("Invalid value for ADT_REBOOT_MARK: not a test: {}".format(ADT_REBOOT_MARK))
sys.exit(2)
for test_name in remaining_tests:
result_check = False
if test_name == current_test_after_reboot:
result_check = True
suite = unittest.TestSuite()
suite.addTest(FsckdTest(test_name, result_check, return_code))
result = unittest.TextTestRunner(stream=sys.stdout, verbosity=2).run(suite)
if len(result.failures) != 0 or len(result.errors) != 0:
return_code = 1
sys.exit(return_code)
|