blob: f1b509163ecf6b63ae80d8a38d2f548a1cdb91ec [file] [log] [blame]
import email, os, re, smtplib
from autotest_lib.server import frontend
class trigger(object):
Base trigger class. You are allowed to derive from it and
override functions to suit your needs in the configuration file.
def __init__(self):
self.__actions = []
def run(self, files):
# Call each of the actions and pass in the kernel list
for action in self.__actions:
def add_action(self, func):
class base_action(object):
Base class for functor actions. Since actions can also be simple functions
all action classes need to override __call__ to be callable.
def __call__(self, kernel_list):
Perform the action for that given kernel filenames list.
@param kernel_list: a sequence of kernel filenames (strings)
raise NotImplementedError('__call__ not implemented')
class map_action(base_action):
Action that uses a map between machines and their associated control
files and kernel configuration files and it schedules them using
the AFE.
_encode_sep = re.compile('(\D+)')
class machine_info(object):
Class to organize the machine associated information for this action.
def __init__(self, tests, kernel_configs):
Instantiate a machine_info object.
@param tests: a sequence of test names (as named in the frontend
database) to run for this host
@param kernel_configs: a dictionary of
kernel_version -> config_filename associating kernel
versions with corresponding kernel configuration files
("~" inside the filename will be expanded)
self.tests = tests
self.kernel_configs = kernel_configs
def __init__(self, tests_map, jobname_pattern, job_owner='autotest',
Instantiate a map_action.
@param tests_map: a dictionary of hostname -> machine_info
@param jobname_pattern: a string pattern used to make the job name
containing a single "%s" that will be replaced with the kernel
@param job_owner: the user used to talk with the RPC server
@param upload_kernel_config: specify if the generate control file
should contain code that downloads and sends to the client the
kernel config file (in case it is an URL); this requires that
the tests_map refers only to server side tests
self._tests_map = tests_map
self._jobname_pattern = jobname_pattern
self._afe = frontend.AFE(user=job_owner)
self._upload_kernel_config = upload_kernel_config
def __call__(self, kernel_list):
Schedule jobs to run on the given list of kernel versions using
the configured machines -> machine_info mapping for test name
selection and kernel config file selection.
for kernel in kernel_list:
# Get a list of all the machines available for testing
# and the tests that each one is to execute and group them by
# test/kernel-config so we can run a single job for the same
# group
# dictionary of (test-name,kernel-config)-><list-of-machines>
jobs = {}
for machine, info in self._tests_map.iteritems():
config_paths = info.kernel_configs
kernel_config = '/boot/config'
if config_paths:
kvers = config_paths.keys()
close = self._closest_kver_leq(kvers, kernel)
kernel_config = config_paths[close]
for test in info.tests:
jobs.setdefault((test, kernel_config), [])
jobs[(test, kernel_config)].append(machine)
for (test, kernel_config), hosts in jobs.iteritems():
c = self._generate_control(test, kernel, kernel_config)
self._schedule_job(self._jobname_pattern % kernel, c, hosts)
def _kver_encode(cls, version):
Encode the various kernel version strings (ex 2.6.20, 2.6.21-rc1,
2.7.30-rc2-git3, etc) in a way that makes them easily comparable using
lexicographic ordering.
@param version: kernel version string to encode
@return processed kernel version string that can be compared using
lexicographic comparison
# if it's not a "rc" release, add a -rc99 so it orders at the end of
# all other rc releases for the same base kernel version
if 'rc' not in version:
version += '-rc99'
# if it's not a git snapshot add a -git99 so it orders at the end of
# all other git snapshots for the same base kernel version
if 'git' not in version:
version += '-git99'
# make all number sequences to be at least 2 in size (with a leading 0
# if necessary)
bits = cls._encode_sep.split(version)
for n in range(0, len(bits), 2):
if len(bits[n]) < 2:
bits[n] = '0' + bits[n]
return ''.join(bits)
def _kver_cmp(cls, a, b):
Compare 2 kernel versions.
@param a, b: kernel version strings to compare
@return True if 'a' is less than 'b' or False otherwise
a, b = cls._kver_encode(a), cls._kver_encode(b)
return cmp(a, b)
def _closest_kver_leq(cls, klist, kver):
Return the closest kernel ver in the list that is <= kver unless
kver is the lowest, in which case return the lowest in klist.
if kver in klist:
return kver
l = list(klist)
i = l.index(kver)
if i == 0:
return l[1]
return l[i - 1]
def _generate_control(self, test, kernel, kernel_config):
Uses generate_control_file RPC to generate a control file given
a test name and kernel information.
@param test: The test name string as it's named in the frontend
@param kernel: A str of the kernel version (i.e. x.x.xx)
@param kernel_config: A str filename to the kernel config on the
@returns a dict representing a control file as described by
kernel_info = dict(version=kernel,
return self._afe.generate_control_file(
tests=[test], kernel=[kernel_info],
def _schedule_job(self, jobname, control, hosts):
control_type = ('Client', 'Server')[control.is_server]
self._afe.create_job(control.control_file, jobname,
control_type=control_type, hosts=hosts)
class email_action(base_action):
An action object to send emails about found new kernel versions.
_MAIL = 'sendmail'
def __init__(self, dest_addr, from_addr='autotest-server@localhost'):
Create an email_action instance.
@param dest_addr: a string or a list of strings with the destination
email address(es)
@param from_addr: optional source email address for the sent mails
(default 'autotest-server@localhost')
# if passed a string for the dest_addr convert it to a tuple
if type(dest_addr) is str:
self._dest_addr = (dest_addr,)
self._dest_addr = dest_addr
self._from_addr = from_addr
def __call__(self, kernel_list):
if not kernel_list:
message = '\n'.join(kernel_list)
message = 'Testing new kernel releases:\n%s' % message
self._mail('autotest new kernel notification', message)
def _mail(self, subject, message_text):
message = email.Message.Message()
message['To'] = ', '.join(self._dest_addr)
message['From'] = self._from_addr
message['Subject'] = subject
if self._sendmail(message.as_string()):
server = smtplib.SMTP('localhost')
server.sendmail(self._from_addr, self._dest_addr,
def _sendmail(cls, message):
Send an email using the sendmail command.
# open a pipe to the mail program and
# write the data to the pipe
p = os.popen('%s -t' % cls._MAIL, 'w')
return p.close()