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