1# 2# SPDX-License-Identifier: MIT 3# 4 5import re 6import time 7 8from oeqa.runtime.case import OERuntimeTestCase 9from oeqa.core.decorator.depends import OETestDepends 10from oeqa.core.decorator.data import skipIfDataVar, skipIfNotDataVar 11from oeqa.runtime.decorator.package import OEHasPackage 12from oeqa.core.decorator.data import skipIfNotFeature, skipIfFeature 13 14class SystemdTest(OERuntimeTestCase): 15 16 def systemctl(self, action='', target='', expected=0, verbose=False): 17 command = 'SYSTEMD_BUS_TIMEOUT=240s systemctl %s %s' % (action, target) 18 status, output = self.target.run(command) 19 message = '\n'.join([command, output]) 20 if status != expected and verbose: 21 cmd = 'SYSTEMD_BUS_TIMEOUT=240s systemctl status --full %s' % target 22 message += self.target.run(cmd)[1] 23 self.assertEqual(status, expected, message) 24 return output 25 26 #TODO: use pyjournalctl instead 27 def journalctl(self, args='',l_match_units=None): 28 """ 29 Request for the journalctl output to the current target system 30 31 Arguments: 32 -args, an optional argument pass through argument 33 -l_match_units, an optional list of units to filter the output 34 Returns: 35 -string output of the journalctl command 36 Raises: 37 -AssertionError, on remote commands that fail 38 -ValueError, on a journalctl call with filtering by l_match_units that 39 returned no entries 40 """ 41 42 query_units='' 43 if l_match_units: 44 query_units = ['_SYSTEMD_UNIT='+unit for unit in l_match_units] 45 query_units = ' '.join(query_units) 46 command = 'journalctl %s %s' %(args, query_units) 47 status, output = self.target.run(command) 48 if status: 49 raise AssertionError("Command '%s' returned non-zero exit " 50 'code %d:\n%s' % (command, status, output)) 51 if len(output) == 1 and "-- No entries --" in output: 52 raise ValueError('List of units to match: %s, returned no entries' 53 % l_match_units) 54 return output 55 56class SystemdBasicTests(SystemdTest): 57 58 def settle(self): 59 """ 60 Block until systemd has finished activating any units being activated, 61 or until two minutes has elapsed. 62 63 Returns a tuple, either (True, '') if all units have finished 64 activating, or (False, message string) if there are still units 65 activating (generally, failing units that restart). 66 """ 67 endtime = time.time() + (60 * 2) 68 while True: 69 status, output = self.target.run('SYSTEMD_BUS_TIMEOUT=240s systemctl --state=activating') 70 if "0 loaded units listed" in output: 71 return (True, '') 72 if time.time() >= endtime: 73 return (False, output) 74 time.sleep(10) 75 76 @skipIfNotFeature('systemd', 77 'Test requires systemd to be in DISTRO_FEATURES') 78 @skipIfNotDataVar('VIRTUAL-RUNTIME_init_manager', 'systemd', 79 'systemd is not the init manager for this image') 80 @OETestDepends(['ssh.SSHTest.test_ssh']) 81 def test_systemd_basic(self): 82 self.systemctl('--version') 83 84 @OETestDepends(['systemd.SystemdBasicTests.test_systemd_basic']) 85 def test_systemd_list(self): 86 self.systemctl('list-unit-files') 87 88 @OETestDepends(['systemd.SystemdBasicTests.test_systemd_basic']) 89 def test_systemd_failed(self): 90 settled, output = self.settle() 91 msg = "Timed out waiting for systemd to settle:\n%s" % output 92 self.assertTrue(settled, msg=msg) 93 94 output = self.systemctl('list-units', '--failed') 95 match = re.search('0 loaded units listed', output) 96 if not match: 97 output += self.systemctl('status --full --failed') 98 self.assertTrue(match, msg='Some systemd units failed:\n%s' % output) 99 100 101class SystemdServiceTests(SystemdTest): 102 103 @OEHasPackage(['avahi-daemon']) 104 @OETestDepends(['systemd.SystemdBasicTests.test_systemd_basic']) 105 def test_systemd_status(self): 106 self.systemctl('status --full', 'avahi-daemon.service') 107 108 @OETestDepends(['systemd.SystemdServiceTests.test_systemd_status']) 109 def test_systemd_stop_start(self): 110 self.systemctl('stop', 'avahi-daemon.service') 111 self.systemctl('is-active', 'avahi-daemon.service', 112 expected=3, verbose=True) 113 self.systemctl('start','avahi-daemon.service') 114 self.systemctl('is-active', 'avahi-daemon.service', verbose=True) 115 116 @OETestDepends(['systemd.SystemdServiceTests.test_systemd_status']) 117 @skipIfFeature('read-only-rootfs', 118 'Test is only meant to run without read-only-rootfs in IMAGE_FEATURES') 119 def test_systemd_disable_enable(self): 120 self.systemctl('disable', 'avahi-daemon.service') 121 self.systemctl('is-enabled', 'avahi-daemon.service', expected=1) 122 self.systemctl('enable', 'avahi-daemon.service') 123 self.systemctl('is-enabled', 'avahi-daemon.service') 124 125 @OETestDepends(['systemd.SystemdServiceTests.test_systemd_status']) 126 @skipIfNotFeature('read-only-rootfs', 127 'Test is only meant to run with read-only-rootfs in IMAGE_FEATURES') 128 def test_systemd_disable_enable_ro(self): 129 status = self.target.run('mount -orw,remount /')[0] 130 self.assertTrue(status == 0, msg='Remounting / as r/w failed') 131 try: 132 self.test_systemd_disable_enable() 133 finally: 134 status = self.target.run('mount -oro,remount /')[0] 135 self.assertTrue(status == 0, msg='Remounting / as r/o failed') 136 137class SystemdJournalTests(SystemdTest): 138 139 @OETestDepends(['systemd.SystemdBasicTests.test_systemd_basic']) 140 def test_systemd_journal(self): 141 status, output = self.target.run('journalctl') 142 self.assertEqual(status, 0, output) 143 144 @OETestDepends(['systemd.SystemdBasicTests.test_systemd_basic']) 145 def test_systemd_boot_time(self, systemd_TimeoutStartSec=90): 146 """ 147 Get the target boot time from journalctl and log it 148 149 Arguments: 150 -systemd_TimeoutStartSec, an optional argument containing systemd's 151 unit start timeout to compare against 152 """ 153 154 # The expression chain that uniquely identifies the time boot message. 155 expr_items=['Startup finished', 'kernel', 'userspace','\.$'] 156 try: 157 output = self.journalctl(args='-o cat --reverse') 158 except AssertionError: 159 self.fail('Error occurred while calling journalctl') 160 if not len(output): 161 self.fail('Error, unable to get startup time from systemd journal') 162 163 # Check for the regular expression items that match the startup time. 164 for line in output.split('\n'): 165 check_match = ''.join(re.findall('.*'.join(expr_items), line)) 166 if check_match: 167 break 168 # Put the startup time in the test log 169 if check_match: 170 self.tc.logger.info('%s' % check_match) 171 else: 172 self.skipTest('Error at obtaining the boot time from journalctl') 173 boot_time_sec = 0 174 175 # Get the numeric values from the string and convert them to seconds 176 # same data will be placed in list and string for manipulation. 177 l_boot_time = check_match.split(' ')[-2:] 178 s_boot_time = ' '.join(l_boot_time) 179 try: 180 # Obtain the minutes it took to boot. 181 if l_boot_time[0].endswith('min') and l_boot_time[0][0].isdigit(): 182 boot_time_min = s_boot_time.split('min')[0] 183 # Convert to seconds and accumulate it. 184 boot_time_sec += int(boot_time_min) * 60 185 # Obtain the seconds it took to boot and accumulate. 186 boot_time_sec += float(l_boot_time[1].split('s')[0]) 187 except ValueError: 188 self.skipTest('Error when parsing time from boot string') 189 190 # Assert the target boot time against systemd's unit start timeout. 191 if boot_time_sec > systemd_TimeoutStartSec: 192 msg = ("Target boot time %s exceeds systemd's TimeoutStartSec %s" 193 % (boot_time_sec, systemd_TimeoutStartSec)) 194 self.tc.logger.info(msg) 195