Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More

How to use aborted_on method in autotest

Best Python code snippet using autotest_python

jobx.py

Source:jobx.py

1# Copyright 2017 The Chromium Authors. All rights reserved.2# Use of this source code is governed by a BSD-style license that can be3# found in the LICENSE file.4"""Extra functions for frontend.afe.models.Job objects.5Most of these exist in tightly coupled forms in legacy Autotest code6(e.g., part of methods with completely unrelated names on Task objects7under multiple layers of abstract classes).  These are defined here to8sanely reuse without having to commit to a long refactor of legacy code9that is getting deleted soon.10It's not really a good idea to define these on the Job class either;11they are specialized and the Job class already suffers from method12bloat.13"""14from __future__ import absolute_import15from __future__ import division16from __future__ import print_function17import os18import time19import urllib20from lucifer import autotest21from lucifer import results22def is_hostless(job):23    """Return True if the job is hostless.24    @param job: frontend.afe.models.Job instance25    """26    return not hostnames(job)27def hostnames(job):28    """Return a list of hostnames for a job.29    @param job: frontend.afe.models.Job instance30    """31    hqes = job.hostqueueentry_set.all().prefetch_related('host')32    return [hqe.host.hostname for hqe in hqes if hqe.host is not None]33def is_aborted(job):34    """Return if the job is aborted.35    (This means the job is marked for abortion; the job can still be36    running.)37    @param job: frontend.afe.models.Job instance38    """39    return job.hostqueueentry_set.filter(aborted=True).exists()40def is_server_job(job):41    """Return whether the job is a server job.42    @param job: frontend.afe.models.Job instance43    """44    return not is_client_job(job)45def is_client_job(job):46    """Return whether the job is a client job.47    If the job is not a client job, it is a server job.48    (In theory a job can be neither.  I have no idea what you should do49    in that case.)50    @param job: frontend.afe.models.Job instance51    """52    CONTROL_TYPE = autotest.load('client.common_lib.control_data').CONTROL_TYPE53    return CONTROL_TYPE.get_value(job.control_type) == CONTROL_TYPE.CLIENT54def needs_ssp(job):55    """Return whether the job needs SSP.56    This also looks up the config for jobs that do not have a value57    specified.58    @param job: frontend.afe.models.Job instance59    """60    return (_ssp_enabled()61            and is_server_job(job)62            # None is the same as True.63            and job.require_ssp != False)64def _ssp_enabled():65    """Return whether SSP is enabled in the config."""66    global_config = autotest.load('client.common_lib.global_config')67    return global_config.global_config.get_config_value(68            'AUTOSERV', 'enable_ssp_container', type=bool,69            default=True)70def control_file_path(workdir):71    """Path to control file for a job.72    This makes more sense in the old Autotest drone world.  The73    scheduler has to copy the control file to the drone.  It goes to a74    temporary path `drone_tmp/attach.N`.75    The drone is then able to run `autoserv <args> drone_tmp/attach.N`.76    But in the Lucifer world, we are already running on the drone, so we77    don't need to rendezvous with a temporary directory through78    drone_manager first.79    So pick an arbitrary filename to plop into the workdir.  autoserv80    will later copy this back to the standard control/control.srv.81    """82    return os.path.join(workdir, 'lucifer', 'control_attach')83def prepare_control_file(job, workdir):84    """Prepare control file for a job."""85    with open(control_file_path(workdir), 'w') as f:86        f.write(job.control_file)87def prepare_keyvals_files(job, workdir):88    """Prepare Autotest keyvals files for a job."""89    keyvals = job.keyval_dict()90    keyvals['job_queued'] = \91            int(time.mktime(job.created_on.timetuple()))92    results.write_keyvals(workdir, keyvals)93    if is_hostless(job):94        return95    for hqe in job.hostqueueentry_set.all().prefetch_related('host'):96        results.write_host_keyvals(97                workdir, hqe.host.hostname, _host_keyvals(hqe.host))98def write_aborted_keyvals_and_status(job, workdir):99    """Write the keyvals and status for an aborted job."""100    aborted_by = 'autotest_system'101    aborted_on = int(time.time())102    for hqe in job.hostqueueentry_set.all():103        if not hasattr(hqe, 'abortedhostqueueentry'):104            continue105        ahqe = hqe.abortedhostqueueentry106        aborted_by = ahqe.aborted_by107        aborted_on = int(time.mktime(ahqe.aborted_on.timetuple()))108        break109    results.write_keyvals(workdir, {110            'aborted_by': aborted_by,111            'aborted_on': aborted_on,112    })113    results.write_status_comment(114            workdir, 'Job aborted by %s on %s' % (aborted_by, aborted_on))115def _host_keyvals(host):116    """Return keyvals dict for a host.117    @param host: frontend.afe.models.Host instance118    """119    labels = list(_host_labels(host))120    platform = None121    for label in labels:122        if label.platform:123            platform = label.name124    return {125            'platform': platform,126            'labels': ','.join(urllib.quote(label.name) for label in labels),127    }128def _host_labels(host):129    """Return an iterable of labels for a host.130    @param host: frontend.afe.models.Host instance131    """132    if autotest.load('scheduler.scheduler_models').RESPECT_STATIC_LABELS:133        return _host_labels_with_static(host)134    else:135        return host.labels.all()136def _host_labels_with_static(host):137    """Return a generator of labels for a host, respecting static labels.138    @param host: frontend.afe.models.Host instance139    """140    models = autotest.load('frontend.afe.models')141    replaced_label_ids = frozenset(models.ReplacedLabel.objects.all()142                                   .values_list('label_id', flat=True))143    shadowed_labels = set()144    for label in host.labels.all():145        if label.id in replaced_label_ids:146            shadowed_labels.add(label.name)147        else:148            yield label149    for label in host.static_labels.all():150        if label.name in shadowed_labels:...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.