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
|
"""The ABAT harness interface
The interface as required for ABAT.
"""
__author__ = """Copyright Andy Whitcroft 2006"""
from autotest_lib.client.bin import autotest_utils
from autotest_lib.client.common_lib import utils
import os, harness, time, re
def autobench_load(fn):
disks = re.compile(r'^\s*DATS_FREE_DISKS\s*=(.*\S)\s*$')
parts = re.compile(r'^\s*DATS_FREE_PARTITIONS\s*=(.*\S)\s*$')
modules = re.compile(r'^\s*INITRD_MODULES\s*=(.*\S)\s*$')
conf = {}
try:
fd = file(fn, "r")
except:
return conf
for ln in fd.readlines():
m = disks.match(ln)
if m:
val = m.groups()[0]
conf['disks'] = val.strip('"').split()
m = parts.match(ln)
if m:
val = m.groups()[0]
conf['partitions'] = val.strip('"').split()
m = modules.match(ln)
if m:
val = m.groups()[0]
conf['modules'] = val.strip('"').split()
fd.close()
return conf
class harness_ABAT(harness.harness):
"""The ABAT server harness
Properties:
job
The job object for this job
"""
def __init__(self, job):
"""
job
The job object for this job
"""
self.setup(job)
if 'ABAT_STATUS' in os.environ:
self.status = file(os.environ['ABAT_STATUS'], "w")
else:
self.status = None
def __send(self, msg):
if self.status:
msg = msg.rstrip()
self.status.write(msg + "\n")
self.status.flush()
def __send_status(self, code, subdir, operation, msg):
self.__send("STATUS %s %s %s %s" % \
(code, subdir, operation, msg))
def __root_device(self):
device = None
root = re.compile(r'^\S*(/dev/\S+).*\s/\s*$')
df = utils.system_output('df -lP')
for line in df.split("\n"):
m = root.match(line)
if m:
device = m.groups()[0]
return device
def run_start(self):
"""A run within this job is starting"""
self.__send_status('GOOD', '----', '----', 'run starting')
# Load up the autobench.conf if it exists.
conf = autobench_load("/etc/autobench.conf")
if 'partitions' in conf:
self.job.config_set('filesystem.partitions',
conf['partitions'])
# Search the boot loader configuration for the autobench entry,
# and extract its args.
entry_args = None
args = None
for line in self.job.bootloader.info('all').split('\n'):
if line.startswith('args'):
entry_args = line.split(None, 2)[2]
if line.startswith('title'):
title = line.split()[2]
if title == 'autobench':
args = entry_args
if args:
args = re.sub(r'autobench_args:.*', '', args)
args = re.sub(r'root=\S*', '', args)
args += " root=" + self.__root_device()
self.job.config_set('boot.default_args', args)
# Turn off boot_once semantics.
self.job.config_set('boot.set_default', True)
# For RedHat installs we do not load up the module.conf
# as they cannot be builtin. Pass them as arguments.
vendor = autotest_utils.get_os_vendor()
if vendor in ['Red Hat', 'Fedora Core'] and 'modules' in conf:
args = '--allow-missing'
for mod in conf['modules']:
args += " --with " + mod
self.job.config_set('kernel.mkinitrd_extra_args', args)
def run_reboot(self):
"""A run within this job is performing a reboot
(expect continue following reboot)
"""
self.__send("REBOOT")
def run_complete(self):
"""A run within this job is completing (all done)"""
self.__send("DONE")
def test_status_detail(self, code, subdir, operation, msg, tag):
"""A test within this job is completing (detail)"""
# Send the first line with the status code as a STATUS message.
lines = msg.split("\n")
self.__send_status(code, subdir, operation, lines[0])
def test_status(self, msg, tag):
lines = msg.split("\n")
# Send each line as a SUMMARY message.
for line in lines:
self.__send("SUMMARY :" + line)
|
