1#
2# Copyright OpenEmbedded Contributors
3#
4# SPDX-License-Identifier: MIT
5#
6
7import os
8import time
9
10from oeqa.runtime.case import OERuntimeTestCase
11from oeqa.core.decorator.depends import OETestDepends
12from oeqa.core.decorator.data import skipIfNotFeature
13
14# need some kernel fragments
15# echo "KERNEL_FEATURES:append = \" features\/kernel\-sample\/kernel\-sample.scc\"" >> local.conf
16class KSample(OERuntimeTestCase):
17    def cmd_and_check(self, cmd='', match_string=''):
18        status, output = self.target.run(cmd)
19        if not match_string:
20            # send cmd
21            msg = '%s failed, %s' % (cmd, output)
22            self.assertEqual(status, 0, msg=msg)
23        else:
24            # check result
25            result = ("%s" % match_string) in output
26            msg = output
27            self.assertTrue(result, msg)
28            self.assertEqual(status, 0, cmd)
29
30    def check_arch(self, archset=''):
31        status, output = self.target.run("uname -m")
32        result = ("%s" % output) in archset
33        if not result:
34            self.skipTest("This case doesn't support %s" % output)
35
36    def check_config(self, config_opt=''):
37        cmd = "zcat /proc/config.gz | grep %s" % config_opt
38        status, output = self.target.run(cmd)
39        result = ("%s=y" % config_opt) in output
40        if not result:
41            self.skipTest("%s is not set" % config_opt)
42
43    def check_module_exist(self, path='', module_name=''):
44        status, output = self.target.run("uname -r")
45        cmd = "ls " + "/lib/modules/" + output + "/kernel/samples/" + path + module_name
46        status, output = self.target.run(cmd)
47        if status != 0:
48            error_info = module_name + " doesn't exist"
49            self.skipTest(error_info)
50
51    def kfifo_func(self, name=''):
52        module_prename = name + "-example"
53        module_name = name + "-example.ko"
54        sysmbol_name = name + "_example"
55
56        # make sure if module exists
57        self.check_module_exist("kfifo/", module_name)
58        # modprobe
59        self.cmd_and_check("modprobe %s" % module_prename)
60        # lsmod
61        self.cmd_and_check("lsmod | grep %s | cut -d\' \' -f1" % sysmbol_name, sysmbol_name)
62        # check result
63        self.cmd_and_check("dmesg | grep \"test passed\" ", "test passed")
64        # rmmod
65        self.cmd_and_check("rmmod %s" %  module_prename)
66
67    def kprobe_func(self, name=''):
68        # check config
69        self.check_config("CONFIG_KPROBES")
70
71        module_prename = name + "_example"
72        module_name = name + "_example.ko"
73        sysmbol_name = module_prename
74
75        # make sure if module exists
76        self.check_module_exist("kprobes/", module_name)
77        # modprobe
78        self.cmd_and_check("modprobe %s" % module_prename)
79        # lsmod
80        self.cmd_and_check("lsmod | grep %s | cut -d\' \' -f1" % sysmbol_name, sysmbol_name)
81        # check result
82        self.cmd_and_check("dmesg | grep Planted | head -n10", "Planted")
83        # rmmod
84        self.cmd_and_check("rmmod %s" % module_prename)
85
86    def kobject_func(self, name=''):
87        module_prename = name + "_example"
88        module_name = name + "-example.ko"
89        sysmbol_name = module_prename
90
91        # make sure if module exists
92        self.check_module_exist("kobject/", module_name)
93        # modprobe
94        self.cmd_and_check("modprobe %s" % module_prename)
95        # lsmod
96        self.cmd_and_check("lsmod | grep %s | cut -d\' \' -f1" % sysmbol_name, sysmbol_name)
97        # check result
98        self.cmd_and_check("ls /sys/kernel/%s/" % sysmbol_name, "bar")
99        # rmmod
100        self.cmd_and_check("rmmod %s" % module_prename)
101
102class KSampleTest(KSample):
103    # kfifo
104    @OETestDepends(['ssh.SSHTest.test_ssh'])
105    def test_kfifo_test(self):
106        index = ["dma", "bytestream", "inttype", "record"]
107        for i in index:
108            self.kfifo_func(i)
109
110    # kprobe
111    @OETestDepends(['ssh.SSHTest.test_ssh'])
112    def test_kprobe_test(self):
113        self.check_arch("x86_64 i686 ppc")
114        index = ["kprobe", "kretprobe"]
115        for i in index:
116            self.kprobe_func(i)
117
118    # kobject
119    @OETestDepends(['ssh.SSHTest.test_ssh'])
120    def test_kobject_test(self):
121        index = ["kobject", "kset"]
122        for i in index:
123            self.kobject_func(i)
124
125    #trace
126    @OETestDepends(['ssh.SSHTest.test_ssh'])
127    def test_trace_events(self):
128        # check config
129        self.check_config("CONFIG_TRACING_SUPPORT")
130        # make sure if module exists
131        self.check_module_exist("trace_events/", "trace-events-sample.ko")
132        # modprobe
133        self.cmd_and_check("modprobe trace-events-sample")
134        # lsmod
135        self.cmd_and_check("lsmod | grep trace_events_sample | cut -d\' \' -f1", "trace_events_sample")
136        # check dir
137        self.cmd_and_check("ls /sys/kernel/debug/tracing/events/ | grep sample-trace", "sample-trace")
138        # enable trace
139        self.cmd_and_check("echo 1 > /sys/kernel/debug/tracing/events/sample-trace/enable")
140        self.cmd_and_check("cat /sys/kernel/debug/tracing/events/sample-trace/enable")
141        # check result
142        status = 1
143        count = 0
144        while status != 0:
145            time.sleep(1)
146            status, output = self.target.run('cat /sys/kernel/debug/tracing/trace | grep hello | head -n1 | cut -d\':\' -f2')
147            if " foo_bar" in output:
148                break
149            count = count + 1
150            if count > 5:
151                self.assertTrue(False, "Time out when check result")
152        # disable trace
153        self.cmd_and_check("echo 0 > /sys/kernel/debug/tracing/events/sample-trace/enable")
154        # clean up trace
155        self.cmd_and_check("echo > /sys/kernel/debug/tracing/trace")
156        # rmmod
157        self.cmd_and_check("rmmod trace-events-sample")
158
159    @OETestDepends(['ssh.SSHTest.test_ssh'])
160    def test_trace_printk(self):
161        # check config
162        self.check_config("CONFIG_TRACING_SUPPORT")
163        # make sure if module exists
164        self.check_module_exist("trace_printk/", "trace-printk.ko")
165        # modprobe
166        self.cmd_and_check("modprobe trace-printk")
167        # lsmod
168        self.cmd_and_check("lsmod | grep trace_printk | cut -d\' \' -f1", "trace_printk")
169        # check result
170        self.cmd_and_check("cat /sys/kernel/debug/tracing/trace | grep trace_printk_irq_work | head -n1 | cut -d\':\' -f2", " trace_printk_irq_work")
171        # clean up trace
172        self.cmd_and_check("echo > /sys/kernel/debug/tracing/trace")
173        # rmmod
174        self.cmd_and_check("rmmod trace-printk")
175
176    # hw breakpoint
177    @OETestDepends(['ssh.SSHTest.test_ssh'])
178    def test_hw_breakpoint_example(self):
179        # check arch
180        status, output = self.target.run("uname -m")
181        result = ("x86_64" in output) or ("aarch64" in output)
182        if not result:
183            self.skipTest("the arch %s doesn't support hw breakpoint" % output)
184        # check config
185        self.check_config("CONFIG_KALLSYMS_ALL")
186        # make sure if module exists
187        self.check_module_exist("hw_breakpoint/", "data_breakpoint.ko")
188        # modprobe
189        self.cmd_and_check("modprobe data_breakpoint")
190        # lsmod
191        self.cmd_and_check("lsmod | grep data_breakpoint | cut -d\' \' -f1", "data_breakpoint")
192        # check result
193        self.cmd_and_check("cat /var/log/messages | grep sample_hbp_handler", "sample_hbp_handler")
194        # rmmod
195        self.cmd_and_check("rmmod data_breakpoint")
196
197    @OETestDepends(['ssh.SSHTest.test_ssh'])
198    def test_configfs_sample(self):
199        # check config
200        status, ret = self.target.run('zcat /proc/config.gz | grep CONFIG_CONFIGFS_FS')
201        if not ["CONFIG_CONFIGFS_FS=m" in ret or "CONFIG_CONFIGFS_FS=y" in ret]:
202            self.skipTest("CONFIG error")
203        # make sure if module exists
204        self.check_module_exist("configfs/", "configfs_sample.ko")
205        # modprobe
206        self.cmd_and_check("modprobe configfs_sample")
207        # lsmod
208        self.cmd_and_check("lsmod | grep configfs_sample | cut -d\' \' -f1 | head -n1", "configfs_sample")
209
210        status = 1
211        count = 0
212        while status != 0:
213            time.sleep(1)
214            status, ret = self.target.run('cat /sys/kernel/config/01-childless/description')
215            count = count + 1
216            if count > 200:
217                self.skipTest("Time out for check dir")
218
219        # rmmod
220        self.cmd_and_check("rmmod configfs_sample")
221
222    @OETestDepends(['ssh.SSHTest.test_ssh'])
223    def test_cn_test(self):
224        # make sure if module exists
225        self.check_module_exist("connector/", "cn_test.ko")
226        # modprobe
227        self.cmd_and_check("modprobe cn_test")
228        # lsmod
229        self.cmd_and_check("lsmod | grep cn_test | cut -d\' \' -f1", "cn_test")
230        # check result
231        self.cmd_and_check("cat /proc/net/connector | grep cn_test | head -n1 | cut -d\' \' -f1", "cn_test")
232        # rmmod
233        self.cmd_and_check("rmmod cn_test")
234