1#
2# Copyright OpenEmbedded Contributors
3#
4# SPDX-License-Identifier: MIT
5#
6
7from oeqa.runtime.case import OERuntimeTestCase
8from oeqa.core.decorator.depends import OETestDepends
9from oeqa.core.decorator.data import skipIfDataVar
10from oeqa.runtime.decorator.package import OEHasPackage
11import time
12
13class SyslogTest(OERuntimeTestCase):
14
15    @OETestDepends(['ssh.SSHTest.test_ssh'])
16    @OEHasPackage(["busybox-syslog", "sysklogd", "rsyslog", "syslog-ng"])
17    def test_syslog_running(self):
18        status, output = self.target.run(self.tc.target_cmds['ps'])
19        msg = "Failed to execute %s" % self.tc.target_cmds['ps']
20        self.assertEqual(status, 0, msg=msg)
21        msg = "No syslog daemon process; %s output:\n%s" % (self.tc.target_cmds['ps'], output)
22        hasdaemon = "syslogd" in output or "syslog-ng" in output or "svlogd" in output
23        self.assertTrue(hasdaemon, msg=msg)
24
25class SyslogTestConfig(OERuntimeTestCase):
26
27    def verif_not_running(self, pids):
28        for pid in pids:
29            status, err_output = self.target.run('kill -0 %s' %pid)
30            if not status:
31                self.logger.debug("previous %s is still running" %pid)
32                return 1
33
34    def verify_running(self, names):
35        pids = []
36        for name in names:
37            status, pid = self.target.run('pidof %s' %name)
38            if status:
39                self.logger.debug("%s is not running" %name)
40                return 1, pids
41            pids.append(pid)
42        return 0, pids
43
44
45    def restart_sanity(self, names, restart_cmd, pidchange=True):
46        status, original_pids = self.verify_running(names)
47        if status:
48            return False
49
50        status, output = self.target.run(restart_cmd)
51
52        msg = ('Could not restart %s service. Status and output: %s and %s' % (names, status, output))
53        self.assertEqual(status, 0, msg)
54
55        if not pidchange:
56            return True
57
58        # Always check for an error, most likely a race between shutting down and starting up
59        timeout = time.time() + 30
60
61        restarted = False
62        status = ""
63        while time.time() < timeout:
64            # Verify the previous ones are no longer running
65            status = self.verif_not_running(original_pids)
66            if status:
67                status = "Original syslog processes still running"
68                continue
69
70            status, pids = self.verify_running(names)
71            if status:
72                status = "New syslog processes not running"
73                continue
74
75            # Everything is fine now, so exit to continue the test
76            restarted = True
77            break
78
79        msg = ('%s didn\'t appear to restart: %s' % (names, status))
80        self.assertTrue(restarted, msg)
81
82        return True
83
84    @OETestDepends(['oe_syslog.SyslogTest.test_syslog_running'])
85    def test_syslog_logger(self):
86        status, output = self.target.run('logger foobar')
87        msg = "Can't log into syslog. Output: %s " % output
88        self.assertEqual(status, 0, msg=msg)
89
90        # There is no way to flush the logger to disk in all cases
91        time.sleep(1)
92
93        status, output = self.target.run('grep foobar /var/log/messages')
94        if status != 0:
95            if self.tc.td.get("VIRTUAL-RUNTIME_init_manager") == "systemd":
96                status, output = self.target.run('journalctl -o cat | grep foobar')
97            else:
98                status, output = self.target.run('logread | grep foobar')
99        msg = ('Test log string not found in /var/log/messages or logread.'
100               ' Output: %s ' % output)
101        self.assertEqual(status, 0, msg=msg)
102
103
104    @OETestDepends(['oe_syslog.SyslogTest.test_syslog_running'])
105    def test_syslog_restart(self):
106        if self.restart_sanity(['systemd-journald'], 'systemctl restart syslog.service', pidchange=False):
107            pass
108        elif self.restart_sanity(['rsyslogd'], '/etc/init.d/rsyslog restart'):
109            pass
110        elif self.restart_sanity(['syslogd', 'klogd'], '/etc/init.d/syslog restart'):
111            pass
112        else:
113            self.logger.info("No syslog found to restart, ignoring")
114
115
116    @OETestDepends(['oe_syslog.SyslogTestConfig.test_syslog_logger'])
117    @OEHasPackage(["busybox-syslog"])
118    @skipIfDataVar('VIRTUAL-RUNTIME_init_manager', 'systemd',
119                   'Not appropiate for systemd image')
120    def test_syslog_startup_config(self):
121        cmd = 'echo "LOGFILE=/var/log/test" >> /etc/syslog-startup.conf'
122        self.target.run(cmd)
123
124        self.test_syslog_restart()
125
126        cmd = 'logger foobar'
127        status, output = self.target.run(cmd)
128        msg = 'Logger command failed, %s. Output: %s ' % (status, output)
129        self.assertEqual(status, 0, msg=msg)
130
131        cmd = 'cat /var/log/test'
132        status, output = self.target.run(cmd)
133        if "foobar" not in output or status:
134            self.fail("'foobar' not found in logfile, status %s, contents %s" % (status, output))
135
136        cmd = "sed -i 's#LOGFILE=/var/log/test##' /etc/syslog-startup.conf"
137        self.target.run(cmd)
138        self.test_syslog_restart()
139