blob: 76aa4a264492170da576cfae28d4c84caa05f6de [file] [log] [blame]
import re
from django.db import models as dbmodels
import common
from autotest_lib.frontend.afe import models as afe_models
from autotest_lib.frontend.afe import model_logic, rpc_utils
from autotest_lib.frontend.tko import models as tko_models
from autotest_lib.frontend.planner import model_attributes
from autotest_lib.client.common_lib import utils, host_queue_entry_states
class Plan(dbmodels.Model, model_logic.ModelExtensions):
"""A test plan
name: Plan name, unique
complete: True if the plan is completed
dirty: True if the plan has been changed since the execution engine has
last seen it
initialized: True if the plan has started
label_override: A label to apply to each Autotest job.
support: The global support script to apply to this plan
name = dbmodels.CharField(max_length=255, unique=True)
label_override = dbmodels.CharField(max_length=255, null=True, blank=True)
support = dbmodels.TextField(blank=True)
complete = dbmodels.BooleanField(default=False)
dirty = dbmodels.BooleanField(default=False)
initialized = dbmodels.BooleanField(default=False)
owners = dbmodels.ManyToManyField(afe_models.User,
hosts = dbmodels.ManyToManyField(afe_models.Host, through='Host')
host_labels = dbmodels.ManyToManyField(afe_models.Label,
name_field = 'name'
class Meta:
db_table = 'planner_plans'
def __unicode__(self):
return unicode(
class ModelWithPlan(dbmodels.Model):
"""Superclass for models that have a plan_id
plan: The associated test plan
plan = dbmodels.ForeignKey(Plan)
class Meta:
abstract = True
def __unicode__(self):
return u'%s (%s)' % (self._get_details_unicode(),
def _get_details_unicode(self):
"""Gets the first part of the unicode string
subclasses must override this method
raise NotImplementedError(
'Subclasses must override _get_details_unicode()')
class Host(ModelWithPlan, model_logic.ModelExtensions):
"""A plan host
host: The AFE host
complete: True if and only if this host is finished in the test plan
blocked: True if and only if the host is blocked (not executing tests)
added_by_label: True if and only if the host was added because of a host
label (as opposed to being explicitly added)
host = dbmodels.ForeignKey(afe_models.Host)
complete = dbmodels.BooleanField(default=False)
blocked = dbmodels.BooleanField(default=False)
added_by_label = dbmodels.BooleanField(default=False)
class Meta:
db_table = 'planner_hosts'
def status(self):
if self.complete:
return model_attributes.HostStatus.FINISHED
if self.blocked:
return model_attributes.HostStatus.BLOCKED
return model_attributes.HostStatus.RUNNING
def _get_details_unicode(self):
return 'Host: %s' %
class ControlFile(model_logic.ModelWithHash,
"""A control file. Immutable once added to the table
contents: The text of the control file
the_hash: The SHA1 hash of the control file, for duplicate detection
and fast search
contents = dbmodels.TextField()
class Meta:
db_table = 'planner_test_control_files'
def _compute_hash(cls, **kwargs):
return utils.hash('sha1', kwargs['contents']).hexdigest()
def __unicode__(self):
return u'Control file id %s (SHA1: %s)' % (, self.control_hash)
class TestConfig(ModelWithPlan, model_logic.ModelExtensions):
"""A configuration for a planned test
alias: The name to give this test within the plan. Unique with plan id
control_file: The control file to run
is_server: True if this control file is a server-side test
execution_order: An integer describing when this test should be run in
the test plan
estimated_runtime: Time in hours that the test is expected to run. Will
be automatically generated (on the frontend) for
tests in Autotest.
skipped_hosts: Hosts that are going to skip this test.
alias = dbmodels.CharField(max_length=255)
control_file = dbmodels.ForeignKey(ControlFile)
is_server = dbmodels.BooleanField(default=True)
execution_order = dbmodels.IntegerField(blank=True)
estimated_runtime = dbmodels.IntegerField()
skipped_hosts = dbmodels.ManyToManyField(
afe_models.Host, db_table='planner_test_configs_skipped_hosts')
class Meta:
db_table = 'planner_test_configs'
ordering = ('execution_order',)
unique_together = (('plan', 'alias'),)
def _get_details_unicode(self):
return 'Planned test config - Control file id %s' %
class Job(ModelWithPlan, model_logic.ModelExtensions):
"""Represents an Autotest job initiated for a test plan
test: The TestConfig associated with this Job
afe_job: The Autotest job
test_config = dbmodels.ForeignKey(TestConfig)
afe_job = dbmodels.ForeignKey(afe_models.Job)
requires_rerun = dbmodels.BooleanField(default=False)
class Meta:
db_table = 'planner_test_jobs'
def active(self):
for hqe in self.afe_job.hostqueueentry_set.all():
if not hqe.complete:
return True
return False
def all_tests_passed(self):
return False
Status = host_queue_entry_states.Status
if self.afe_job.hostqueueentry_set.exclude(status=Status.COMPLETED):
return False
tko_tests = tko_models.Test.objects.filter(
for tko_test in tko_tests:
if tko_test.status.word != 'GOOD':
return False
return True
def _get_details_unicode(self):
return 'AFE job %s' %
class Bug(dbmodels.Model):
"""Represents a bug ID
external_uid: External unique ID for the bug
external_uid = dbmodels.CharField(max_length=255, unique=True)
class Meta:
db_table = 'planner_bugs'
def __unicode__(self):
return u'Bug external ID %s' % self.external_uid
class TestRun(ModelWithPlan, model_logic.ModelExtensions):
"""An individual test run from an Autotest job for the test plan.
Each Job object may have multiple TestRun objects associated with it.
test_job: The Job object associated with this TestRun
tko_test: The TKO Test associated with this TestRun
status: One of 'Active', 'Passed', 'Failed'
finalized: True if and only if the TestRun is ready to be shown in
invalidated: True if and only if a user has decided to invalidate this
TestRun's results
seen: True if and only if a user has marked this TestRun as "seen"
triaged: True if and only if the TestRun no longer requires any user
bugs: Bugs filed that a relevant to this run
test_job = dbmodels.ForeignKey(Job)
tko_test = dbmodels.ForeignKey(tko_models.Test)
host = dbmodels.ForeignKey(Host)
status = dbmodels.CharField(
finalized = dbmodels.BooleanField(default=False)
seen = dbmodels.BooleanField(default=False)
triaged = dbmodels.BooleanField(default=False)
invalidated = dbmodels.BooleanField(default=False)
bugs = dbmodels.ManyToManyField(Bug, null=True,
class Meta:
db_table = 'planner_test_runs'
unique_together = (('plan', 'test_job', 'tko_test', 'host'),)
def _get_details_unicode(self):
return 'Test Run: %s' %
class DataType(dbmodels.Model):
"""Encodes the data model types
For use in the history table, to identify the type of object that was
name: The name of the data type
db_table: The name of the database table that stores this type
name = dbmodels.CharField(max_length=255)
db_table = dbmodels.CharField(max_length=255)
class Meta:
db_table = 'planner_data_types'
def __unicode__(self):
return u'Data type %s (stored in table %s)' % (, self.db_table)
class History(ModelWithPlan):
"""Represents a history action
action_id: An arbitrary ID that uniquely identifies the user action
related to the history entry. One user action may result in
multiple history entries
user: The user who initiated the change
data_type: The type of object that was changed
object_id: Value of the primary key field for the changed object
old_object_repr: A string representation of the object before the change
new_object_repr: A string representation of the object after the change
time: A timestamp. Automatically generated.
action_id = dbmodels.IntegerField()
user = dbmodels.ForeignKey(afe_models.User)
data_type = dbmodels.ForeignKey(DataType)
object_id = dbmodels.IntegerField()
old_object_repr = dbmodels.TextField(blank=True)
new_object_repr = dbmodels.TextField(blank=True)
time = dbmodels.DateTimeField(auto_now_add=True)
class Meta:
db_table = 'planner_history'
def _get_details_unicode(self):
return 'History entry: %s => %s' % (self.old_object_repr,
class SavedObject(dbmodels.Model):
"""A saved object that can be recalled at certain points in the UI
user: The creator of the object
object_type: One of 'support', 'triage', 'autoprocess', 'custom_query'
name: The name given to the object
encoded_object: The actual object
user = dbmodels.ForeignKey(afe_models.User)
object_type = dbmodels.CharField(
name = dbmodels.CharField(max_length=255)
encoded_object = dbmodels.TextField()
class Meta:
db_table = 'planner_saved_objects'
unique_together = ('user', 'object_type', 'name')
def __unicode__(self):
return u'Saved %s object: %s, by %s' % (self.object_type,,
class CustomQuery(ModelWithPlan):
"""A custom SQL query for the triage page
query: the SQL WHERE clause to attach to the main query
query = dbmodels.TextField()
class Meta:
db_table = 'planner_custom_queries'
def _get_details_unicode(self):
return 'Custom Query: %s' % self.query
class KeyVal(model_logic.ModelWithHash):
"""Represents a keyval. Immutable once added to the table.
key: The key
value: The value
the_hash: The result of SHA1(SHA1(key) ++ value), for duplicate
detection and fast search.
key = dbmodels.CharField(max_length=1024)
value = dbmodels.CharField(max_length=1024)
class Meta:
db_table = 'planner_keyvals'
def _compute_hash(cls, **kwargs):
round1 = utils.hash('sha1', kwargs['key']).hexdigest()
return utils.hash('sha1', round1 + kwargs['value']).hexdigest()
def __unicode__(self):
return u'Keyval: %s = %s' % (self.key, self.value)
class AutoProcess(ModelWithPlan):
"""An autoprocessing directive to perform on test runs that enter triage
condition: A SQL WHERE clause. The autoprocessing will be applied if the
test run matches this condition
enabled: If this is False, this autoprocessing entry will not be applied
labels: Labels to apply to the TKO test
keyvals: Keyval overrides to apply to the TKO test
bugs: Bugs filed that a relevant to this run
reason_override: Override for the AFE reason
condition = dbmodels.TextField()
enabled = dbmodels.BooleanField(default=False)
labels = dbmodels.ManyToManyField(tko_models.TestLabel, null=True,
keyvals = dbmodels.ManyToManyField(KeyVal, null=True,
bugs = dbmodels.ManyToManyField(Bug, null=True,
reason_override = dbmodels.CharField(max_length=255, null=True, blank=True)
class Meta:
db_table = 'planner_autoprocess'
def _get_details_unicode(self):
return 'Autoprocessing condition: %s' % self.condition
class AdditionalParameter(ModelWithPlan):
Allows parameters to be passed to the execution engine for test configs
If this object matches a hostname by regex, it will apply the associated
parameters at their applicable locations.
hostname_regex: A regular expression, for matching on the hostname
param_type: Currently only 'Verify' (and site-specific values) allowed
application_order: The order in which to apply this parameter.
Parameters are attempted in the order specified here,
and stop when the first match is found
hostname_regex = dbmodels.CharField(max_length=255)
param_type = dbmodels.CharField(
application_order = dbmodels.IntegerField(blank=True)
class Meta:
db_table = 'planner_additional_parameters'
unique_together = ('plan', 'hostname_regex', 'param_type')
def find_applicable_additional_parameter(cls, plan, hostname, param_type):
Finds the first AdditionalParameter that matches the given arguments
params = cls.objects.filter(
plan=plan, param_type=param_type).order_by('application_order')
for param in params:
if re.match(param.hostname_regex, hostname):
return param
return None
def _get_details_unicode(self):
return 'Additional %s parameters, regex: %s' % (self.param_type,
class AdditionalParameterValue(dbmodels.Model):
The actual values for the additional parameters
additional_parameter: The associated AdditionalParameter
key: The name of the parameter
value: The value of the parameter
additional_parameter = dbmodels.ForeignKey(AdditionalParameter)
key = dbmodels.CharField(max_length=255)
value = dbmodels.CharField(max_length=255)
class Meta:
db_table = 'planner_additional_parameter_values'
unique_together = ('additional_parameter', 'key')
def __unicode__(self):
return u'Value for parameter %d: %s=%s' % (,
self.key, self.value)