shaba/awx
1
0
Fork 0
mirror of https://github.com/ansible/awx.git synced 2024-11-02 09:51:09 +03:00
Code Releases Activity

Unit test updates for task system... remove old monkeypatch procedure

Browse Source 
for getting job args in favor of using the job info from the database.
Can't do this anymore anyway since the job is running in another process
This commit is contained in:
Matthew Jones 2014-03-13 13:25:45 -04:00
parent fe4b4043db
commit c2950fdbdf
3 changed files with 23 additions and 32 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
awx/main/management/commands/run_task_system.py
View File

@ -188,6 +188,7 @@ def rebuild_graph(message):
dep.status = 'waiting'
dep.save()
waiting_tasks.insert(waiting_tasks.index(task), dep)
if not hasattr(settings, 'CELERY_UNIT_TEST'):
task.status = 'waiting'
task.save()
@ -242,6 +243,7 @@ def process_graph(graph, task_capacity):
continue
remaining_volume -= impact
running_impact += impact
if settings.DEBUG:
print("Started Node: %s (capacity hit: %s) Remaining Capacity: %s" % (str(node_obj), str(impact), str(remaining_volume)))
def run_taskmanager(command_port):

5
awx/main/tests/projects.py
View File

@ -1012,7 +1012,7 @@ class ProjectUpdatesTest(BaseTransactionTest):
if project.scm_type:
self.assertTrue(project.last_update)
self.check_project_update(project,
project_udpate=project.last_update)
project_update=project.last_update)
self.assertTrue(os.path.exists(project_path))
else:
self.assertFalse(os.path.exists(project_path))
@ -1567,8 +1567,7 @@ class ProjectUpdatesTest(BaseTransactionTest):
job = self.create_test_job(job_template=job_template)
self.assertEqual(job.status, 'new')
self.assertFalse(job.passwords_needed_to_start)
self.assertTrue(job.start())
self.assertEqual(job.status, 'waiting')
self.assertTrue(job.signal_start())
job = Job.objects.get(pk=job.pk)
self.assertTrue(job.status in ('successful', 'failed'))
self.assertEqual(self.project.project_updates.count(), 3)

42
awx/main/tests/tasks.py
View File

