aboutsummaryrefslogtreecommitdiffstats
path: root/meta/lib/oeqa/runtime/cases/ksample.py
blob: eb8713351afbd688d51a106d3d66922fbb9495e9 (plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
import os
import time

from oeqa.runtime.case import OERuntimeTestCase
from oeqa.core.decorator.depends import OETestDepends
from oeqa.core.decorator.oeid import OETestID
from oeqa.core.decorator.data import skipIfNotFeature

# need some kernel fragments
# echo "KERNEL_FEATURES_append += \" features\/kernel\-sample\/kernel\-sample.scc\"" >> local.conf
class KSample(OERuntimeTestCase):
    def cmd_and_check(self, cmd='', match_string=''):
        status, output = self.target.run(cmd)
        if not match_string:
            # send cmd
            msg = '%s failed, %s' % (cmd, output)
            self.assertEqual(status, 0, msg=msg)
        else:
            # check result
            result = ("%s" % match_string) in output
            msg = output
            self.assertTrue(result, msg)
            self.assertEqual(status, 0, cmd)

    def check_config(self, config_opt=''):
        cmd = "zcat /proc/config.gz | grep %s" % config_opt
        status, output = self.target.run(cmd)
        result = ("%s=y" % config_opt) in output
        if not result:
            self.skipTest("%s is not set" % config_opt)

    def check_module_exist(self, path='', module_name=''):
        status, output = self.target.run("uname -r")
        cmd = "ls " + "/lib/modules/" + output + "/kernel/samples/" + path + module_name
        status, output = self.target.run(cmd)
        if status != 0:
            error_info = module_name + "doesn't exist"
            self.skipTest(error_info)

    def kfifo_func(self, name=''):
        module_prename = name + "-example"
        module_name = name + "-example.ko"
        sysmbol_name = name + "_example"

        # make sure if module exists
        self.check_module_exist("kfifo/", module_name)
        # modprobe
        self.cmd_and_check("modprobe %s" % module_prename)
        # lsmod
        self.cmd_and_check("lsmod | grep %s | cut -d\' \' -f1" % sysmbol_name, sysmbol_name)
        # check result
        self.cmd_and_check("dmesg | grep \"test passed\" ", "test passed")
        # rmmod
        self.cmd_and_check("rmmod %s" %  module_prename)

    def kprobe_func(self, name=''):
        # check config
        self.check_config("CONFIG_KPROBES")

        module_prename = name + "_example"
        module_name = name + "_example.ko"
        sysmbol_name = module_prename

        # make sure if module exists
        self.check_module_exist("kprobes/", module_name)
        # modprobe
        self.cmd_and_check("modprobe %s" % module_prename)
        # lsmod
        self.cmd_and_check("lsmod | grep %s | cut -d\' \' -f1" % sysmbol_name, sysmbol_name)
        # check result
        self.cmd_and_check("dmesg | grep Planted | head -n10", "Planted")
        # rmmod
        self.cmd_and_check("rmmod %s" % module_prename)

    def kobject_func(self, name=''):
        module_prename = name + "_example"
        module_name = name + "-example.ko"
        sysmbol_name = module_prename

        # make sure if module exists
        self.check_module_exist("kobject/", module_name)
        # modprobe
        self.cmd_and_check("modprobe %s" % module_prename)
        # lsmod
        self.cmd_and_check("lsmod | grep %s | cut -d\' \' -f1" % sysmbol_name, sysmbol_name)
        # check result
        self.cmd_and_check("ls /sys/kernel/%s/" % sysmbol_name, "bar")
        # rmmod
        self.cmd_and_check("rmmod %s" % module_prename)

class KSampleTest(KSample):
    # kfifo
    @OETestDepends(['ssh.SSHTest.test_ssh'])
    def test_kfifo_test(self):
        index = ["dma", "bytestream", "inttype", "record"]
        for i in index:
            self.kfifo_func(i)

    # kprobe
    @OETestDepends(['ssh.SSHTest.test_ssh'])
    def test_kprobe_test(self):
        index = ["kprobe", "kretprobe"]
        for i in index:
            self.kprobe_func(i)

    # kobject
    @OETestDepends(['ssh.SSHTest.test_ssh'])
    def test_kobject_test(self):
        index = ["kobject", "kset"]
        for i in index:
            self.kobject_func(i)

    #trace
    @OETestDepends(['ssh.SSHTest.test_ssh'])
    def test_trace_events(self):
        # check config
        self.check_config("CONFIG_TRACING_SUPPORT")
        # make sure if module exists
        self.check_module_exist("trace_events/", "trace-events-sample.ko")
        # modprobe
        self.cmd_and_check("modprobe trace-events-sample")
        # lsmod
        self.cmd_and_check("lsmod | grep trace_events_sample | cut -d\' \' -f1", "trace_events_sample")
        # check dir
        self.cmd_and_check("ls /sys/kernel/debug/tracing/events/ | grep sample-trace", "sample-trace")
        # enable trace
        self.cmd_and_check("echo 1 > /sys/kernel/debug/tracing/events/sample-trace/enable")
        self.cmd_and_check("cat /sys/kernel/debug/tracing/events/sample-trace/enable")
        # check result
        status = 1
        count = 0
        while status != 0:
            time.sleep(1)
            status, output = self.target.run('cat /sys/kernel/debug/tracing/trace | grep hello | head -n1 | cut -d\':\' -f2')
            if " foo_bar" in output:
                break
            count = count + 1
            if count > 5:
                self.assertTrue(False, "Time out when check result")
        # disable trace
        self.cmd_and_check("echo 0 > /sys/kernel/debug/tracing/events/sample-trace/enable")
        # clean up trace
        self.cmd_and_check("echo > /sys/kernel/debug/tracing/trace")
        # rmmod
        self.cmd_and_check("rmmod trace-events-sample")

    @OETestDepends(['ssh.SSHTest.test_ssh'])
    def test_trace_printk(self):
        # check config
        self.check_config("CONFIG_TRACING_SUPPORT")
        # make sure if module exists
        self.check_module_exist("trace_printk/", "trace-printk.ko")
        # modprobe
        self.cmd_and_check("modprobe trace-printk")
        # lsmod
        self.cmd_and_check("lsmod | grep trace_printk | cut -d\' \' -f1", "trace_printk")
        # check result
        self.cmd_and_check("cat /sys/kernel/debug/tracing/trace | grep trace_printk_irq_work | head -n1 | cut -d\':\' -f2", " trace_printk_irq_work")
        # clean up trace
        self.cmd_and_check("echo > /sys/kernel/debug/tracing/trace")
        # rmmod
        self.cmd_and_check("rmmod trace-printk")

    # hw breakpoint
    @OETestDepends(['ssh.SSHTest.test_ssh'])
    def test_hw_breakpoint_example(self):
        # check arch
        status, output = self.target.run("uname -m")
        result = ("x86" in output) or ("aarch64" in output)
        if not result:
            self.skipTest("the arch doesn't support hw breakpoint" % output)
        # check config
        self.check_config("CONFIG_KALLSYMS_ALL")
        # make sure if module exists
        self.check_module_exist("hw_breakpoint/", "data_breakpoint.ko")
        # modprobe
        self.cmd_and_check("modprobe data_breakpoint")
        # lsmod
        self.cmd_and_check("lsmod | grep data_breakpoint | cut -d\' \' -f1", "data_breakpoint")
        # check result
        self.cmd_and_check("cat /var/log/messages | grep sample_hbp_handler", "sample_hbp_handler")
        # rmmod
        self.cmd_and_check("rmmod data_breakpoint")

    @OETestDepends(['ssh.SSHTest.test_ssh'])
    def test_configfs_sample(self):
        # check config
        status, ret = self.target.run('zcat /proc/config.gz | grep CONFIG_CONFIGFS_FS')
        if not ["CONFIG_CONFIGFS_FS=m" in ret or "CONFIG_CONFIGFS_FS=y" in ret]:
            self.skipTest("CONFIG error")
        # make sure if module exists
        self.check_module_exist("configfs/", "configfs_sample.ko")
        # modprobe
        self.cmd_and_check("modprobe configfs_sample")
        # lsmod
        self.cmd_and_check("lsmod | grep configfs_sample | cut -d\' \' -f1 | head -n1", "configfs_sample")

        status = 1
        count = 0
        while status != 0:
            time.sleep(1)
            status, ret = self.target.run('cat /sys/kernel/config/01-childless/description')
            count = count + 1
            if count > 200:
                self.skipTest("Time out for check dir")

        # rmmod
        self.cmd_and_check("rmmod configfs_sample")

    @OETestDepends(['ssh.SSHTest.test_ssh'])
    def test_cn_test(self):
        # make sure if module exists
        self.check_module_exist("connector/", "cn_test.ko")
        # modprobe
        self.cmd_and_check("modprobe cn_test")
        # lsmod
        self.cmd_and_check("lsmod | grep cn_test | cut -d\' \' -f1", "cn_test")
        # check result
        self.cmd_and_check("cat /proc/net/connector | grep cn_test | head -n1 | cut -d\' \' -f1", "cn_test")
        # rmmod
        self.cmd_and_check("rmmod cn_test")