| import email.Message, 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: |
| action(files) |
| |
| |
| def add_action(self, func): |
| self.__actions.append(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 NotImplemented('__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', |
| upload_kernel_config=False): |
| """ |
| 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 |
| version |
| @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) |
| |
| |
| @classmethod |
| 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) |
| |
| |
| @classmethod |
| 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) |
| |
| |
| @classmethod |
| 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) |
| l.append(kver) |
| l.sort(cmp=cls._kver_cmp) |
| 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 |
| database. |
| @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 |
| client |
| |
| @returns a dict representing a control file as described by |
| frontend.afe.rpc_interface.generate_control_file |
| """ |
| kernel_info = dict(version=kernel, |
| config_file=os.path.expanduser(kernel_config)) |
| return self._afe.generate_control_file( |
| tests=[test], kernel=[kernel_info], |
| upload_kernel_config=self._upload_kernel_config) |
| |
| |
| 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,) |
| else: |
| self._dest_addr = dest_addr |
| |
| self._from_addr = from_addr |
| |
| |
| def __call__(self, kernel_list): |
| if not kernel_list: |
| return |
| |
| 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 |
| message.set_payload(message_text) |
| |
| if self._sendmail(message.as_string()): |
| server = smtplib.SMTP('localhost') |
| try: |
| server.sendmail(self._from_addr, self._dest_addr, |
| message.as_string()) |
| finally: |
| server.quit() |
| |
| |
| @classmethod |
| 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') |
| p.write(message) |
| return p.close() |