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