External-Mirrors/awx
2
0
Fork 0
mirror of https://github.com/ansible/awx.git synced 2026-01-13 11:00:03 -03:30
Code Issues Packages Projects Releases Wiki Activity

refactoring survey spec tests and fixtures

Browse Source
This commit is contained in:
Wayne Witzel III 2016-06-06 13:39:25 -04:00
parent 4cd16e3749
commit 8f3c0ff884
9 changed files with 147 additions and 126 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
awx/main/tests/factories/__init__.py
View File

@ -2,6 +2,7 @@ from .tower import (
create_organization,
create_job_template,
create_notification_template,
create_survey_spec,
)
from .exc import (
@ -12,5 +13,6 @@ __all__ = [
'create_organization',
'create_job_template',
'create_notification_template',
'create_survey_spec',
'NotUnique',
]

6
awx/main/tests/factories/fixtures.py
View File

@ -106,7 +106,7 @@ def mk_inventory(name, organization=None, persisted=True):
def mk_job_template(name, job_type='run',
organization=None, inventory=None,
credential=None, persisted=True,
project=None):
project=None, spec=None):
jt = JobTemplate(name=name, job_type=job_type, playbook='mocked')
jt.inventory = inventory
@ -119,6 +119,10 @@ def mk_job_template(name, job_type='run',
jt.project = project
jt.survey_spec = spec
if jt.survey_spec is not None:
jt.survey_enabled = True
if persisted:
jt.save()
return jt

59
awx/main/tests/factories/objects.py Normal file
View File

@ -0,0 +1,59 @@
from collections import namedtuple
from .exc import NotUnique
def generate_objects(artifacts, kwargs):
'''generate_objects takes a list of artifacts that are supported by
a create function and compares it to the kwargs passed in to the create
function. If a kwarg is found that is not in the artifacts list a RuntimeError
is raised.
'''
for k in kwargs.keys():
if k not in artifacts:
raise RuntimeError('{} is not a valid argument'.format(k))
return namedtuple("Objects", ",".join(artifacts))
def generate_role_objects(objects):
'''generate_role_objects assembles a dictionary of all possible objects by name.
It will raise an exception if any of the objects share a name due to the fact that
it is to be used with apply_roles, which expects unique object names.
roles share a common name e.g. admin_role, member_role. This ensures that the
roles short hand used for mapping Roles and Users in apply_roles will function as desired.
'''
combined_objects = {}
for o in objects:
if type(o) is dict:
for k,v in o.iteritems():
if combined_objects.get(k) is not None:
raise NotUnique(k, combined_objects)
combined_objects[k] = v
elif hasattr(o, 'name'):
if combined_objects.get(o.name) is not None:
raise NotUnique(o.name, combined_objects)
combined_objects[o.name] = o
else:
if o is not None:
raise RuntimeError('expected a list of dict or list of list, got a type {}'.format(type(o)))
return combined_objects
class _Mapped(object):
'''_Mapped is a helper class that replaces spaces and dashes
in the name of an object and assigns the object as an attribute
input: {'my org': Organization}
output: instance.my_org = Organization
'''
def __init__(self, d):
self.d = d
for k,v in d.items():
k = k.replace(' ', '_')
k = k.replace('-', '_')
setattr(self, k.replace(' ','_'), v)
def all(self):
return self.d.values()

115
awx/main/tests/factories/tower.py
View File

@ -1,5 +1,3 @@
from collections import namedtuple
from django.contrib.auth.models import User
from awx.main.models import (
@ -12,6 +10,12 @@ from awx.main.models import (
Label,
)
from .objects import (
generate_objects,
generate_role_objects,
_Mapped,
)
from .fixtures import (
mk_organization,
mk_team,
@ -24,43 +28,6 @@ from .fixtures import (
mk_notification_template,
)
from .exc import NotUnique
def generate_objects(artifacts, kwargs):
'''generate_objects takes a list of artifacts that are supported by
a create function and compares it to the kwargs passed in to the create
function. If a kwarg is found that is not in the artifacts list a RuntimeError
is raised.
'''
for k in kwargs.keys():
if k not in artifacts:
raise RuntimeError('{} is not a valid argument'.format(k))
return namedtuple("Objects", ",".join(artifacts))
def generate_role_objects(objects):
'''generate_role_objects assembles a dictionary of all possible objects by name.
It will raise an exception if any of the objects share a name due to the fact that
it is to be used with apply_roles, which expects unique object names.
roles share a common name e.g. admin_role, member_role. This ensures that the
roles short hand used for mapping Roles and Users in apply_roles will function as desired.
'''
combined_objects = {}
for o in objects:
if type(o) is dict:
for k,v in o.iteritems():
if combined_objects.get(k) is not None:
raise NotUnique(k, combined_objects)
combined_objects[k] = v
elif hasattr(o, 'name'):
if combined_objects.get(o.name) is not None:
raise NotUnique(o.name, combined_objects)
combined_objects[o.name] = o
else:
if o is not None:
raise RuntimeError('expected a list of dict or list of list, got a type {}'.format(type(o)))
return combined_objects
def apply_roles(roles, objects, persisted):
'''apply_roles evaluates a list of Role relationships represented as strings.
@ -155,24 +122,51 @@ def generate_teams(organization, persisted, **kwargs):
teams[t] = mk_team(t, organization=organization, persisted=persisted)
return teams
class _Mapped(object):
'''_Mapped is a helper class that replaces spaces and dashes
in the name of an object and assigns the object as an attribute
input: {'my org': Organization}
output: instance.my_org = Organization
def create_survey_spec(variables=None, default_type='integer', required=True):
'''
def __init__(self, d):
self.d = d
for k,v in d.items():
k = k.replace(' ', '_')
k = k.replace('-', '_')
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]
setattr(self, k.replace(' ','_'), v)
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
def all(self):
return self.d.values()
# create methods are intended to be called directly as needed
# or encapsulated by specific factory fixtures in a conftest
@ -184,12 +178,14 @@ def create_job_template(name, roles=None, persisted=True, **kwargs):
"inventory",
"project",
"credential",
"job_type",], kwargs)
"job_type",
"survey",], kwargs)
org = None
proj = None
inv = None
cred = None
spec = None
job_type = kwargs.get('job_type', 'run')
if 'organization' in kwargs:
@ -212,9 +208,13 @@ def create_job_template(name, roles=None, persisted=True, **kwargs):
if type(inv) is not Inventory:
inv = mk_inventory(inv, organization=org, persisted=persisted)
if 'survey' in kwargs:
spec = create_survey_spec(kwargs['survey'])
jt = mk_job_template(name, project=proj,
inventory=inv, credential=cred,
job_type=job_type, persisted=persisted)
job_type=job_type, spec=spec,
persisted=persisted)
role_objects = generate_role_objects([org, proj, inv, cred])
apply_roles(roles, role_objects, persisted)
@ -224,7 +224,8 @@ def create_job_template(name, roles=None, persisted=True, **kwargs):
inventory=inv,
credential=cred,
job_type=job_type,
organization=org,)
organization=org,
survey=spec,)
def create_organization(name, roles=None, persisted=True, **kwargs):
Objects = generate_objects(["organization",

44
awx/main/tests/factories/utils.py
View File

@ -1,44 +0,0 @@
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
awx/main/tests/functional/api/test_job_runtime_params.py
View File

@ -5,7 +5,6 @@ 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
@ -278,10 +277,10 @@ 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, admin_user):
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 = generate_survey_spec('survey_var')
job_template.survey_spec = survey_spec_factory('survey_var')
job_template.save()
with mocker.patch('awx.main.access.BaseAccess.check_license'):

24
awx/main/tests/functional/api/test_survey_spec.py
View File

@ -5,7 +5,6 @@ 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
@ -17,12 +16,8 @@ def mock_no_surveys(self, add_host=False, feature=None, check_expiration=True):
@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
objects = job_template_factory('jt', project='prj', survey='submitted_email')
return objects.job_template
# Survey license-based denial tests
@mock.patch('awx.api.views.feature_enabled', lambda feature: False)
@ -78,8 +73,8 @@ def test_survey_spec_view_allowed(deploy_jobtemplate, get, admin_user):
@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')
def test_survey_spec_sucessful_creation(survey_spec_factory, job_template, post, admin_user):
survey_input_data = survey_spec_factory('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)
@ -100,10 +95,10 @@ def test_survey_spec_non_dict_error(deploy_jobtemplate, post, admin_user):
@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):
def test_survey_spec_dual_names_error(survey_spec_factory, 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']),
data=survey_spec_factory(['submitter_email', 'submitter_email']),
user=user('admin', True), expect=400)
assert response.data['error'] == "'variable' 'submitter_email' duplicated in survey question 1."
@ -130,10 +125,9 @@ def test_delete_survey_access_without_license(job_template_with_survey, admin_us
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')
objects = job_template_factory('jt', project='prj', survey='submitter_email')
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, {})
@ -180,12 +174,12 @@ def test_launch_survey_enabled_but_no_survey_spec(job_template_factory, post, ad
@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):
def test_launch_with_non_empty_survey_spec_no_license(survey_spec_factory, 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.survey_spec = survey_spec_factory('survey_var')
obj.save()
post(reverse('api:job_template_launch', args=[obj.pk]), {}, admin_user, expect=201)

5
awx/main/tests/functional/conftest.py
View File

@ -42,6 +42,7 @@ from awx.main.tests.factories import (
create_organization,
create_job_template,
create_notification_template,
create_survey_spec,
)
'''
@ -501,3 +502,7 @@ def organization_factory():
def notification_template_factory():
return create_notification_template
@pytest.fixture
def survey_spec_factory():
return create_survey_spec

13
awx/main/tests/functional/test_fixture_factories.py
View File

@ -1,7 +1,6 @@
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:
@ -78,23 +77,25 @@ def test_org_factory(organization_factory):
def test_job_template_factory(job_template_factory):
jt_objects = job_template_factory('testJT', organization='org1',
project='proj1', inventory='inventory1',
credential='cred1')
credential='cred1', survey='test-survey')
assert jt_objects.job_template.name == 'testJT'
assert jt_objects.project.name == 'proj1'
assert jt_objects.inventory.name == 'inventory1'
assert jt_objects.credential.name == 'cred1'
assert jt_objects.inventory.organization.name == 'org1'
assert jt_objects.job_template.survey_enabled is True
assert jt_objects.job_template.survey_spec is not None
def test_survey_spec_generator_simple():
survey_spec = generate_survey_spec('survey_variable')
def test_survey_spec_generator_simple(survey_spec_factory):
survey_spec = survey_spec_factory('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(
def test_survey_spec_generator_mixed(survey_spec_factory):
survey_spec = survey_spec_factory(
[{'variable': 'question1', 'type': 'integer', 'max': 87},
{'variable': 'question2', 'type': 'str'},
'some_variable'])