mirror of
https://github.com/ZwareBear/awx.git
synced 2026-03-26 18:53:35 -05:00
623 lines
25 KiB
Python
623 lines
25 KiB
Python
import mock
|
|
import pytest
|
|
import yaml
|
|
import json
|
|
|
|
from awx.api.serializers import JobLaunchSerializer
|
|
from awx.main.models.credential import Credential
|
|
from awx.main.models.inventory import Inventory, Host
|
|
from awx.main.models.jobs import Job, JobTemplate
|
|
|
|
from awx.api.versioning import reverse
|
|
|
|
|
|
@pytest.fixture
|
|
def runtime_data(organization, credentialtype_ssh):
|
|
cred_obj = Credential.objects.create(
|
|
name='runtime-cred',
|
|
credential_type=credentialtype_ssh,
|
|
inputs={
|
|
'username': 'test_user2',
|
|
'password': 'pas4word2'
|
|
}
|
|
)
|
|
inv_obj = organization.inventories.create(name="runtime-inv")
|
|
return dict(
|
|
extra_vars='{"job_launch_var": 4}',
|
|
limit='test-servers',
|
|
job_type='check',
|
|
job_tags='provision',
|
|
skip_tags='restart',
|
|
inventory=inv_obj.pk,
|
|
credentials=[cred_obj.pk],
|
|
diff_mode=True,
|
|
verbosity=2
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def job_with_links(machine_credential, inventory):
|
|
return Job.objects.create(name='existing-job', credential=machine_credential, inventory=inventory)
|
|
|
|
|
|
@pytest.fixture
|
|
def job_template_prompts(project, inventory, machine_credential):
|
|
def rf(on_off):
|
|
jt = JobTemplate.objects.create(
|
|
job_type='run',
|
|
project=project,
|
|
inventory=inventory,
|
|
name='deploy-job-template',
|
|
# JT values must differ from prompted vals in order to register
|
|
limit='webservers',
|
|
job_tags = 'foobar',
|
|
skip_tags = 'barfoo',
|
|
ask_variables_on_launch=on_off,
|
|
ask_tags_on_launch=on_off,
|
|
ask_skip_tags_on_launch=on_off,
|
|
ask_job_type_on_launch=on_off,
|
|
ask_inventory_on_launch=on_off,
|
|
ask_limit_on_launch=on_off,
|
|
ask_credential_on_launch=on_off,
|
|
ask_diff_mode_on_launch=on_off,
|
|
ask_verbosity_on_launch=on_off,
|
|
)
|
|
jt.credentials.add(machine_credential)
|
|
return jt
|
|
return rf
|
|
|
|
|
|
@pytest.fixture
|
|
def job_template_prompts_null(project):
|
|
return JobTemplate.objects.create(
|
|
job_type='run',
|
|
project=project,
|
|
inventory=None,
|
|
name='deploy-job-template',
|
|
ask_variables_on_launch=True,
|
|
ask_tags_on_launch=True,
|
|
ask_skip_tags_on_launch=True,
|
|
ask_job_type_on_launch=True,
|
|
ask_inventory_on_launch=True,
|
|
ask_limit_on_launch=True,
|
|
ask_credential_on_launch=True,
|
|
ask_diff_mode_on_launch=True,
|
|
ask_verbosity_on_launch=True,
|
|
)
|
|
|
|
|
|
def data_to_internal(data):
|
|
'''
|
|
returns internal representation, model objects, dictionaries, etc
|
|
as opposed to integer primary keys and JSON strings
|
|
'''
|
|
internal = data.copy()
|
|
if 'extra_vars' in data:
|
|
internal['extra_vars'] = json.loads(data['extra_vars'])
|
|
if 'credentials' in data:
|
|
internal['credentials'] = set(Credential.objects.get(pk=_id) for _id in data['credentials'])
|
|
if 'inventory' in data:
|
|
internal['inventory'] = Inventory.objects.get(pk=data['inventory'])
|
|
return internal
|
|
|
|
|
|
# End of setup, tests start here
|
|
@pytest.mark.django_db
|
|
@pytest.mark.job_runtime_vars
|
|
def test_job_ignore_unprompted_vars(runtime_data, job_template_prompts, post, admin_user, mocker):
|
|
job_template = job_template_prompts(False)
|
|
|
|
mock_job = mocker.MagicMock(spec=Job, id=968, **runtime_data)
|
|
|
|
with mocker.patch.object(JobTemplate, 'create_unified_job', return_value=mock_job):
|
|
with mocker.patch('awx.api.serializers.JobSerializer.to_representation'):
|
|
response = post(reverse('api:job_template_launch', kwargs={'pk':job_template.pk}),
|
|
runtime_data, admin_user, expect=201)
|
|
assert JobTemplate.create_unified_job.called
|
|
assert JobTemplate.create_unified_job.call_args == ()
|
|
|
|
# Check that job is serialized correctly
|
|
job_id = response.data['job']
|
|
assert job_id == 968
|
|
|
|
# If job is created with no arguments, it will inherit JT attributes
|
|
mock_job.signal_start.assert_called_once()
|
|
|
|
# Check that response tells us what things were ignored
|
|
assert 'job_launch_var' in response.data['ignored_fields']['extra_vars']
|
|
assert 'job_type' in response.data['ignored_fields']
|
|
assert 'limit' in response.data['ignored_fields']
|
|
assert 'inventory' in response.data['ignored_fields']
|
|
assert 'credentials' in response.data['ignored_fields']
|
|
assert 'job_tags' in response.data['ignored_fields']
|
|
assert 'skip_tags' in response.data['ignored_fields']
|
|
|
|
|
|
@pytest.mark.django_db
|
|
@pytest.mark.job_runtime_vars
|
|
def test_job_accept_prompted_vars(runtime_data, job_template_prompts, post, admin_user, mocker):
|
|
job_template = job_template_prompts(True)
|
|
|
|
mock_job = mocker.MagicMock(spec=Job, id=968, **runtime_data)
|
|
|
|
with mocker.patch.object(JobTemplate, 'create_unified_job', return_value=mock_job):
|
|
with mocker.patch('awx.api.serializers.JobSerializer.to_representation'):
|
|
response = post(reverse('api:job_template_launch', kwargs={'pk':job_template.pk}),
|
|
runtime_data, admin_user, expect=201)
|
|
assert JobTemplate.create_unified_job.called
|
|
called_with = data_to_internal(runtime_data)
|
|
JobTemplate.create_unified_job.assert_called_with(**called_with)
|
|
|
|
job_id = response.data['job']
|
|
assert job_id == 968
|
|
|
|
mock_job.signal_start.assert_called_once()
|
|
|
|
|
|
@pytest.mark.django_db
|
|
@pytest.mark.job_runtime_vars
|
|
def test_job_accept_empty_tags(job_template_prompts, post, admin_user, mocker):
|
|
job_template = job_template_prompts(True)
|
|
|
|
mock_job = mocker.MagicMock(spec=Job, id=968)
|
|
|
|
with mocker.patch.object(JobTemplate, 'create_unified_job', return_value=mock_job):
|
|
with mocker.patch('awx.api.serializers.JobSerializer.to_representation'):
|
|
post(reverse('api:job_template_launch', kwargs={'pk': job_template.pk}),
|
|
{'job_tags': '', 'skip_tags': ''}, admin_user, expect=201)
|
|
assert JobTemplate.create_unified_job.called
|
|
assert JobTemplate.create_unified_job.call_args == ({'job_tags':'', 'skip_tags':''},)
|
|
|
|
mock_job.signal_start.assert_called_once()
|
|
|
|
|
|
@pytest.mark.django_db
|
|
@pytest.mark.job_runtime_vars
|
|
def test_job_accept_prompted_vars_null(runtime_data, job_template_prompts_null, post, rando, mocker):
|
|
job_template = job_template_prompts_null
|
|
|
|
# Give user permission to execute the job template
|
|
job_template.execute_role.members.add(rando)
|
|
|
|
# Give user permission to use inventory and credential at runtime
|
|
credential = Credential.objects.get(pk=runtime_data['credentials'][0])
|
|
credential.use_role.members.add(rando)
|
|
inventory = Inventory.objects.get(pk=runtime_data['inventory'])
|
|
inventory.use_role.members.add(rando)
|
|
|
|
mock_job = mocker.MagicMock(spec=Job, id=968, **runtime_data)
|
|
|
|
with mocker.patch.object(JobTemplate, 'create_unified_job', return_value=mock_job):
|
|
with mocker.patch('awx.api.serializers.JobSerializer.to_representation'):
|
|
response = post(reverse('api:job_template_launch', kwargs={'pk': job_template.pk}),
|
|
runtime_data, rando, expect=201)
|
|
assert JobTemplate.create_unified_job.called
|
|
expected_call = data_to_internal(runtime_data)
|
|
assert JobTemplate.create_unified_job.call_args == (expected_call,)
|
|
|
|
job_id = response.data['job']
|
|
assert job_id == 968
|
|
mock_job.signal_start.assert_called_once()
|
|
|
|
|
|
@pytest.mark.django_db
|
|
@pytest.mark.job_runtime_vars
|
|
def test_job_reject_invalid_prompted_vars(runtime_data, job_template_prompts, post, admin_user):
|
|
job_template = job_template_prompts(True)
|
|
|
|
response = post(
|
|
reverse('api:job_template_launch', kwargs={'pk':job_template.pk}),
|
|
dict(job_type='foobicate', # foobicate is not a valid job type
|
|
inventory=87865, credentials=[48474]), admin_user, expect=400)
|
|
|
|
assert response.data['job_type'] == [u'"foobicate" is not a valid choice.']
|
|
assert response.data['inventory'] == [u'Invalid pk "87865" - object does not exist.']
|
|
assert response.data['credentials'] == [u'Invalid pk "48474" - object does not exist.']
|
|
|
|
|
|
@pytest.mark.django_db
|
|
@pytest.mark.job_runtime_vars
|
|
def test_job_reject_invalid_prompted_extra_vars(runtime_data, job_template_prompts, post, admin_user):
|
|
job_template = job_template_prompts(True)
|
|
|
|
response = post(
|
|
reverse('api:job_template_launch', kwargs={'pk':job_template.pk}),
|
|
dict(extra_vars='{"unbalanced brackets":'), admin_user, expect=400)
|
|
|
|
assert 'extra_vars' in response.data
|
|
assert 'Cannot parse as' in str(response.data['extra_vars'][0])
|
|
|
|
|
|
@pytest.mark.django_db
|
|
@pytest.mark.job_runtime_vars
|
|
def test_job_launch_fails_without_inventory(deploy_jobtemplate, post, admin_user):
|
|
deploy_jobtemplate.inventory = None
|
|
deploy_jobtemplate.save()
|
|
|
|
response = post(reverse('api:job_template_launch',
|
|
kwargs={'pk': deploy_jobtemplate.pk}), {}, admin_user, expect=400)
|
|
|
|
assert 'inventory' in response.data['resources_needed_to_start'][0]
|
|
|
|
|
|
@pytest.mark.django_db
|
|
@pytest.mark.job_runtime_vars
|
|
def test_job_launch_fails_without_inventory_access(job_template_prompts, runtime_data, post, rando):
|
|
job_template = job_template_prompts(True)
|
|
job_template.execute_role.members.add(rando)
|
|
|
|
# Assure that giving an inventory without access to the inventory blocks the launch
|
|
response = post(reverse('api:job_template_launch', kwargs={'pk':job_template.pk}),
|
|
dict(inventory=runtime_data['inventory']), rando, expect=403)
|
|
|
|
assert response.data['detail'] == u'You do not have permission to perform this action.'
|
|
|
|
|
|
@pytest.mark.django_db
|
|
@pytest.mark.job_runtime_vars
|
|
def test_job_launch_fails_without_credential_access(job_template_prompts, runtime_data, post, rando):
|
|
job_template = job_template_prompts(True)
|
|
job_template.execute_role.members.add(rando)
|
|
|
|
# Assure that giving a credential without access blocks the launch
|
|
post(reverse('api:job_template_launch', kwargs={'pk':job_template.pk}),
|
|
dict(credentials=runtime_data['credentials']), rando, expect=403)
|
|
|
|
|
|
@pytest.mark.django_db
|
|
@pytest.mark.job_runtime_vars
|
|
def test_job_block_scan_job_type_change(job_template_prompts, post, admin_user):
|
|
job_template = job_template_prompts(True)
|
|
|
|
# Assure that changing the type of a scan job blocks the launch
|
|
response = post(reverse('api:job_template_launch', kwargs={'pk':job_template.pk}),
|
|
dict(job_type='scan'), admin_user, expect=400)
|
|
|
|
assert 'job_type' in response.data
|
|
|
|
|
|
@pytest.mark.django_db
|
|
def test_job_launch_JT_with_validation(machine_credential, credential, deploy_jobtemplate):
|
|
deploy_jobtemplate.extra_vars = '{"job_template_var": 3}'
|
|
deploy_jobtemplate.ask_credential_on_launch = True
|
|
deploy_jobtemplate.ask_variables_on_launch = True
|
|
deploy_jobtemplate.save()
|
|
|
|
kv = dict(extra_vars={"job_launch_var": 4}, credentials=[machine_credential.pk])
|
|
serializer = JobLaunchSerializer(data=kv, context={'template': deploy_jobtemplate})
|
|
validated = serializer.is_valid()
|
|
assert validated, serializer.errors
|
|
|
|
kv['credentials'] = [machine_credential] # conversion to internal value
|
|
job_obj = deploy_jobtemplate.create_unified_job(**kv)
|
|
|
|
final_job_extra_vars = yaml.load(job_obj.extra_vars)
|
|
assert 'job_launch_var' in final_job_extra_vars
|
|
assert 'job_template_var' in final_job_extra_vars
|
|
assert set([cred.pk for cred in job_obj.credentials.all()]) == set([machine_credential.id, credential.id])
|
|
|
|
|
|
@pytest.mark.django_db
|
|
def test_job_launch_with_default_creds(machine_credential, vault_credential, deploy_jobtemplate):
|
|
deploy_jobtemplate.ask_credential_on_launch = True
|
|
deploy_jobtemplate.credentials.add(machine_credential)
|
|
deploy_jobtemplate.credentials.add(vault_credential)
|
|
kv = dict()
|
|
serializer = JobLaunchSerializer(data=kv, context={'template': deploy_jobtemplate})
|
|
validated = serializer.is_valid()
|
|
assert validated
|
|
|
|
prompted_fields, ignored_fields, errors = deploy_jobtemplate._accept_or_ignore_job_kwargs(**kv)
|
|
job_obj = deploy_jobtemplate.create_unified_job(**prompted_fields)
|
|
assert job_obj.credential == machine_credential.pk
|
|
assert job_obj.vault_credential == vault_credential.pk
|
|
|
|
|
|
@pytest.mark.django_db
|
|
def test_job_launch_JT_enforces_unique_credentials_kinds(machine_credential, credentialtype_aws, deploy_jobtemplate):
|
|
"""
|
|
JT launching should require that extra_credentials have distinct CredentialTypes
|
|
"""
|
|
creds = []
|
|
for i in range(2):
|
|
aws = Credential.objects.create(
|
|
name='cred-%d' % i,
|
|
credential_type=credentialtype_aws,
|
|
inputs={
|
|
'username': 'test_user',
|
|
'password': 'pas4word'
|
|
}
|
|
)
|
|
aws.save()
|
|
creds.append(aws)
|
|
|
|
kv = dict(credentials=creds, credential=machine_credential.id)
|
|
serializer = JobLaunchSerializer(data=kv, context={'template': deploy_jobtemplate})
|
|
validated = serializer.is_valid()
|
|
assert not validated
|
|
|
|
|
|
@pytest.mark.django_db
|
|
def test_job_launch_with_empty_creds(machine_credential, vault_credential, deploy_jobtemplate):
|
|
deploy_jobtemplate.ask_credential_on_launch = True
|
|
deploy_jobtemplate.credentials.add(machine_credential)
|
|
deploy_jobtemplate.credentials.add(vault_credential)
|
|
kv = dict(credentials=[])
|
|
serializer = JobLaunchSerializer(data=kv, context={'template': deploy_jobtemplate})
|
|
validated = serializer.is_valid()
|
|
assert validated
|
|
|
|
prompted_fields, ignored_fields, errors = deploy_jobtemplate._accept_or_ignore_job_kwargs(**kv)
|
|
job_obj = deploy_jobtemplate.create_unified_job(**prompted_fields)
|
|
assert job_obj.credential is deploy_jobtemplate.credential
|
|
assert job_obj.vault_credential is deploy_jobtemplate.vault_credential
|
|
|
|
|
|
@pytest.mark.django_db
|
|
def test_job_launch_fails_with_missing_vault_password(machine_credential, vault_credential,
|
|
deploy_jobtemplate, post, rando):
|
|
vault_credential.vault_password = 'ASK'
|
|
vault_credential.save()
|
|
deploy_jobtemplate.credentials.add(vault_credential)
|
|
deploy_jobtemplate.execute_role.members.add(rando)
|
|
deploy_jobtemplate.save()
|
|
|
|
response = post(
|
|
reverse('api:job_template_launch', kwargs={'pk': deploy_jobtemplate.pk}),
|
|
rando,
|
|
expect=400
|
|
)
|
|
assert response.data['passwords_needed_to_start'] == ['vault_password']
|
|
|
|
|
|
@pytest.mark.django_db
|
|
@pytest.mark.parametrize('launch_kwargs', [
|
|
{'vault_password.abc': 'vault-me-1', 'vault_password.xyz': 'vault-me-2'},
|
|
{'credential_passwords': {'vault_password.abc': 'vault-me-1', 'vault_password.xyz': 'vault-me-2'}}
|
|
])
|
|
def test_job_launch_fails_with_missing_multivault_password(machine_credential, vault_credential,
|
|
deploy_jobtemplate, launch_kwargs,
|
|
get, post, rando):
|
|
vault_cred_first = Credential(
|
|
name='Vault #1',
|
|
credential_type=vault_credential.credential_type,
|
|
inputs={
|
|
'vault_password': 'ASK',
|
|
'vault_id': 'abc'
|
|
}
|
|
)
|
|
vault_cred_first.save()
|
|
vault_cred_second = Credential(
|
|
name='Vault #2',
|
|
credential_type=vault_credential.credential_type,
|
|
inputs={
|
|
'vault_password': 'ASK',
|
|
'vault_id': 'xyz'
|
|
}
|
|
)
|
|
vault_cred_second.save()
|
|
deploy_jobtemplate.credentials.add(vault_cred_first)
|
|
deploy_jobtemplate.credentials.add(vault_cred_second)
|
|
deploy_jobtemplate.execute_role.members.add(rando)
|
|
deploy_jobtemplate.save()
|
|
|
|
url = reverse('api:job_template_launch', kwargs={'pk': deploy_jobtemplate.pk})
|
|
resp = get(url, rando, expect=200)
|
|
|
|
assert {
|
|
'credential_type': vault_cred_first.credential_type_id,
|
|
'passwords_needed': ['vault_password.abc'],
|
|
'vault_id': u'abc',
|
|
'name': u'Vault #1',
|
|
'id': vault_cred_first.id
|
|
} in resp.data['defaults']['credentials']
|
|
assert {
|
|
'credential_type': vault_cred_second.credential_type_id,
|
|
'passwords_needed': ['vault_password.xyz'],
|
|
'vault_id': u'xyz',
|
|
'name': u'Vault #2',
|
|
'id': vault_cred_second.id
|
|
} in resp.data['defaults']['credentials']
|
|
|
|
assert resp.data['passwords_needed_to_start'] == ['vault_password.abc', 'vault_password.xyz']
|
|
assert sum([
|
|
cred['passwords_needed'] for cred in resp.data['defaults']['credentials']
|
|
if cred['credential_type'] == vault_credential.credential_type_id
|
|
], []) == ['vault_password.abc', 'vault_password.xyz']
|
|
|
|
resp = post(url, rando, expect=400)
|
|
assert resp.data['passwords_needed_to_start'] == ['vault_password.abc', 'vault_password.xyz']
|
|
|
|
with mock.patch.object(Job, 'signal_start') as signal_start:
|
|
post(url, launch_kwargs, rando, expect=201)
|
|
signal_start.assert_called_with(**{
|
|
'vault_password.abc': 'vault-me-1',
|
|
'vault_password.xyz': 'vault-me-2'
|
|
})
|
|
|
|
|
|
@pytest.mark.django_db
|
|
def test_job_launch_fails_with_missing_ssh_password(machine_credential, deploy_jobtemplate, post,
|
|
rando):
|
|
machine_credential.password = 'ASK'
|
|
machine_credential.save()
|
|
deploy_jobtemplate.credentials.add(machine_credential)
|
|
deploy_jobtemplate.execute_role.members.add(rando)
|
|
deploy_jobtemplate.save()
|
|
|
|
response = post(
|
|
reverse('api:job_template_launch', kwargs={'pk': deploy_jobtemplate.pk}),
|
|
rando,
|
|
expect=400
|
|
)
|
|
assert response.data['passwords_needed_to_start'] == ['ssh_password']
|
|
|
|
|
|
@pytest.mark.django_db
|
|
def test_job_launch_fails_with_missing_vault_and_ssh_password(machine_credential, vault_credential,
|
|
deploy_jobtemplate, post, rando):
|
|
vault_credential.vault_password = 'ASK'
|
|
vault_credential.save()
|
|
machine_credential.password = 'ASK'
|
|
machine_credential.save()
|
|
deploy_jobtemplate.credentials.add(machine_credential)
|
|
deploy_jobtemplate.credentials.add(vault_credential)
|
|
deploy_jobtemplate.execute_role.members.add(rando)
|
|
deploy_jobtemplate.save()
|
|
|
|
response = post(
|
|
reverse('api:job_template_launch', kwargs={'pk': deploy_jobtemplate.pk}),
|
|
rando,
|
|
expect=400
|
|
)
|
|
assert sorted(response.data['passwords_needed_to_start']) == ['ssh_password', 'vault_password']
|
|
|
|
|
|
@pytest.mark.django_db
|
|
def test_job_launch_pass_with_prompted_vault_password(machine_credential, vault_credential,
|
|
deploy_jobtemplate, post, rando):
|
|
vault_credential.vault_password = 'ASK'
|
|
vault_credential.save()
|
|
deploy_jobtemplate.credentials.add(machine_credential)
|
|
deploy_jobtemplate.credentials.add(vault_credential)
|
|
deploy_jobtemplate.execute_role.members.add(rando)
|
|
deploy_jobtemplate.save()
|
|
|
|
with mock.patch.object(Job, 'signal_start') as signal_start:
|
|
post(
|
|
reverse('api:job_template_launch', kwargs={'pk': deploy_jobtemplate.pk}),
|
|
{'vault_password': 'vault-me'},
|
|
rando,
|
|
expect=201
|
|
)
|
|
signal_start.assert_called_with(vault_password='vault-me')
|
|
|
|
|
|
@pytest.mark.django_db
|
|
def test_job_launch_JT_with_credentials(machine_credential, credential, net_credential, deploy_jobtemplate):
|
|
deploy_jobtemplate.ask_credential_on_launch = True
|
|
deploy_jobtemplate.save()
|
|
|
|
kv = dict(credentials=[credential.pk, net_credential.pk, machine_credential.pk])
|
|
serializer = JobLaunchSerializer(data=kv, context={'template': deploy_jobtemplate})
|
|
validated = serializer.is_valid()
|
|
assert validated, serializer.errors
|
|
|
|
kv['credentials'] = [credential, net_credential, machine_credential] # convert to internal value
|
|
prompted_fields, ignored_fields, errors = deploy_jobtemplate._accept_or_ignore_job_kwargs(
|
|
_exclude_errors=['required', 'prompts'], **kv)
|
|
job_obj = deploy_jobtemplate.create_unified_job(**prompted_fields)
|
|
|
|
creds = job_obj.credentials.all()
|
|
assert len(creds) == 3
|
|
assert credential in creds
|
|
assert net_credential in creds
|
|
assert machine_credential in creds
|
|
|
|
|
|
@pytest.mark.django_db
|
|
@pytest.mark.job_runtime_vars
|
|
def test_job_launch_unprompted_vars_with_survey(mocker, survey_spec_factory, job_template_prompts, post, admin_user):
|
|
job_template = job_template_prompts(False)
|
|
job_template.survey_enabled = True
|
|
job_template.survey_spec = survey_spec_factory('survey_var')
|
|
job_template.save()
|
|
|
|
with mocker.patch('awx.main.access.BaseAccess.check_license'):
|
|
mock_job = mocker.MagicMock(spec=Job, id=968, extra_vars={"job_launch_var": 3, "survey_var": 4})
|
|
with mocker.patch.object(JobTemplate, 'create_unified_job', return_value=mock_job):
|
|
with mocker.patch('awx.api.serializers.JobSerializer.to_representation', return_value={}):
|
|
response = post(
|
|
reverse('api:job_template_launch', kwargs={'pk':job_template.pk}),
|
|
dict(extra_vars={"job_launch_var": 3, "survey_var": 4}),
|
|
admin_user, expect=201)
|
|
assert JobTemplate.create_unified_job.called
|
|
assert JobTemplate.create_unified_job.call_args == ({'extra_vars':{'survey_var': 4}},)
|
|
|
|
job_id = response.data['job']
|
|
assert job_id == 968
|
|
|
|
# Check that the survey variable is accepted and the job variable isn't
|
|
mock_job.signal_start.assert_called_once()
|
|
|
|
|
|
@pytest.mark.django_db
|
|
@pytest.mark.job_runtime_vars
|
|
def test_callback_accept_prompted_extra_var(mocker, survey_spec_factory, job_template_prompts, post, admin_user, host):
|
|
job_template = job_template_prompts(True)
|
|
job_template.host_config_key = "foo"
|
|
job_template.survey_enabled = True
|
|
job_template.survey_spec = survey_spec_factory('survey_var')
|
|
job_template.save()
|
|
|
|
with mocker.patch('awx.main.access.BaseAccess.check_license'):
|
|
mock_job = mocker.MagicMock(spec=Job, id=968, extra_vars={"job_launch_var": 3, "survey_var": 4})
|
|
with mocker.patch.object(JobTemplate, 'create_unified_job', return_value=mock_job):
|
|
with mocker.patch('awx.api.serializers.JobSerializer.to_representation', return_value={}):
|
|
with mocker.patch('awx.api.views.JobTemplateCallback.find_matching_hosts', return_value=[host]):
|
|
post(
|
|
reverse('api:job_template_callback', kwargs={'pk': job_template.pk}),
|
|
dict(extra_vars={"job_launch_var": 3, "survey_var": 4}, host_config_key="foo"),
|
|
admin_user, expect=201, format='json')
|
|
assert JobTemplate.create_unified_job.called
|
|
assert JobTemplate.create_unified_job.call_args == ({
|
|
'extra_vars': {'survey_var': 4, 'job_launch_var': 3},
|
|
'_eager_fields': {'launch_type': 'callback'},
|
|
'limit': 'single-host'},
|
|
)
|
|
|
|
mock_job.signal_start.assert_called_once()
|
|
|
|
|
|
@pytest.mark.django_db
|
|
@pytest.mark.job_runtime_vars
|
|
def test_callback_ignore_unprompted_extra_var(mocker, survey_spec_factory, job_template_prompts, post, admin_user, host):
|
|
job_template = job_template_prompts(False)
|
|
job_template.host_config_key = "foo"
|
|
job_template.save()
|
|
|
|
with mocker.patch('awx.main.access.BaseAccess.check_license'):
|
|
mock_job = mocker.MagicMock(spec=Job, id=968, extra_vars={"job_launch_var": 3, "survey_var": 4})
|
|
with mocker.patch.object(JobTemplate, 'create_unified_job', return_value=mock_job):
|
|
with mocker.patch('awx.api.serializers.JobSerializer.to_representation', return_value={}):
|
|
with mocker.patch('awx.api.views.JobTemplateCallback.find_matching_hosts', return_value=[host]):
|
|
post(
|
|
reverse('api:job_template_callback', kwargs={'pk':job_template.pk}),
|
|
dict(extra_vars={"job_launch_var": 3, "survey_var": 4}, host_config_key="foo"),
|
|
admin_user, expect=201, format='json')
|
|
assert JobTemplate.create_unified_job.called
|
|
assert JobTemplate.create_unified_job.call_args == ({
|
|
'_eager_fields': {'launch_type': 'callback'},
|
|
'limit': 'single-host'},
|
|
)
|
|
|
|
mock_job.signal_start.assert_called_once()
|
|
|
|
|
|
@pytest.mark.django_db
|
|
@pytest.mark.job_runtime_vars
|
|
def test_callback_find_matching_hosts(mocker, get, job_template_prompts, admin_user):
|
|
job_template = job_template_prompts(False)
|
|
job_template.host_config_key = "foo"
|
|
job_template.save()
|
|
host_with_alias = Host(name='localhost', inventory=job_template.inventory)
|
|
host_with_alias.save()
|
|
with mocker.patch('awx.main.access.BaseAccess.check_license'):
|
|
r = get(reverse('api:job_template_callback', kwargs={'pk': job_template.pk}),
|
|
user=admin_user, expect=200)
|
|
assert tuple(r.data['matching_hosts']) == ('localhost',)
|
|
|
|
|
|
@pytest.mark.django_db
|
|
@pytest.mark.job_runtime_vars
|
|
def test_callback_extra_var_takes_priority_over_host_name(mocker, get, job_template_prompts, admin_user):
|
|
job_template = job_template_prompts(False)
|
|
job_template.host_config_key = "foo"
|
|
job_template.save()
|
|
host_with_alias = Host(name='localhost', variables={'ansible_host': 'foobar'}, inventory=job_template.inventory)
|
|
host_with_alias.save()
|
|
with mocker.patch('awx.main.access.BaseAccess.check_license'):
|
|
r = get(reverse('api:job_template_callback', kwargs={'pk': job_template.pk}),
|
|
user=admin_user, expect=200)
|
|
assert not r.data['matching_hosts']
|