mirror of
https://github.com/ansible/awx.git
synced 2026-01-13 19:10:07 -03:30
Bug fixes related to survey corner cases and survey test refactor
This commit is contained in:
AlanCoding
parent
860d0eea20
commit
53327dea2b
@ -814,7 +814,7 @@ class JobTemplateAccess(BaseAccess):
|
||||
|
||||
def can_change(self, obj, data):
|
||||
data_for_change = data
|
||||
if self.user not in obj.admin_role:
|
||||
if self.user not in obj.admin_role and not self.user.is_superuser:
|
||||
return False
|
||||
if data is not None:
|
||||
data = dict(data)
|
||||
@ -873,6 +873,7 @@ class JobTemplateAccess(BaseAccess):
|
||||
|
||||
return True
|
||||
|
||||
@check_superuser
|
||||
def can_delete(self, obj):
|
||||
return self.user in obj.admin_role
|
||||
|
||||
|
||||
@ -437,9 +437,9 @@ class JobTemplate(UnifiedJobTemplate, JobOptions, ResourceMixin):
|
||||
if ask_for_vars_dict[field]:
|
||||
prompted_fields[field] = kwargs[field]
|
||||
else:
|
||||
if field == 'extra_vars' and self.survey_enabled:
|
||||
if field == 'extra_vars' and self.survey_enabled and self.survey_spec:
|
||||
# Accept vars defined in the survey and no others
|
||||
survey_vars = [question['variable'] for question in self.survey_spec['spec']]
|
||||
survey_vars = [question['variable'] for question in self.survey_spec.get('spec', [])]
|
||||
for key in kwargs[field]:
|
||||
if key in survey_vars:
|
||||
prompted_fields[field][key] = kwargs[field][key]
|
||||
|
||||
44
awx/main/tests/factories/utils.py
Normal file
44
awx/main/tests/factories/utils.py
Normal file
@ -0,0 +1,44 @@
|
||||
def generate_survey_spec(variables=None, default_type='integer', required=True):
|
||||
'''
|
||||
Returns a valid survey spec for a job template, based on the input
|
||||
argument specifying variable name(s)
|
||||
'''
|
||||
if isinstance(variables, list):
|
||||
name = "%s survey" % variables[0]
|
||||
description = "A survey that starts with %s." % variables[0]
|
||||
vars_list = variables
|
||||
else:
|
||||
name = "%s survey" % variables
|
||||
description = "A survey about %s." % variables
|
||||
vars_list = [variables]
|
||||
|
||||
spec = []
|
||||
index = 0
|
||||
for var in vars_list:
|
||||
spec_item = {}
|
||||
spec_item['index'] = index
|
||||
index += 1
|
||||
spec_item['required'] = required
|
||||
spec_item['choices'] = ''
|
||||
spec_item['type'] = default_type
|
||||
if isinstance(var, dict):
|
||||
spec_item.update(var)
|
||||
var_name = spec_item.get('variable', 'variable')
|
||||
else:
|
||||
var_name = var
|
||||
spec_item.setdefault('variable', var_name)
|
||||
spec_item.setdefault('question_name', "Enter a value for %s." % var_name)
|
||||
spec_item.setdefault('question_description', "A question about %s." % var_name)
|
||||
if spec_item['type'] == 'integer':
|
||||
spec_item.setdefault('default', 0)
|
||||
spec_item.setdefault('max', spec_item['default'] + 100)
|
||||
spec_item.setdefault('min', spec_item['default'] - 100)
|
||||
else:
|
||||
spec_item.setdefault('default', '')
|
||||
spec.append(spec_item)
|
||||
|
||||
survey_spec = {}
|
||||
survey_spec['spec'] = spec
|
||||
survey_spec['name'] = name
|
||||
survey_spec['description'] = description
|
||||
return survey_spec
|
||||
@ -5,6 +5,7 @@ from awx.api.serializers import JobLaunchSerializer
|
||||
from awx.main.models.credential import Credential
|
||||
from awx.main.models.inventory import Inventory
|
||||
from awx.main.models.jobs import Job, JobTemplate
|
||||
from awx.main.tests.factories.utils import generate_survey_spec
|
||||
|
||||
from django.core.urlresolvers import reverse
|
||||
|
||||
@ -70,7 +71,7 @@ def bad_scan_JT(job_template_prompts):
|
||||
# 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, user, mocker):
|
||||
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)
|
||||
@ -78,8 +79,7 @@ def test_job_ignore_unprompted_vars(runtime_data, job_template_prompts, post, us
|
||||
with mocker.patch('awx.main.models.unified_jobs.UnifiedJobTemplate.create_unified_job', return_value=mock_job):
|
||||
with mocker.patch('awx.api.serializers.JobSerializer.to_representation'):
|
||||
response = post(reverse('api:job_template_launch', args=[job_template.pk]),
|
||||
runtime_data, user('admin', True))
|
||||
assert response.status_code == 201
|
||||
runtime_data, admin_user, expect=201)
|
||||
|
||||
# Check that job is serialized correctly
|
||||
job_id = response.data['job']
|
||||
@ -99,7 +99,7 @@ def test_job_ignore_unprompted_vars(runtime_data, job_template_prompts, post, us
|
||||
|
||||
@pytest.mark.django_db
|
||||
@pytest.mark.job_runtime_vars
|
||||
def test_job_accept_prompted_vars(runtime_data, job_template_prompts, post, user, mocker):
|
||||
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)
|
||||
@ -107,9 +107,8 @@ def test_job_accept_prompted_vars(runtime_data, job_template_prompts, post, user
|
||||
with mocker.patch('awx.main.models.unified_jobs.UnifiedJobTemplate.create_unified_job', return_value=mock_job):
|
||||
with mocker.patch('awx.api.serializers.JobSerializer.to_representation'):
|
||||
response = post(reverse('api:job_template_launch', args=[job_template.pk]),
|
||||
runtime_data, user('admin', True))
|
||||
runtime_data, admin_user, expect=201)
|
||||
|
||||
assert response.status_code == 201
|
||||
job_id = response.data['job']
|
||||
assert job_id == 968
|
||||
|
||||
@ -134,50 +133,46 @@ def test_job_accept_prompted_vars_null(runtime_data, job_template_prompts_null,
|
||||
with mocker.patch('awx.main.models.unified_jobs.UnifiedJobTemplate.create_unified_job', return_value=mock_job):
|
||||
with mocker.patch('awx.api.serializers.JobSerializer.to_representation'):
|
||||
response = post(reverse('api:job_template_launch', args=[job_template.pk]),
|
||||
runtime_data, rando)
|
||||
runtime_data, rando, expect=201)
|
||||
|
||||
assert response.status_code == 201
|
||||
job_id = response.data['job']
|
||||
assert job_id == 968
|
||||
mock_job.signal_start.assert_called_once_with(**runtime_data)
|
||||
|
||||
@pytest.mark.django_db
|
||||
@pytest.mark.job_runtime_vars
|
||||
def test_job_reject_invalid_prompted_vars(runtime_data, job_template_prompts, post, user):
|
||||
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', args=[job_template.pk]),
|
||||
dict(job_type='foobicate', # foobicate is not a valid job type
|
||||
inventory=87865, credential=48474), user('admin', True))
|
||||
inventory=87865, credential=48474), admin_user, expect=400)
|
||||
|
||||
assert response.status_code == 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['credential'] == [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, user):
|
||||
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', args=[job_template.pk]),
|
||||
dict(extra_vars='{"unbalanced brackets":'), user('admin', True))
|
||||
dict(extra_vars='{"unbalanced brackets":'), admin_user, expect=400)
|
||||
|
||||
assert response.status_code == 400
|
||||
assert response.data['extra_vars'] == ['Must be a valid JSON or YAML dictionary.']
|
||||
|
||||
@pytest.mark.django_db
|
||||
@pytest.mark.job_runtime_vars
|
||||
def test_job_launch_fails_without_inventory(deploy_jobtemplate, post, user):
|
||||
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',
|
||||
args=[deploy_jobtemplate.pk]), {}, user('admin', True))
|
||||
args=[deploy_jobtemplate.pk]), {}, admin_user, expect=400)
|
||||
|
||||
assert response.status_code == 400
|
||||
assert response.data['inventory'] == ["Job Template 'inventory' is missing or undefined."]
|
||||
|
||||
@pytest.mark.django_db
|
||||
@ -188,9 +183,8 @@ def test_job_launch_fails_without_inventory_access(job_template_prompts, runtime
|
||||
|
||||
# Assure that giving an inventory without access to the inventory blocks the launch
|
||||
response = post(reverse('api:job_template_launch', args=[job_template.pk]),
|
||||
dict(inventory=runtime_data['inventory']), rando)
|
||||
dict(inventory=runtime_data['inventory']), rando, expect=403)
|
||||
|
||||
assert response.status_code == 403
|
||||
assert response.data['detail'] == u'You do not have permission to perform this action.'
|
||||
|
||||
@pytest.mark.django_db
|
||||
@ -201,9 +195,8 @@ def test_job_launch_fails_without_credential_access(job_template_prompts, runtim
|
||||
|
||||
# Assure that giving a credential without access blocks the launch
|
||||
response = post(reverse('api:job_template_launch', args=[job_template.pk]),
|
||||
dict(credential=runtime_data['credential']), rando)
|
||||
dict(credential=runtime_data['credential']), rando, expect=403)
|
||||
|
||||
assert response.status_code == 403
|
||||
assert response.data['detail'] == u'You do not have permission to perform this action.'
|
||||
|
||||
@pytest.mark.django_db
|
||||
@ -213,20 +206,19 @@ def test_job_block_scan_job_type_change(job_template_prompts, post, admin_user):
|
||||
|
||||
# Assure that changing the type of a scan job blocks the launch
|
||||
response = post(reverse('api:job_template_launch', args=[job_template.pk]),
|
||||
dict(job_type='scan'), admin_user)
|
||||
dict(job_type='scan'), admin_user, expect=400)
|
||||
|
||||
assert response.status_code == 400
|
||||
assert 'job_type' in response.data
|
||||
|
||||
@pytest.mark.django_db
|
||||
@pytest.mark.job_runtime_vars
|
||||
def test_job_block_scan_job_inv_change(mocker, bad_scan_JT, runtime_data, post, admin_user):
|
||||
# Assure that giving a new inventory for a scan job blocks the launch
|
||||
with mocker.patch('awx.main.access.BaseAccess.check_license', return_value=True):
|
||||
with mocker.patch('awx.main.access.BaseAccess.check_license'):
|
||||
response = post(reverse('api:job_template_launch', args=[bad_scan_JT.pk]),
|
||||
dict(inventory=runtime_data['inventory']), admin_user)
|
||||
dict(inventory=runtime_data['inventory']), admin_user,
|
||||
expect=400)
|
||||
|
||||
assert response.status_code == 400
|
||||
assert 'inventory' in response.data
|
||||
|
||||
@pytest.mark.django_db
|
||||
@ -286,41 +278,23 @@ def test_job_launch_JT_with_validation(machine_credential, deploy_jobtemplate):
|
||||
|
||||
@pytest.mark.django_db
|
||||
@pytest.mark.job_runtime_vars
|
||||
def test_job_launch_unprompted_vars_with_survey(mocker, job_template_prompts, post, user):
|
||||
with mocker.patch('awx.main.access.BaseAccess.check_license', return_value=False):
|
||||
job_template = job_template_prompts(False)
|
||||
job_template.survey_enabled = True
|
||||
job_template.survey_spec = {
|
||||
"spec": [
|
||||
{
|
||||
"index": 0,
|
||||
"question_name": "survey_var",
|
||||
"min": 0,
|
||||
"default": "",
|
||||
"max": 100,
|
||||
"question_description": "A survey question",
|
||||
"required": True,
|
||||
"variable": "survey_var",
|
||||
"choices": "",
|
||||
"type": "integer"
|
||||
}
|
||||
],
|
||||
"description": "",
|
||||
"name": ""
|
||||
}
|
||||
job_template.save()
|
||||
def test_job_launch_unprompted_vars_with_survey(mocker, job_template_prompts, post, admin_user):
|
||||
job_template = job_template_prompts(False)
|
||||
job_template.survey_enabled = True
|
||||
job_template.survey_spec = generate_survey_spec('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('awx.main.models.unified_jobs.UnifiedJobTemplate.create_unified_job', return_value=mock_job):
|
||||
with mocker.patch('awx.api.serializers.JobSerializer.to_representation'):
|
||||
with mocker.patch('awx.api.serializers.JobSerializer.to_representation', return_value={}):
|
||||
response = post(
|
||||
reverse('api:job_template_launch', args=[job_template.pk]),
|
||||
dict(extra_vars={"job_launch_var": 3, "survey_var": 4}),
|
||||
user('admin', True))
|
||||
assert response.status_code == 201
|
||||
admin_user, expect=201)
|
||||
|
||||
job_id = response.data['job']
|
||||
assert job_id == 968
|
||||
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_with(extra_vars={"survey_var": 4})
|
||||
# Check that the survey variable is accepted and the job variable isn't
|
||||
mock_job.signal_start.assert_called_once_with(extra_vars={"survey_var": 4})
|
||||
|
||||
191
awx/main/tests/functional/api/test_survey_spec.py
Normal file
191
awx/main/tests/functional/api/test_survey_spec.py
Normal file
@ -0,0 +1,191 @@
|
||||
import mock
|
||||
import pytest
|
||||
|
||||
from django.core.urlresolvers import reverse
|
||||
|
||||
from awx.main.models.jobs import JobTemplate, Job
|
||||
from awx.api.license import LicenseForbids
|
||||
from awx.main.tests.factories.utils import generate_survey_spec
|
||||
from awx.main.access import JobTemplateAccess
|
||||
|
||||
|
||||
def mock_no_surveys(self, add_host=False, feature=None, check_expiration=True):
|
||||
if feature == 'surveys':
|
||||
raise LicenseForbids("Feature %s is not enabled in the active license." % feature)
|
||||
else:
|
||||
pass
|
||||
|
||||
@pytest.fixture
|
||||
def job_template_with_survey(job_template_factory):
|
||||
objects = job_template_factory('jt', project='prj')
|
||||
obj = objects.job_template
|
||||
obj.survey_enabled = True
|
||||
obj.survey_spec = generate_survey_spec('submitter_email')
|
||||
obj.save()
|
||||
return obj
|
||||
|
||||
# Survey license-based denial tests
|
||||
@mock.patch('awx.api.views.feature_enabled', lambda feature: False)
|
||||
@pytest.mark.django_db
|
||||
@pytest.mark.survey
|
||||
def test_survey_spec_view_denied(job_template_with_survey, get, admin_user):
|
||||
# TODO: Test non-enterprise license
|
||||
response = get(reverse('api:job_template_survey_spec',
|
||||
args=(job_template_with_survey.id,)), admin_user, expect=402)
|
||||
assert response.data['detail'] == 'Your license does not allow adding surveys.'
|
||||
|
||||
@mock.patch('awx.main.access.BaseAccess.check_license', mock_no_surveys)
|
||||
@pytest.mark.django_db
|
||||
@pytest.mark.survey
|
||||
def test_deny_enabling_survey(deploy_jobtemplate, patch, admin_user):
|
||||
response = patch(url=reverse('api:job_template_detail', args=(deploy_jobtemplate.id,)),
|
||||
data=dict(survey_enabled=True), user=admin_user, expect=402)
|
||||
assert response.data['detail'] == 'Feature surveys is not enabled in the active license.'
|
||||
|
||||
@mock.patch('awx.main.access.BaseAccess.check_license', new=mock_no_surveys)
|
||||
@pytest.mark.django_db
|
||||
@pytest.mark.survey
|
||||
def test_job_start_blocked_without_survey_license(job_template_with_survey, admin_user):
|
||||
"""Check that user can't start a job with surveys without a survey license."""
|
||||
access = JobTemplateAccess(admin_user)
|
||||
with pytest.raises(LicenseForbids):
|
||||
access.can_start(job_template_with_survey)
|
||||
|
||||
@mock.patch('awx.main.access.BaseAccess.check_license', mock_no_surveys)
|
||||
@pytest.mark.django_db
|
||||
@pytest.mark.survey
|
||||
def test_deny_creating_with_survey(project, post, admin_user):
|
||||
response = post(
|
||||
url=reverse('api:job_template_list'),
|
||||
data=dict(
|
||||
name = 'JT with survey',
|
||||
job_type = 'run',
|
||||
project = project.pk,
|
||||
playbook = 'helloworld.yml',
|
||||
ask_credential_on_launch = True,
|
||||
ask_inventory_on_launch = True,
|
||||
survey_enabled = True),
|
||||
user=admin_user, expect=402)
|
||||
assert response.data['detail'] == 'Feature surveys is not enabled in the active license.'
|
||||
|
||||
@mock.patch('awx.api.views.feature_enabled', lambda feature: True)
|
||||
@pytest.mark.django_db
|
||||
@pytest.mark.survey
|
||||
def test_survey_spec_view_allowed(deploy_jobtemplate, get, admin_user):
|
||||
get(reverse('api:job_template_survey_spec', args=(deploy_jobtemplate.id,)),
|
||||
admin_user, expect=200)
|
||||
|
||||
@mock.patch('awx.api.views.feature_enabled', lambda feature: True)
|
||||
@pytest.mark.django_db
|
||||
@pytest.mark.survey
|
||||
def test_survey_spec_sucessful_creation(job_template, post, admin_user):
|
||||
survey_input_data = generate_survey_spec('new_question')
|
||||
post(url=reverse('api:job_template_survey_spec', args=(job_template.id,)),
|
||||
data=survey_input_data, user=admin_user, expect=200)
|
||||
updated_jt = JobTemplate.objects.get(pk=job_template.pk)
|
||||
assert updated_jt.survey_spec == survey_input_data
|
||||
|
||||
@mock.patch('awx.api.views.feature_enabled', lambda feature: True)
|
||||
@pytest.mark.django_db
|
||||
@pytest.mark.survey
|
||||
def test_survey_spec_non_dict_error(deploy_jobtemplate, post, admin_user):
|
||||
"""When a question doesn't follow the standard format, verify error thrown."""
|
||||
response = post(
|
||||
url=reverse('api:job_template_survey_spec', args=(deploy_jobtemplate.id,)),
|
||||
data={"description": "Email of the submitter",
|
||||
"spec": ["What is your email?"], "name": "Email survey"},
|
||||
user=admin_user, expect=400)
|
||||
assert response.data['error'] == "Survey question 0 is not a json object."
|
||||
|
||||
@mock.patch('awx.api.views.feature_enabled', lambda feature: True)
|
||||
@pytest.mark.django_db
|
||||
@pytest.mark.survey
|
||||
def test_survey_spec_dual_names_error(deploy_jobtemplate, post, user):
|
||||
response = post(
|
||||
url=reverse('api:job_template_survey_spec', args=(deploy_jobtemplate.id,)),
|
||||
data=generate_survey_spec(['submitter_email', 'submitter_email']),
|
||||
user=user('admin', True), expect=400)
|
||||
assert response.data['error'] == "'variable' 'submitter_email' duplicated in survey question 1."
|
||||
|
||||
@mock.patch('awx.main.access.BaseAccess.check_license', new=mock_no_surveys)
|
||||
@pytest.mark.django_db
|
||||
@pytest.mark.survey
|
||||
def test_disable_survey_access_without_license(job_template_with_survey, admin_user):
|
||||
"""Assure that user can disable a JT survey after downgrading license."""
|
||||
access = JobTemplateAccess(admin_user)
|
||||
assert access.can_change(job_template_with_survey, dict(survey_enabled=False))
|
||||
|
||||
@mock.patch('awx.main.access.BaseAccess.check_license', new=mock_no_surveys)
|
||||
@pytest.mark.django_db
|
||||
@pytest.mark.survey
|
||||
def test_delete_survey_access_without_license(job_template_with_survey, admin_user):
|
||||
"""Assure that access.py allows deleting surveys after downgrading license."""
|
||||
access = JobTemplateAccess(admin_user)
|
||||
assert access.can_change(job_template_with_survey, dict(survey_spec=None))
|
||||
assert access.can_change(job_template_with_survey, dict(survey_spec={}))
|
||||
|
||||
@mock.patch('awx.main.access.BaseAccess.check_license', new=mock_no_surveys)
|
||||
@pytest.mark.django_db
|
||||
@pytest.mark.survey
|
||||
def test_job_start_allowed_with_survey_spec(job_template_factory, admin_user):
|
||||
"""After user downgrades survey license and disables survey on the JT,
|
||||
check that jobs still launch even if the survey_spec data persists."""
|
||||
objects = job_template_factory('jt', project='prj')
|
||||
obj = objects.job_template
|
||||
obj.survey_enabled = False
|
||||
obj.survey_spec = generate_survey_spec('submitter_email')
|
||||
obj.save()
|
||||
access = JobTemplateAccess(admin_user)
|
||||
assert access.can_start(job_template_with_survey, {})
|
||||
|
||||
@mock.patch('awx.main.access.BaseAccess.check_license', new=mock_no_surveys)
|
||||
@pytest.mark.django_db
|
||||
@pytest.mark.survey
|
||||
def test_job_template_delete_access_with_survey(job_template_with_survey, admin_user):
|
||||
"""The survey_spec view relies on JT `can_delete` to determine permission
|
||||
to delete the survey. This checks that system admins can delete the survey on a JT."""
|
||||
access = JobTemplateAccess(admin_user)
|
||||
assert access.can_delete(job_template_with_survey)
|
||||
|
||||
@mock.patch('awx.api.views.feature_enabled', lambda feature: False)
|
||||
@mock.patch('awx.main.access.BaseAccess.check_license', new=mock_no_surveys)
|
||||
@pytest.mark.django_db
|
||||
@pytest.mark.survey
|
||||
def test_delete_survey_spec_without_license(job_template_with_survey, delete, admin_user):
|
||||
"""Functional delete test through the survey_spec view."""
|
||||
delete(reverse('api:job_template_survey_spec', args=[job_template_with_survey.pk]),
|
||||
admin_user, expect=200)
|
||||
new_jt = JobTemplate.objects.get(pk=job_template_with_survey.pk)
|
||||
assert new_jt.survey_spec == {}
|
||||
|
||||
@mock.patch('awx.main.access.BaseAccess.check_license', lambda self, **kwargs: True)
|
||||
@mock.patch('awx.main.models.unified_jobs.UnifiedJobTemplate.create_unified_job',
|
||||
lambda self, extra_vars: mock.MagicMock(spec=Job, id=968))
|
||||
@mock.patch('awx.api.serializers.JobSerializer.to_representation', lambda self, obj: {})
|
||||
@pytest.mark.django_db
|
||||
@pytest.mark.survey
|
||||
def test_launch_survey_enabled_but_no_survey_spec(job_template_factory, post, admin_user):
|
||||
"""False-ish values for survey_spec are interpreted as a survey with 0 questions."""
|
||||
objects = job_template_factory('jt', organization='org1', project='prj',
|
||||
inventory='inv', credential='cred')
|
||||
obj = objects.job_template
|
||||
obj.survey_enabled = True
|
||||
obj.save()
|
||||
post(reverse('api:job_template_launch', args=[obj.pk]),
|
||||
dict(extra_vars=dict(survey_var=7)), admin_user, expect=201)
|
||||
|
||||
@mock.patch('awx.main.access.BaseAccess.check_license', new=mock_no_surveys)
|
||||
@mock.patch('awx.main.models.unified_jobs.UnifiedJobTemplate.create_unified_job',
|
||||
lambda self: mock.MagicMock(spec=Job, id=968))
|
||||
@mock.patch('awx.api.serializers.JobSerializer.to_representation', lambda self, obj: {})
|
||||
@pytest.mark.django_db
|
||||
@pytest.mark.survey
|
||||
def test_launch_with_non_empty_survey_spec_no_license(job_template_factory, post, admin_user):
|
||||
"""Assure jobs can still be launched from JTs with a survey_spec
|
||||
when the survey is diabled."""
|
||||
objects = job_template_factory('jt', organization='org1', project='prj',
|
||||
inventory='inv', credential='cred')
|
||||
obj = objects.job_template
|
||||
obj.survey_spec = generate_survey_spec('survey_var')
|
||||
obj.save()
|
||||
post(reverse('api:job_template_launch', args=[obj.pk]), {}, admin_user, expect=201)
|
||||
@ -1,135 +0,0 @@
|
||||
import mock
|
||||
import pytest
|
||||
|
||||
from django.core.urlresolvers import reverse
|
||||
from awx.main.models.jobs import JobTemplate
|
||||
from awx.api.license import LicenseForbids
|
||||
|
||||
def mock_feature_enabled(feature, bypass_database=None):
|
||||
return True
|
||||
|
||||
def mock_feature_disabled(feature, bypass_database=None):
|
||||
return False
|
||||
|
||||
def mock_check_license(self, add_host=False, feature=None, check_expiration=True):
|
||||
raise LicenseForbids("Feature %s is not enabled in the active license." % feature)
|
||||
|
||||
@pytest.fixture
|
||||
def survey_jobtemplate(project, inventory, credential):
|
||||
return JobTemplate.objects.create(
|
||||
job_type='run',
|
||||
project=project,
|
||||
inventory=inventory,
|
||||
credential=credential,
|
||||
name='deploy-job-template'
|
||||
)
|
||||
|
||||
@mock.patch('awx.api.views.feature_enabled', new=mock_feature_disabled)
|
||||
@pytest.mark.django_db
|
||||
@pytest.mark.survey
|
||||
def test_survey_spec_view_denied(deploy_jobtemplate, get, user):
|
||||
# TODO: Test non-enterprise license
|
||||
spec_url = reverse('api:job_template_survey_spec', args=(deploy_jobtemplate.id,))
|
||||
response = get(spec_url, user('admin', True))
|
||||
|
||||
assert response.status_code == 402
|
||||
assert response.data['detail'] == 'Your license does not allow adding surveys.'
|
||||
|
||||
@mock.patch('awx.main.access.BaseAccess.check_license', mock_check_license)
|
||||
@pytest.mark.django_db
|
||||
@pytest.mark.survey
|
||||
def test_deny_enabling_survey(deploy_jobtemplate, patch, user):
|
||||
JT_url = reverse('api:job_template_detail', args=(deploy_jobtemplate.id,))
|
||||
response = patch(url=JT_url, data=dict(survey_enabled=True), user=user('admin', True))
|
||||
assert response.status_code == 402
|
||||
assert response.data['detail'] == 'Feature surveys is not enabled in the active license.'
|
||||
|
||||
@mock.patch('awx.main.access.BaseAccess.check_license', mock_check_license)
|
||||
@pytest.mark.django_db
|
||||
@pytest.mark.survey
|
||||
def test_deny_creating_with_survey(machine_credential, project, inventory, post, user):
|
||||
JT_url = reverse('api:job_template_list')
|
||||
JT_data = dict(
|
||||
name = 'JT with survey',
|
||||
job_type = 'run',
|
||||
inventory = inventory.pk,
|
||||
project = project.pk,
|
||||
playbook = 'hiworld.yml',
|
||||
credential = machine_credential.pk,
|
||||
survey_enabled = True,
|
||||
)
|
||||
response = post(url=JT_url, data=JT_data, user=user('admin', True))
|
||||
|
||||
assert response.status_code == 402
|
||||
assert response.data['detail'] == 'Feature surveys is not enabled in the active license.'
|
||||
|
||||
@mock.patch('awx.api.views.feature_enabled', new=mock_feature_enabled)
|
||||
@pytest.mark.django_db
|
||||
@pytest.mark.survey
|
||||
def test_survey_spec_view_allowed(deploy_jobtemplate, get, user):
|
||||
spec_url = reverse('api:job_template_survey_spec', args=(deploy_jobtemplate.id,))
|
||||
response = get(spec_url, user('admin', True))
|
||||
|
||||
assert response.status_code == 200
|
||||
|
||||
@mock.patch('awx.api.views.feature_enabled', new=mock_feature_enabled)
|
||||
@pytest.mark.django_db
|
||||
@pytest.mark.survey
|
||||
def test_survey_spec_sucessful_creation(deploy_jobtemplate, post, user):
|
||||
spec_url = reverse('api:job_template_survey_spec', args=(deploy_jobtemplate.id,))
|
||||
response = post(
|
||||
url=spec_url,
|
||||
data={
|
||||
"description": "Email of the submitter",
|
||||
"spec": [{
|
||||
"variable": "submitter_email",
|
||||
"question_name": "Enter your email",
|
||||
"type": "text",
|
||||
"required": False
|
||||
}],
|
||||
"name": "Email survey"
|
||||
},
|
||||
user=user('admin', True))
|
||||
|
||||
assert response.status_code == 200
|
||||
|
||||
@mock.patch('awx.api.views.feature_enabled', new=mock_feature_enabled)
|
||||
@pytest.mark.django_db
|
||||
@pytest.mark.survey
|
||||
def test_survey_spec_non_dict_error(deploy_jobtemplate, post, user):
|
||||
spec_url = reverse('api:job_template_survey_spec', args=(deploy_jobtemplate.id,))
|
||||
response = post(
|
||||
url=spec_url,
|
||||
data={"description": "Email of the submitter",
|
||||
"spec": ["What is your email?"], "name": "Email survey"},
|
||||
user=user('admin', True))
|
||||
|
||||
assert response.status_code == 400
|
||||
assert response.data['error'] == "Survey question 0 is not a json object."
|
||||
|
||||
@mock.patch('awx.api.views.feature_enabled', new=mock_feature_enabled)
|
||||
@pytest.mark.django_db
|
||||
@pytest.mark.survey
|
||||
def test_survey_spec_dual_names_error(deploy_jobtemplate, post, user):
|
||||
spec_url = reverse('api:job_template_survey_spec', args=(deploy_jobtemplate.id,))
|
||||
response = post(
|
||||
url=spec_url,
|
||||
data={
|
||||
"description": "Email of the submitter",
|
||||
"spec": [{
|
||||
"variable": "submitter_email",
|
||||
"question_name": "Enter your email",
|
||||
"type": "text",
|
||||
"required": False
|
||||
}, {
|
||||
"variable": "submitter_email",
|
||||
"question_name": "Same variable as last question",
|
||||
"type": "integer",
|
||||
"required": False
|
||||
}],
|
||||
"name": "Email survey"
|
||||
},
|
||||
user=user('admin', True))
|
||||
|
||||
assert response.status_code == 400
|
||||
assert response.data['error'] == "'variable' 'submitter_email' duplicated in survey question 1."
|
||||
@ -1,6 +1,7 @@
|
||||
import pytest
|
||||
|
||||
from awx.main.tests.factories import NotUnique
|
||||
from awx.main.tests.factories.utils import generate_survey_spec
|
||||
|
||||
def test_roles_exc_not_persisted(organization_factory):
|
||||
with pytest.raises(RuntimeError) as exc:
|
||||
@ -83,3 +84,20 @@ def test_job_template_factory(job_template_factory):
|
||||
assert jt_objects.inventory.name == 'inventory1'
|
||||
assert jt_objects.credential.name == 'cred1'
|
||||
assert jt_objects.inventory.organization.name == 'org1'
|
||||
|
||||
def test_survey_spec_generator_simple():
|
||||
survey_spec = generate_survey_spec('survey_variable')
|
||||
assert 'name' in survey_spec
|
||||
assert 'spec' in survey_spec
|
||||
assert type(survey_spec['spec']) is list
|
||||
assert type(survey_spec['spec'][0]) is dict
|
||||
assert survey_spec['spec'][0]['type'] == 'integer'
|
||||
|
||||
def test_survey_spec_generator_mixed():
|
||||
survey_spec = generate_survey_spec(
|
||||
[{'variable': 'question1', 'type': 'integer', 'max': 87},
|
||||
{'variable': 'question2', 'type': 'str'},
|
||||
'some_variable'])
|
||||
assert len(survey_spec['spec']) == 3
|
||||
assert [spec_item['type'] for spec_item in survey_spec['spec']] == ['integer', 'str', 'integer']
|
||||
assert survey_spec['spec'][0]['max'] == 87
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user