mirror of
https://github.com/ZwareBear/awx.git
synced 2026-05-16 03:48:37 -05:00
correctly inherit extra_vars
* inherit extra_vars from job_template * overwrite with extra vars survey default vars * overwrite with job extra_vars * bypass extra_vars inherit logic for job relaunch * with tests
This commit is contained in:
@@ -5,6 +5,7 @@ from __future__ import absolute_import
|
||||
|
||||
from .jobs_monolithic import * # noqa
|
||||
from .job_launch import * # noqa
|
||||
from .job_relaunch import * # noqa
|
||||
from .survey_password import * # noqa
|
||||
from .start_cancel import * # noqa
|
||||
from .base import * # noqa
|
||||
|
||||
@@ -0,0 +1,75 @@
|
||||
# Copyright (c) 2015 Ansible, Inc.
|
||||
# All Rights Reserved
|
||||
|
||||
# Python
|
||||
from __future__ import absolute_import
|
||||
import json
|
||||
|
||||
# Django
|
||||
from django.core.urlresolvers import reverse
|
||||
|
||||
# AWX
|
||||
from awx.main.models import * # noqa
|
||||
from awx.main.tests.base import BaseLiveServerTest
|
||||
from .base import BaseJobTestMixin
|
||||
|
||||
__all__ = ['JobRelaunchTest',]
|
||||
|
||||
class JobRelaunchTest(BaseJobTestMixin, BaseLiveServerTest):
|
||||
|
||||
def test_job_relaunch(self):
|
||||
job = self.make_job(self.jt_ops_east_run, self.user_sue, 'success')
|
||||
url = reverse('api:job_relaunch', args=(job.pk,))
|
||||
with self.current_user(self.user_sue):
|
||||
response = self.post(url, {}, expect=201)
|
||||
j = Job.objects.get(pk=response['job'])
|
||||
self.assertTrue(j.status == 'successful')
|
||||
self.assertEqual(j.launch_type, 'relaunch')
|
||||
# Test with a job that prompts for SSH and sudo passwords.
|
||||
job = self.make_job(self.jt_sup_run, self.user_sue, 'success')
|
||||
url = reverse('api:job_start', args=(job.pk,))
|
||||
with self.current_user(self.user_sue):
|
||||
response = self.get(url)
|
||||
self.assertEqual(set(response['passwords_needed_to_start']),
|
||||
set(['ssh_password', 'become_password']))
|
||||
data = dict()
|
||||
response = self.post(url, data, expect=400)
|
||||
data['ssh_password'] = 'sshpass'
|
||||
response = self.post(url, data, expect=400)
|
||||
data2 = dict(become_password='sudopass')
|
||||
response = self.post(url, data2, expect=400)
|
||||
data.update(data2)
|
||||
response = self.post(url, data, expect=202)
|
||||
job = Job.objects.get(pk=job.pk)
|
||||
|
||||
# Create jt with no extra_vars
|
||||
# Launch j1 with runtime extra_vars
|
||||
# Assign extra_vars to jt backing job
|
||||
# Relaunch j1
|
||||
# j2 should not contain jt extra_vars
|
||||
def test_relaunch_job_does_not_inherit_jt_extra_vars(self):
|
||||
jt_extra_vars = {
|
||||
"hello": "world"
|
||||
}
|
||||
j_extra_vars = {
|
||||
"goodbye": "cruel universe"
|
||||
}
|
||||
job = self.make_job(self.jt_ops_east_run, self.user_sue, 'success', extra_vars=j_extra_vars)
|
||||
url = reverse('api:job_relaunch', args=(job.pk,))
|
||||
with self.current_user(self.user_sue):
|
||||
response = self.post(url, {}, expect=201)
|
||||
j = Job.objects.get(pk=response['job'])
|
||||
self.assertTrue(j.status == 'successful')
|
||||
|
||||
self.jt_ops_east_run.extra_vars = jt_extra_vars
|
||||
self.jt_ops_east_run.save()
|
||||
|
||||
response = self.post(url, {}, expect=201)
|
||||
j = Job.objects.get(pk=response['job'])
|
||||
self.assertTrue(j.status == 'successful')
|
||||
|
||||
resp_extra_vars = json.loads(response['extra_vars'])
|
||||
self.assertNotIn("hello", resp_extra_vars)
|
||||
self.assertEqual(resp_extra_vars, j_extra_vars)
|
||||
|
||||
|
||||
@@ -115,30 +115,6 @@ class JobStartCancelTest(BaseJobTestMixin, BaseLiveServerTest):
|
||||
|
||||
# FIXME: Test with other users, test when passwords are required.
|
||||
|
||||
def test_job_relaunch(self):
|
||||
job = self.make_job(self.jt_ops_east_run, self.user_sue, 'success')
|
||||
url = reverse('api:job_relaunch', args=(job.pk,))
|
||||
with self.current_user(self.user_sue):
|
||||
response = self.post(url, {}, expect=201)
|
||||
j = Job.objects.get(pk=response['job'])
|
||||
self.assertTrue(j.status == 'successful')
|
||||
# Test with a job that prompts for SSH and sudo passwords.
|
||||
job = self.make_job(self.jt_sup_run, self.user_sue, 'success')
|
||||
url = reverse('api:job_start', args=(job.pk,))
|
||||
with self.current_user(self.user_sue):
|
||||
response = self.get(url)
|
||||
self.assertEqual(set(response['passwords_needed_to_start']),
|
||||
set(['ssh_password', 'become_password']))
|
||||
data = dict()
|
||||
response = self.post(url, data, expect=400)
|
||||
data['ssh_password'] = 'sshpass'
|
||||
response = self.post(url, data, expect=400)
|
||||
data2 = dict(become_password='sudopass')
|
||||
response = self.post(url, data2, expect=400)
|
||||
data.update(data2)
|
||||
response = self.post(url, data, expect=202)
|
||||
job = Job.objects.get(pk=job.pk)
|
||||
|
||||
def test_job_cancel(self):
|
||||
#job = self.job_ops_east_run
|
||||
job = self.make_job(self.jt_ops_east_run, self.user_sue, 'new')
|
||||
|
||||
Reference in New Issue
Block a user