@ -177,16 +177,6 @@ class RunJobTest(BaseCeleryTest):
self.project = None
self.credential = None
self.cloud_credential = None
# Monkeypatch RunJob to capture list of command line arguments.
self.original_build_args = RunJob.build_args
self.run_job_args = None
self.build_args_callback = lambda: None
def new_build_args(_self, job, **kw):
args = self.original_build_args(_self, job, **kw)
self.run_job_args = args
self.build_args_callback()
return args
RunJob.build_args = new_build_args
settings.INTERNAL_API_URL = self.live_server_url
self.start_taskmanager(settings.TASK_COMMAND_PORT)
if settings.CALLBACK_CONSUMER_PORT:
@ -684,9 +674,9 @@ class RunJobTest(BaseCeleryTest):
self.assertTrue(job.signal_start())
job = Job.objects.get(pk=job.pk)
self.check_job_result(job, 'successful')
self.assertTrue('--forks=3' in self.run_job_args)
self.assertTrue('-vv' in self.run_job_args)
self.assertTrue('-e' in self.run_job_args)
self.assertTrue('--forks=3' in job.job_args)
self.assertTrue('-vv' in job.job_args)
self.assertTrue('-e' in job.job_args)
# Test with extra_vars as key=value (old format).
job_template2 = self.create_test_job_template(extra_vars='foo=1')
job2 = self.create_test_job(job_template=job_template2)
@ -713,7 +703,7 @@ class RunJobTest(BaseCeleryTest):
job = Job.objects.get(pk=job.pk)
self.assertTrue(len(job.job_args) > 1024)
self.check_job_result(job, 'successful')
self.assertTrue('-e' in self.run_job_args)
self.assertTrue('-e' in job.job_args)
def test_limit_option(self):
self.create_test_project(TEST_PLAYBOOK)
@ -724,7 +714,7 @@ class RunJobTest(BaseCeleryTest):
self.assertTrue(job.signal_start())
job = Job.objects.get(pk=job.pk)
self.check_job_result(job, 'failed')
self.assertTrue('-l' in self.run_job_args)
self.assertTrue('-l' in job.job_args)
def test_limit_option_with_group_pattern_and_ssh_agent(self):
self.create_test_credential(ssh_key_data=TEST_SSH_KEY_DATA)
@ -736,7 +726,7 @@ class RunJobTest(BaseCeleryTest):
self.assertTrue(job.signal_start())
job = Job.objects.get(pk=job.pk)
self.check_job_result(job, 'successful')
self.assertTrue('ssh-agent' in self.run_job_args)
self.assertTrue('ssh-agent' in job.job_args)
def test_ssh_username_and_password(self):
self.create_test_credential(username='sshuser', password='sshpass')
@ -748,8 +738,8 @@ class RunJobTest(BaseCeleryTest):
self.assertTrue(job.signal_start())
job = Job.objects.get(pk=job.pk)
self.check_job_result(job, 'successful')
self.assertTrue('-u' in self.run_job_args)
self.assertTrue('--ask-pass' in self.run_job_args)
self.assertTrue('-u' in job.job_args)
self.assertTrue('--ask-pass' in job.job_args)
def test_ssh_ask_password(self):
self.create_test_credential(password='ASK')
@ -764,7 +754,7 @@ class RunJobTest(BaseCeleryTest):
self.assertTrue(job.signal_start(ssh_password='sshpass'))
job = Job.objects.get(pk=job.pk)
self.check_job_result(job, 'successful')
self.assertTrue('--ask-pass' in self.run_job_args)
self.assertTrue('--ask-pass' in job.job_args)
def test_sudo_username_and_password(self):
self.create_test_credential(sudo_username='sudouser',
@ -779,8 +769,8 @@ class RunJobTest(BaseCeleryTest):
# Job may fail if current user doesn't have password-less sudo
# privileges, but we're mainly checking the command line arguments.
self.check_job_result(job, ('successful', 'failed'))
self.assertTrue('-U' in self.run_job_args)
self.assertTrue('--ask-sudo-pass' in self.run_job_args)
self.assertTrue('-U' in job.job_args)
self.assertTrue('--ask-sudo-pass' in job.job_args)
def test_sudo_ask_password(self):
self.create_test_credential(sudo_password='ASK')
@ -796,7 +786,7 @@ class RunJobTest(BaseCeleryTest):
# Job may fail if current user doesn't have password-less sudo
# privileges, but we're mainly checking the command line arguments.
self.assertTrue(job.status in ('successful', 'failed'))
self.assertTrue('--ask-sudo-pass' in self.run_job_args)
self.assertTrue('--ask-sudo-pass' in job.job_args)
def test_unlocked_ssh_key(self):
self.create_test_credential(ssh_key_data=TEST_SSH_KEY_DATA)
@ -808,7 +798,7 @@ class RunJobTest(BaseCeleryTest):
self.assertTrue(job.signal_start())
job = Job.objects.get(pk=job.pk)
self.check_job_result(job, 'successful')
self.assertTrue('ssh-agent' in self.run_job_args)
self.assertTrue('ssh-agent' in job.job_args)
def test_locked_ssh_key_with_password(self):
self.create_test_credential(ssh_key_data=TEST_SSH_KEY_DATA_LOCKED,
@ -821,7 +811,7 @@ class RunJobTest(BaseCeleryTest):
self.assertTrue(job.signal_start())
job = Job.objects.get(pk=job.pk)
self.check_job_result(job, 'successful')
self.assertTrue('ssh-agent' in self.run_job_args)
self.assertTrue('ssh-agent' in job.job_args)
self.assertTrue('Bad passphrase' not in job.result_stdout)
def test_locked_ssh_key_with_bad_password(self):
@ -835,7 +825,7 @@ class RunJobTest(BaseCeleryTest):
self.assertTrue(job.signal_start())
job = Job.objects.get(pk=job.pk)
self.check_job_result(job, 'failed')
self.assertTrue('ssh-agent' in self.run_job_args)
self.assertTrue('ssh-agent' in job.job_args)
self.assertTrue('Bad passphrase' in job.result_stdout)
def test_locked_ssh_key_ask_password(self):
@ -852,7 +842,7 @@ class RunJobTest(BaseCeleryTest):
self.assertTrue(job.signal_start(ssh_key_unlock=TEST_SSH_KEY_DATA_UNLOCK))
job = Job.objects.get(pk=job.pk)
self.check_job_result(job, 'successful')
self.assertTrue('ssh-agent' in self.run_job_args)
self.assertTrue('ssh-agent' in job.job_args)
self.assertTrue('Bad passphrase' not in job.result_stdout)
def _test_cloud_credential_environment_variables(self, kind):