External-Mirrors/awx
2
0
Fork 0
mirror of https://github.com/ansible/awx.git synced 2026-05-17 06:17:36 -02:30
Code Issues Packages Projects Releases Wiki Activity

refactor to be more testable

Browse Source
This commit is contained in:
Chris Meyers
2016-09-12 12:19:36 -04:00
parent cea5ebadb7
commit 4dc5c33442
3 changed files with 93 additions and 23 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -14,7 +14,9 @@ from awx.main.models import (
Inventory,
Label,
WorkflowJobTemplate,
WorkflowNode,
WorkflowJob,
WorkflowJobNode,
WorkflowJobTemplateNode,
)
# mk methods should create only a single object of a single type.
@@ -155,7 +157,20 @@ def mk_job_template(name, job_type='run',
jt.save()
return jt
def mk_workflow_job(status='new', workflow_job_template=None, extra_vars={},
persisted=True):
job = WorkflowJob(status=status, extra_vars=json.dumps(extra_vars))
job.workflow_job_template = workflow_job_template
if persisted:
job.save()
return job
def mk_workflow_job_template(name, extra_vars='', spec=None, persisted=True):
if extra_vars:
extra_vars = json.dumps(extra_vars)
wfjt = WorkflowJobTemplate(name=name, extra_vars=extra_vars)
wfjt.survey_spec = spec
@@ -166,15 +181,35 @@ def mk_workflow_job_template(name, extra_vars='', spec=None, persisted=True):
wfjt.save()
return wfjt
def mk_workflow_node(workflow_job_template=None, unified_job_template=None,
success_nodes=None, failure_nodes=None, always_nodes=None,
job=None, persisted=True):
workflow_node = WorkflowNode(workflow_job_template=workflow_job_template,
unified_job_template=unified_job_template,
success_nodes=success_nodes,
failure_nodes=failure_nodes,
always_nodes=always_nodes,
job=job)
def mk_workflow_job_template_node(workflow_job_template=None,
unified_job_template=None,
success_nodes=None,
failure_nodes=None,
always_nodes=None,
persisted=True):
workflow_node = WorkflowJobTemplateNode(workflow_job_template=workflow_job_template,
unified_job_template=unified_job_template,
success_nodes=success_nodes,
failure_nodes=failure_nodes,
always_nodes=always_nodes)
if persisted:
workflow_node.save()
return workflow_node
def mk_workflow_job_node(unified_job_template=None,
success_nodes=None,
failure_nodes=None,
always_nodes=None,
workflow_job=None,
job=None,
persisted=True):
workflow_node = WorkflowJobNode(unified_job_template=unified_job_template,
success_nodes=success_nodes,
failure_nodes=failure_nodes,
always_nodes=always_nodes,
workflow_job=workflow_job,
job=job)
if persisted:
workflow_node.save()
return workflow_node

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

@@ -29,6 +29,7 @@ from .fixtures import (
mk_label,
mk_notification_template,
mk_workflow_job_template,
#mk_workflow_job_template_node,
)
@@ -344,8 +345,16 @@ def create_notification_template(name, roles=None, persisted=True, **kwargs):
users=_Mapped(users),
superusers=_Mapped(superusers),
teams=teams)
'''
def generate_workflow_job_template_nodes(workflow_job_template,
unified_job_template,
persisted=True,
**kwargs):
'''
def create_workflow_job_template(name, persisted=True, **kwargs):
# TODO: Implement survey
'''
def create_workflow_job(name, persisted=True, **kwargs):
Objects = generate_objects(["workflow_job_template",
"survey",], kwargs)
@@ -353,13 +362,27 @@ def create_workflow_job_template(name, persisted=True, **kwargs):
jobs = None
extra_vars = kwargs.get('extra_vars', '')
'''
# TODO: Implement survey
def create_workflow_job_template(name, persisted=True, **kwargs):
Objects = generate_objects(["workflow_job_template",
"survey",], kwargs)
spec = None
#jobs = None
extra_vars = kwargs.get('extra_vars', '')
if 'survey' in kwargs:
spec = create_survey_spec(kwargs['survey'])
wfjt = mk_workflow_job_template(name, spec=spec, extra_vars=extra_vars,
persisted=persisted)
#workflow_nodes = generate_workflow_job_template_nodes(wfjt, persisted, workflow_nodes=kwargs.get('workflow_nodes'))
'''
if 'jobs' in kwargs:
for i in kwargs['jobs']:
if type(i) is Job:
@@ -367,8 +390,8 @@ def create_workflow_job_template(name, persisted=True, **kwargs):
else:
# TODO: Create the job
raise RuntimeError("Currently, only already created jobs are supported")
'''
return Objects(workflow_job_template=wfjt,
jobs=jobs,
#jobs=jobs,
survey=spec,)