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")
|
