| #!/usr/bin/python |
| |
| import unittest |
| import common |
| from autotest_lib.frontend import setup_django_environment |
| from autotest_lib.frontend import setup_test_environment |
| from autotest_lib.frontend.afe import model_logic, models as afe_models |
| from autotest_lib.frontend.planner import planner_test_utils, model_attributes |
| from autotest_lib.frontend.planner import models, rpc_utils, failure_actions |
| from autotest_lib.frontend.tko import models as tko_models |
| from autotest_lib.client.common_lib import utils, host_queue_entry_states |
| |
| |
| class RpcUtilsTest(unittest.TestCase, |
| planner_test_utils.PlannerTestMixin): |
| def setUp(self): |
| self._planner_common_setup() |
| |
| |
| def tearDown(self): |
| self._planner_common_teardown() |
| |
| |
| def test_create_plan_label(self): |
| label, group = self._create_label_helper() |
| |
| label.delete() |
| group.invalid = True |
| group.save() |
| |
| label, group = self._create_label_helper() |
| |
| self.assertRaises(model_logic.ValidationError, |
| rpc_utils.create_plan_label, self._plan) |
| |
| |
| def _create_label_helper(self): |
| label = rpc_utils.create_plan_label(self._plan) |
| group = afe_models.AtomicGroup.objects.get( |
| name=rpc_utils.PLANNER_ATOMIC_GROUP_NAME) |
| self.assertFalse(group.invalid) |
| self.assertEqual(label.atomic_group, group) |
| |
| return (label, group) |
| |
| |
| def test_lazy_load(self): |
| self.god.stub_function(utils, 'read_file') |
| |
| DUMMY_PATH_1 = object() |
| DUMMY_PATH_2 = object() |
| DUMMY_FILE_1 = object() |
| DUMMY_FILE_2 = object() |
| |
| utils.read_file.expect_call(DUMMY_PATH_1).and_return(DUMMY_FILE_1) |
| self.assertEqual(DUMMY_FILE_1, rpc_utils.lazy_load(DUMMY_PATH_1)) |
| self.god.check_playback() |
| |
| # read_file should not be called again for this path |
| self.assertEqual(DUMMY_FILE_1, rpc_utils.lazy_load(DUMMY_PATH_1)) |
| self.god.check_playback() |
| |
| # new file; read_file must be called again |
| utils.read_file.expect_call(DUMMY_PATH_2).and_return(DUMMY_FILE_2) |
| self.assertEqual(DUMMY_FILE_2, rpc_utils.lazy_load(DUMMY_PATH_2)) |
| self.god.check_playback() |
| |
| |
| def test_update_hosts_table(self): |
| label = self.labels[3] |
| default_hosts = set(self._plan.hosts.all()) |
| |
| rpc_utils.update_hosts_table(self._plan) |
| self.assertEqual(default_hosts, set(self._plan.hosts.all())) |
| self.assertEqual(set(), self._get_added_by_label_hosts()) |
| |
| self._plan.host_labels.add(label) |
| rpc_utils.update_hosts_table(self._plan) |
| self.assertEqual(default_hosts.union(label.host_set.all()), |
| set(self._plan.hosts.all())) |
| self.assertEqual(set(label.host_set.all()), |
| self._get_added_by_label_hosts()) |
| |
| self._plan.host_labels.remove(label) |
| rpc_utils.update_hosts_table(self._plan) |
| self.assertEqual(default_hosts, set(self._plan.hosts.all())) |
| self.assertEqual(set(), self._get_added_by_label_hosts()) |
| |
| |
| def _get_added_by_label_hosts(self): |
| return set(host.host for host in models.Host.objects.filter( |
| plan=self._plan, added_by_label=True)) |
| |
| |
| def test_compute_next_test_config(self): |
| self._setup_active_plan() |
| test_config = models.TestConfig.objects.create( |
| plan=self._plan, alias='config2', control_file=self._control, |
| execution_order=2, estimated_runtime=1) |
| |
| self.assertEqual(1, self._afe_job.hostqueueentry_set.count()) |
| self.assertEqual( |
| None, rpc_utils.compute_next_test_config(self._plan, |
| self._planner_host)) |
| self.assertFalse(self._planner_host.complete) |
| |
| hqe = self._afe_job.hostqueueentry_set.all()[0] |
| hqe.status = host_queue_entry_states.Status.COMPLETED |
| hqe.save() |
| |
| self.assertEqual( |
| test_config, |
| rpc_utils.compute_next_test_config(self._plan, |
| self._planner_host)) |
| self.assertFalse(self._planner_host.complete) |
| |
| afe_job = self._create_job(hosts=(1,)) |
| planner_job = models.Job.objects.create(plan=self._plan, |
| test_config=test_config, |
| afe_job=afe_job) |
| |
| self.assertEqual( |
| None, rpc_utils.compute_next_test_config(self._plan, |
| self._planner_host)) |
| self.assertFalse(self._planner_host.complete) |
| |
| hqe = afe_job.hostqueueentry_set.all()[0] |
| hqe.status = host_queue_entry_states.Status.COMPLETED |
| hqe.save() |
| |
| self.assertEqual( |
| None, rpc_utils.compute_next_test_config(self._plan, |
| self._planner_host)) |
| self.assertTrue(self._planner_host.complete) |
| |
| |
| def test_process_failure(self): |
| self._setup_active_plan() |
| tko_test = tko_models.Test.objects.create(job=self._tko_job, |
| machine=self._tko_machine, |
| kernel=self._tko_kernel, |
| status=self._running_status) |
| failure = models.TestRun.objects.create( |
| plan=self._plan, |
| test_job=self._planner_job, |
| tko_test=tko_test, |
| host=self._planner_host, |
| status=model_attributes.TestRunStatus.FAILED, |
| finalized=True, seen=False, triaged=False) |
| host_action = failure_actions.HostAction.UNBLOCK |
| test_action = failure_actions.TestAction.SKIP |
| labels = ['label1', 'label2'] |
| keyvals = {'key1': 'value1', |
| 'key2': 'value2'} |
| bugs = ['bug1', 'bug2'] |
| reason = 'overriden reason' |
| invalidate = True |
| |
| self.god.stub_function(rpc_utils, '_process_host_action') |
| self.god.stub_function(rpc_utils, '_process_test_action') |
| |
| rpc_utils._process_host_action.expect_call(self._planner_host, |
| host_action) |
| rpc_utils._process_test_action.expect_call(self._planner_job, |
| test_action) |
| |
| rpc_utils.process_failure( |
| failure_id=failure.id, host_action=host_action, |
| test_action=test_action, labels=labels, keyvals=keyvals, |
| bugs=bugs, reason=reason, invalidate=invalidate) |
| failure = models.TestRun.objects.get(id=failure.id) |
| |
| self.assertEqual( |
| set(failure.tko_test.testlabel_set.all()), |
| set(tko_models.TestLabel.objects.filter(name__in=labels))) |
| self.assertEqual( |
| set(failure.tko_test.job.jobkeyval_set.all()), |
| set(tko_models.JobKeyval.objects.filter( |
| key__in=keyvals.iterkeys()))) |
| self.assertEqual(set(failure.bugs.all()), |
| set(models.Bug.objects.filter(external_uid__in=bugs))) |
| self.assertEqual(failure.tko_test.reason, reason) |
| self.assertEqual(failure.invalidated, invalidate) |
| self.assertTrue(failure.seen) |
| self.assertTrue(failure.triaged) |
| self.god.check_playback() |
| |
| |
| def _replace_site_process_host_action(self, replacement): |
| self.god.stub_function(utils, 'import_site_function') |
| utils.import_site_function.expect_any_call().and_return(replacement) |
| |
| |
| def _remove_site_process_host_action(self): |
| def _site_process_host_action_dummy(host, action): |
| return False |
| self._replace_site_process_host_action(_site_process_host_action_dummy) |
| |
| |
| def test_process_host_action_block(self): |
| self._remove_site_process_host_action() |
| host = models.Host.objects.create(plan=self._plan, host=self.hosts[0], |
| blocked=False) |
| assert not host.blocked |
| |
| rpc_utils._process_host_action(host, failure_actions.HostAction.BLOCK) |
| host = models.Host.objects.get(id=host.id) |
| |
| self.assertTrue(host.blocked) |
| self.god.check_playback() |
| |
| |
| def test_process_host_action_unblock(self): |
| self._remove_site_process_host_action() |
| host = models.Host.objects.create(plan=self._plan, host=self.hosts[0], |
| blocked=True) |
| assert host.blocked |
| |
| rpc_utils._process_host_action(host, failure_actions.HostAction.UNBLOCK) |
| host = models.Host.objects.get(id=host.id) |
| |
| self.assertFalse(host.blocked) |
| self.god.check_playback() |
| |
| |
| def test_process_host_action_site(self): |
| self._remove_site_process_host_action |
| action = object() |
| failure_actions.HostAction.values.append(action) |
| host = models.Host.objects.create(plan=self._plan, host=self.hosts[0]) |
| |
| self.assertRaises(AssertionError, rpc_utils._process_host_action, |
| host, action) |
| self.god.check_playback() |
| |
| self._called = False |
| def _site_process_host_action(host, action): |
| self._called = True |
| return True |
| self._replace_site_process_host_action(_site_process_host_action) |
| |
| rpc_utils._process_host_action(host, action) |
| |
| self.assertTrue(self._called) |
| self.god.check_playback() |
| |
| |
| def test_process_test_action_skip(self): |
| self._setup_active_plan() |
| planner_job = self._planner_job |
| assert not planner_job.requires_rerun |
| |
| rpc_utils._process_test_action(planner_job, |
| failure_actions.TestAction.SKIP) |
| planner_job = models.Job.objects.get(id=planner_job.id) |
| |
| self.assertFalse(planner_job.requires_rerun) |
| |
| |
| def test_process_test_action_rerun(self): |
| self._setup_active_plan() |
| planner_job = self._planner_job |
| assert not planner_job.requires_rerun |
| |
| rpc_utils._process_test_action(planner_job, |
| failure_actions.TestAction.RERUN) |
| planner_job = models.Job.objects.get(id=planner_job.id) |
| |
| self.assertTrue(planner_job.requires_rerun) |
| |
| |
| def test_set_additional_parameters(self): |
| hostname_regex = 'host[0-9]' |
| param_type = model_attributes.AdditionalParameterType.VERIFY |
| param_values = {'key1': 'value1', |
| 'key2': []} |
| |
| additional_parameters = {'hostname_regex': hostname_regex, |
| 'param_type': param_type, |
| 'param_values': param_values} |
| |
| rpc_utils.set_additional_parameters(self._plan, [additional_parameters]) |
| |
| additional_parameters_query = ( |
| models.AdditionalParameter.objects.filter(plan=self._plan)) |
| self.assertEqual(additional_parameters_query.count(), 1) |
| |
| additional_parameter = additional_parameters_query[0] |
| self.assertEqual(additional_parameter.hostname_regex, hostname_regex) |
| self.assertEqual(additional_parameter.param_type, param_type) |
| self.assertEqual(additional_parameter.application_order, 0) |
| |
| values_query = additional_parameter.additionalparametervalue_set.all() |
| self.assertEqual(values_query.count(), 2) |
| |
| value_query1 = values_query.filter(key='key1') |
| value_query2 = values_query.filter(key='key2') |
| self.assertEqual(value_query1.count(), 1) |
| self.assertEqual(value_query2.count(), 1) |
| |
| self.assertEqual(value_query1[0].value, repr('value1')) |
| self.assertEqual(value_query2[0].value, repr([])) |
| |
| |
| def test_get_wrap_arguments(self): |
| hostname_regex = '.*' |
| param_type = model_attributes.AdditionalParameterType.VERIFY |
| |
| additional_param = models.AdditionalParameter.objects.create( |
| plan=self._plan, hostname_regex=hostname_regex, |
| param_type=param_type, application_order=0) |
| models.AdditionalParameterValue.objects.create( |
| additional_parameter=additional_param, |
| key='key1', value=repr('value1')) |
| models.AdditionalParameterValue.objects.create( |
| additional_parameter=additional_param, |
| key='key2', value=repr([])) |
| |
| actual = rpc_utils.get_wrap_arguments(self._plan, 'host', param_type) |
| expected = {'key1': repr('value1'), |
| 'key2': repr([])} |
| |
| self.assertEqual(actual, expected) |
| |
| |
| def test_compute_test_config_status_scheduled(self): |
| self._setup_active_plan() |
| self._planner_job.delete() |
| |
| self.assertEqual( |
| rpc_utils.compute_test_config_status(self._planner_host), |
| rpc_utils.ComputeTestConfigStatusResult.SCHEDULED) |
| |
| |
| def test_compute_test_config_status_running(self): |
| self._setup_active_plan() |
| self.god.stub_function(models.Job, 'active') |
| models.Job.active.expect_call().and_return(True) |
| |
| self.assertEqual( |
| rpc_utils.compute_test_config_status(self._planner_host), |
| rpc_utils.ComputeTestConfigStatusResult.RUNNING) |
| self.god.check_playback() |
| |
| |
| def test_compute_test_config_status_good(self): |
| self._setup_active_plan() |
| tko_test = self._tko_job.test_set.create(kernel=self._tko_kernel, |
| status=self._good_status, |
| machine=self._tko_machine) |
| self._plan.testrun_set.create(test_job=self._planner_job, |
| tko_test=tko_test, |
| host=self._planner_host) |
| self._planner_host.complete = True |
| self._planner_host.save() |
| self.god.stub_function(models.Job, 'active') |
| models.Job.active.expect_call().and_return(False) |
| |
| self.assertEqual( |
| rpc_utils.compute_test_config_status(self._planner_host), |
| rpc_utils.ComputeTestConfigStatusResult.PASS) |
| self.god.check_playback() |
| |
| |
| def test_compute_test_config_status_bad(self): |
| self._setup_active_plan() |
| tko_test = self._tko_job.test_set.create(kernel=self._tko_kernel, |
| status=self._fail_status, |
| machine=self._tko_machine) |
| self._plan.testrun_set.create(test_job=self._planner_job, |
| tko_test=tko_test, |
| host=self._planner_host) |
| self._planner_host.complete = True |
| self._planner_host.save() |
| self.god.stub_function(models.Job, 'active') |
| models.Job.active.expect_call().and_return(False) |
| |
| self.assertEqual( |
| rpc_utils.compute_test_config_status(self._planner_host), |
| rpc_utils.ComputeTestConfigStatusResult.FAIL) |
| self.god.check_playback() |
| |
| |
| if __name__ == '__main__': |
| unittest.main() |
