forked from open-power/op-test
-
Notifications
You must be signed in to change notification settings - Fork 0
/
Copy pathOpTestKernelTest.py
275 lines (254 loc) · 11.5 KB
/
OpTestKernelTest.py
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
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
#!/usr/bin/env python3
# OpenPOWER Automated Test Project
#
# Contributors Listed Below - COPYRIGHT 2024
# [+] International Business Machines Corp.
#
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
# implied. See the License for the specific language governing
# permissions and limitations under the License.
#
# Author : Tejas Manhas <[email protected]>
# Co-Author : Abdul Haleem <[email protected]>
import json
import OpTestConfiguration
import OpTestLogger
import os
import unittest
from urllib.parse import urlparse
import re
import subprocess
import sys
import time
from common.OpTestSystem import OpSystemState
from common.OpTestSOL import OpSOLMonitorThread
from common.Exceptions import CommandFailed
from common.OpTestUtil import OpTestUtil
log = OpTestLogger.optest_logger_glob.get_logger(__name__)
class KernelTest(unittest.TestCase):
def setUp(self):
"""
Set up the test environment.
Initializes test parameters and checks required configurations.
"""
self.conf = OpTestConfiguration.conf
self.cv_HOST = self.conf.host()
self.cv_SYSTEM = self.conf.system()
self.con = self.cv_SYSTEM.cv_HOST.get_ssh_connection()
self.host_cmd_timeout = self.conf.args.host_cmd_timeout
self.repo = self.conf.args.git_repo
self.repo_reference = self.conf.args.git_repo_reference
self.branch = self.conf.args.git_branch
self.home = self.conf.args.git_home
self.config_path = self.conf.args.git_repoconfigpath
self.config = self.conf.args.git_repoconfig
self.good_commit = self.conf.args.good_commit
self.bad_commit = self.conf.args.bad_commit
self.bisect_script = self.conf.args.bisect_script
self.bisect_category = self.conf.args.bisect_category
self.append_kernel_cmdline = self.conf.args.append_kernel_cmdline
self.linux_path = os.path.join(self.home, "linux")
self.bisect_flag = self.conf.args.bisect_flag
self.util = OpTestUtil(OpTestConfiguration.conf)
self.host_distro_name = self.util.distro_name()
self.console_thread = OpSOLMonitorThread(1, "console")
# in case bisection see if we need powercycle not for build, but for boot
self.cv_SYSTEM.goto_state(OpSystemState.OFF)
self.cv_SYSTEM.goto_state(OpSystemState.OS)
self.console_thread.start()
if not self.repo:
self.fail("Provide git repo of kernel to install")
if not (self.conf.args.host_ip and self.conf.args.host_user and self.conf.args.host_password):
self.fail(
"Provide host ip user details refer, --host-{ip,user,password}")
def build_kernel(self):
"""
Build and install the Linux kernel.
"""
self.config_path = self.conf.args.git_repoconfigpath
def is_url(path):
'''
param path: path to download
return: boolean True if given path is url False Otherwise
'''
valid_schemes = ['http', 'https', 'git', 'ftp']
if urlparse(path).scheme in valid_schemes:
return True
return False
if self.config_path:
if is_url(self.config_path):
self.con.run_command("wget %s -O linux/.config" % self.config_path)
else:
self.cv_HOST.copy_test_file_to_host(self.config_path, sourcedir="", dstdir=os.path.join(linux_path, ".config"))
self.con.run_command("cd linux && make olddefconfig")
# the below part of the code is needed for only first run and will be decided bisect flag false
ker_ver = self.con.run_command("make kernelrelease")[-1]
sha = self.con.run_command("git rev-parse HEAD")
tcommit = self.con.run_command("export 'TERM=xterm-256color';git show -s --format=%ci")
tcommit = re.sub(r"\x1b\[[0-9;]*[mGKHF]", "", tcommit[1])
log.info("Upstream kernel version: %s", ker_ver)
log.info("Upstream kernel commit-id: %s", sha[-1])
log.info("Upstream kernel commit-time: %s", tcommit)
log.debug("Compile the upstream kernel")
try:
cpu= self.con.run_command("lscpu | grep '^CPU(s):' | awk '{print $2}'")
err=self.con.run_command("make -j {} -s vmlinux".format(cpu[-1]), timeout=self.host_cmd_timeout)
log.info("Kernel build successful")
return 0,err
except CommandFailed as e:
log.error("Kernel build failed: {}".format(e))
return 4,e
def Store_loc(self, er) :
"""
To get location of file in which error is introduced
"""
pattern = r"([\w\d_]+\/(?:(?:[\w\d_]+\/)*[\w\d_]+\b))"
matches = [match.group(1) for match in re.finditer(pattern,er)]
return matches
def boot_kernel(self):
"""
Does kexec boot for Upstream Linux
"""
self.con.run_command("make olddefconfig")
base_version = self.con.run_command("uname -r")
ker_ver = self.con.run_command("make kernelrelease")[-1]
cpu= self.con.run_command("lscpu | grep '^CPU(s):' | awk '{print $2}'")
self.con.run_command("make -j {} all -s vmlinux".format(cpu[-1]), timeout = 60000)
self.con.run_command("make modules_install")
self.con.run_command("make install")
if self.host_distro_name in ['rhel', 'Red Hat', 'ubuntu', 'Ubuntu']:
self.con.run_command('grubby --set-default /boot/vmlinu*-{}'.format(base_version[-1]))
elif self.host_distro_name in ['sles', 'SLES']:
self.con.run_command('grub2-set-default /boot/vmlinu*-{}'.format(base_version[-1]))
else:
raise self.skipTest("Unsupported OS")
cmdline = self.con.run_command("cat /proc/cmdline")[-1]
if self.append_kernel_cmdline:
cmdline += " %s" % self.append_kernel_cmdline
try:
initrd_file = self.con.run_command("ls -l /boot/initr*-%s.img" % ker_ver)[-1].split(" ")[-1]
except Exception:
initrd_file = self.con.run_command("ls -l /boot/initr*-%s" % ker_ver)[-1].split(" ")[-1]
kexec_cmdline = "kexec --initrd %s --command-line=\"%s\" /boot/vmlinu*-%s -l" % (initrd_file, cmdline, ker_ver)
self.con.run_command("grub2-mkconfig --output=/boot/grub2/grub.cfg")
self.con.run_command(kexec_cmdline)
self.console_thread.console_terminate()
self.cv_SYSTEM.util.build_prompt()
self.console_thread.console_terminate()
self.con.close()
time.sleep(10)
for i in range(5):
raw_pty = self.util.wait_for(self.cv_SYSTEM.console.get_console, timeout=20)
time.sleep(10)
if raw_pty is not None:
raw_pty.sendline("uname -r")
break
raw_pty.sendline("kexec -e")
boot_log=raw_pty.before
raw_pty.expect("login:", timeout=600)
raw_pty.close()
con = self.cv_SYSTEM.cv_HOST.get_ssh_connection()
kernel_version_output = con.run_command("uname -r")[-1]
log.info("Installed upstream kernel version: %s", kernel_version_output[-1])
if self.conf.args.host_cmd:
con.run_command(self.conf.args.host_cmd,
timeout=60)
self.cv_HOST.host_gather_opal_msg_log()
self.cv_HOST.host_gather_kernel_log()
if "error" in boot_log.lower() or "warning" in boot_log.lower():
print("Error or warning detected during boot process. Exiting...")
return False
if kernel_version_output != base_version :
print("Kernel booted fine. Kernel version:", kernel_version_output.strip())
return True
else:
return False
class KernelBuild(KernelTest):
"""
Does the build for any Linux repo and in case of build failure, calls build bisector
from OpTestUtils to give first bad commit and related information along with email template.
"""
def setUp(self):
"""
Does setup for KernelBUild from parent KernelTest
"""
super(KernelBuild,self).setUp()
def runTest(self):
"""
Clones git repo and builds to check for failure and do bisection
"""
self.con.run_command("if [ -d {} ]; then rm -rf {}; fi".format(self.home,self.home))
self.con.run_command("if [ ! -d {} ]; then mkdir -p {}; fi".format(self.home,self.home))
self.con.run_command("cd {}".format(self.home))
if not self.branch:
self.branch='master'
self.con.run_command("git clone --depth 1 -b {} {} linux".format( self.branch, self.repo),timeout=3000)
self.con.run_command("cd linux")
commit = self.con.run_command(" git log -1 --format=%H | sed -r 's/\x1B\[[0-9:]*[JKsu]//g'")
self.con.run_command("cd ..")
error = self.build_kernel()
exit_code = error[0]
errVal = str(error[1])
log.info("printing the exit code '{}'".format(exit_code))
entry=[]
err_msg=[]
if exit_code != 0:
entry = self.Store_loc(errVal)[-1]
err_msg= self.util.err_message(error)
badCommit = commit[-1]
if self.bisect_flag == '1':
log.info("STARTING BUILD_BISECTION")
res = self.util.build_bisector(self.linux_path, self.good_commit, self.repo)
log.info("BUILD_BISECTION ENDED")
emaili=res[0]
commiti=res[1]
log.info("revert commit check is manual for now")
else :
emaili=""
commiti=commit[-1]
else :
emaili=""
commiti=commit[-1]
with open('output.json','w') as f:
json.dump({"exit_code":exit_code,"email":emaili,"commit": commiti,"error":entry,"err_msg":err_msg,"flag":self.bisect_flag},f)
if exit_code != 0:
self.util.format_email(self.linux_path, self.repo)
def tearDown(self):
self.console_thread.console_terminate()
self.con.close()
class KernelBoot(KernelTest):
def setUp(self):
"""
Does setup for KernelBoot from parent KernelTest
"""
super(KernelBoot,self).setUp()
def runTest(self):
"""
Clones git repo and boots the kernel to check for failure and do bisection
"""
self.con.run_command("if [ -d {} ]; then rm -rf {}; fi".format(self.home,self.home))
self.con.run_command("if [ ! -d {} ]; then mkdir -p {}; fi".format(self.home,self.home))
self.con.run_command("cd {}".format(self.home))
if not self.branch:
self.branch='master'
self.con.run_command("git clone --depth 1 -b {} {} linux".format( self.branch, self.repo),timeout=3000)
self.con.run_command("cd linux")
commit = self.con.run_command(" git log -1 --format=%H | sed -r 's/\x1B\[[0-9:]*[JKsu]//g'")
self.con.run_command("cd ..")
error = self.build_kernel()
exit_code = error[0]
if exit_code != 0:
return "Build Failure in boot, check build bisection Aborting"
self.boot_kernel()
def tearDown(self):
self.console_thread.console_terminate()
self.con.close()