mirror of
https://github.com/ansible/awx.git
synced 2026-01-12 18:40:01 -03:30
copy WFJT node prompted fields to WJ nodes, reject or accept on unified job creation
This commit is contained in:
AlanCoding
parent
549273e90f
commit
fdca3b41ad
@ -98,9 +98,11 @@ class WorkflowNodeBase(CreatedModifiedModel):
|
||||
if ujt_obj is None:
|
||||
return {}
|
||||
prompts_dict = self.prompts_dict()
|
||||
from awx.main.models import JobTemplate
|
||||
if not isinstance(ujt_obj, JobTemplate):
|
||||
return {'ignored': {'all': 'Can not use prompts on unified_job_template that is not type of job template'}}
|
||||
if not hasattr(ujt_obj, '_ask_for_vars_dict'):
|
||||
if prompts_dict:
|
||||
return {'ignored': {'all': 'Can not use prompts on unified_job_template that is not type of job template'}}
|
||||
else:
|
||||
return {}
|
||||
ask_for_vars_dict = ujt_obj._ask_for_vars_dict()
|
||||
ignored_dict = {}
|
||||
missing_dict = {}
|
||||
@ -113,9 +115,9 @@ class WorkflowNodeBase(CreatedModifiedModel):
|
||||
missing_dict[fd] = 'Job Template does not have this field and workflow node does not provide it'
|
||||
data = {}
|
||||
if ignored_dict:
|
||||
data.update(ignored_dict)
|
||||
data['ignored'] = ignored_dict
|
||||
if missing_dict:
|
||||
data.update(missing_dict)
|
||||
data['missing'] = missing_dict
|
||||
return data
|
||||
|
||||
class WorkflowJobTemplateNode(WorkflowNodeBase):
|
||||
@ -154,14 +156,15 @@ class WorkflowJobNode(WorkflowNodeBase):
|
||||
return reverse('api:workflow_job_node_detail', args=(self.pk,))
|
||||
|
||||
def get_job_kwargs(self):
|
||||
# reject/accept prompted fields
|
||||
data = {}
|
||||
# rejecting/accepting prompting variables done with the node copy
|
||||
if self.inventory:
|
||||
data['inventory'] = self.inventory
|
||||
if self.credential:
|
||||
data['credential'] = self.credential
|
||||
if self.char_prompts:
|
||||
data.update(self.char_prompts)
|
||||
ujt_obj = self.unified_job_template
|
||||
if ujt_obj and hasattr(ujt_obj, '_ask_for_vars_dict'):
|
||||
ask_for_vars_dict = ujt_obj._ask_for_vars_dict()
|
||||
prompts_dict = self.prompts_dict()
|
||||
for fd in prompts_dict:
|
||||
if ask_for_vars_dict.get(fd, False):
|
||||
data[fd] = prompts_dict[fd]
|
||||
# process extra_vars
|
||||
extra_vars = {}
|
||||
if self.workflow_job and self.workflow_job.extra_vars:
|
||||
@ -271,27 +274,11 @@ class WorkflowJobInheritNodesMixin(object):
|
||||
Create a WorkflowJobNode for each WorkflowJobTemplateNode
|
||||
'''
|
||||
def _create_workflow_job_nodes(self, old_nodes):
|
||||
new_node_list = []
|
||||
for old_node in old_nodes:
|
||||
kwargs = dict(
|
||||
workflow_job=self,
|
||||
unified_job_template=old_node.unified_job_template,
|
||||
)
|
||||
ujt_obj = old_node.unified_job_template
|
||||
if ujt_obj and hasattr(ujt_obj, '_ask_for_vars_dict'):
|
||||
ask_for_vars_dict = ujt_obj._ask_for_vars_dict()
|
||||
if ask_for_vars_dict['inventory'] and old_node.inventory:
|
||||
kwargs['inventory'] = old_node.inventory
|
||||
if ask_for_vars_dict['credential'] and old_node.credential:
|
||||
kwargs['credential'] = old_node.credential
|
||||
new_char_prompts = {}
|
||||
for fd in CHAR_PROMPTS_LIST:
|
||||
if ask_for_vars_dict[fd] and old_node.char_prompts.get(fd, None):
|
||||
new_char_prompts[fd] = old_node.char_prompts[fd]
|
||||
if new_char_prompts:
|
||||
kwargs['char_prompts'] = new_char_prompts
|
||||
new_node_list.append(WorkflowJobNode.objects.create(**kwargs))
|
||||
return new_node_list
|
||||
return [WorkflowJobNode.objects.create(
|
||||
workflow_job=self, unified_job_template=old_node.unified_job_template,
|
||||
inventory=old_node.inventory, credential=old_node.credential,
|
||||
char_prompts=old_node.char_prompts
|
||||
) for old_node in old_nodes]
|
||||
|
||||
def _map_workflow_job_nodes(self, old_nodes, new_nodes):
|
||||
node_ids_map = {}
|
||||
|
||||
@ -1,8 +1,7 @@
|
||||
import pytest
|
||||
|
||||
from awx.main.models.jobs import JobTemplate
|
||||
from awx.main.models.inventory import Inventory
|
||||
from awx.main.models.credential import Credential
|
||||
from awx.main.models import Inventory, Credential, Project
|
||||
from awx.main.models.workflow import (
|
||||
WorkflowJobTemplateNode, WorkflowJobInheritNodesMixin,
|
||||
WorkflowJob, WorkflowJobNode
|
||||
@ -25,8 +24,9 @@ class TestWorkflowJobInheritNodesMixin():
|
||||
mixin._create_workflow_job_nodes(job_template_nodes)
|
||||
|
||||
for job_template_node in job_template_nodes:
|
||||
workflow_job_node_create.assert_any_call(workflow_job=mixin,
|
||||
unified_job_template=job_template_node.unified_job_template)
|
||||
workflow_job_node_create.assert_any_call(
|
||||
workflow_job=mixin, unified_job_template=job_template_node.unified_job_template,
|
||||
credential=None, inventory=None, char_prompts={})
|
||||
|
||||
class TestMapWorkflowJobNodes():
|
||||
@pytest.fixture
|
||||
@ -84,37 +84,95 @@ class TestWorkflowJobInheritNodesMixin():
|
||||
job_nodes[i].success_nodes.add.assert_any_call(job_nodes[i + 1])
|
||||
|
||||
|
||||
class TestWorkflowJobHelperMethods:
|
||||
@pytest.fixture
|
||||
def workflow_job_unit():
|
||||
return WorkflowJob(name='workflow', status='new')
|
||||
|
||||
@pytest.fixture
|
||||
def workflow_job_unit(self):
|
||||
return WorkflowJob(name='workflow', status='new')
|
||||
@pytest.fixture
|
||||
def node_no_prompts(workflow_job_unit, job_template_factory):
|
||||
# note: factory sets ask_xxxx_on_launch to true for inventory & credential
|
||||
jt = job_template_factory(name='example-jt', persisted=False).job_template
|
||||
jt.ask_job_type_on_launch = True
|
||||
jt.ask_skip_tags_on_launch = True
|
||||
jt.ask_limit_on_launch = True
|
||||
jt.ask_tags_on_launch = True
|
||||
return WorkflowJobNode(workflow_job=workflow_job_unit, unified_job_template=jt)
|
||||
|
||||
@pytest.fixture
|
||||
def workflow_job_node_unit(self, workflow_job_unit, job_template_factory):
|
||||
# note: factory sets ask_inventory_on_launch to true when not provided
|
||||
jt = job_template_factory(name='example-jt', persisted=False).job_template
|
||||
return WorkflowJobNode(workflow_job=workflow_job_unit, unified_job_template=jt)
|
||||
@pytest.fixture
|
||||
def project_unit():
|
||||
return Project(name='example-proj')
|
||||
|
||||
def test_null_kwargs(self, workflow_job_node_unit):
|
||||
assert workflow_job_node_unit.get_job_kwargs() == {}
|
||||
example_prompts = dict(job_type='scan', job_tags='quack', limit='duck', skip_tags='oink')
|
||||
|
||||
def test_inherit_workflow_job_extra_vars(self, workflow_job_node_unit):
|
||||
workflow_job = workflow_job_node_unit.workflow_job
|
||||
@pytest.fixture
|
||||
def node_with_prompts(node_no_prompts):
|
||||
node_no_prompts.char_prompts = example_prompts
|
||||
inv = Inventory(name='example-inv')
|
||||
cred = Credential(name='example-inv', kind='ssh', username='asdf', password='asdf')
|
||||
node_no_prompts.inventory = inv
|
||||
node_no_prompts.credential = cred
|
||||
return node_no_prompts
|
||||
|
||||
class TestWorkflowJobNodeJobKWARGS:
|
||||
"""
|
||||
Tests for building the keyword arguments that go into creating and
|
||||
launching a new job that corresponds to a workflow node.
|
||||
"""
|
||||
|
||||
def test_null_kwargs(self, node_no_prompts):
|
||||
assert node_no_prompts.get_job_kwargs() == {}
|
||||
|
||||
def test_inherit_workflow_job_extra_vars(self, node_no_prompts):
|
||||
workflow_job = node_no_prompts.workflow_job
|
||||
workflow_job.extra_vars = '{"a": 84}'
|
||||
assert workflow_job_node_unit.get_job_kwargs() == {'extra_vars': {'a': 84}}
|
||||
assert node_no_prompts.get_job_kwargs() == {'extra_vars': {'a': 84}}
|
||||
|
||||
def test_char_prompts_and_res_node_prompts(self, workflow_job_node_unit):
|
||||
barnyard_kwargs = dict(
|
||||
job_type='scan',
|
||||
job_tags='quack',
|
||||
limit='duck',
|
||||
skip_tags='oink'
|
||||
)
|
||||
workflow_job_node_unit.char_prompts = barnyard_kwargs
|
||||
inv = Inventory(name='example-inv')
|
||||
cred = Credential(name='example-inv', kind='ssh', username='asdf', password='asdf')
|
||||
workflow_job_node_unit.inventory = inv
|
||||
workflow_job_node_unit.credential = cred
|
||||
assert workflow_job_node_unit.get_job_kwargs() == dict(
|
||||
inventory=inv, credential=cred, **barnyard_kwargs)
|
||||
def test_char_prompts_and_res_node_prompts(self, node_with_prompts):
|
||||
assert node_with_prompts.get_job_kwargs() == dict(
|
||||
inventory=node_with_prompts.inventory,
|
||||
credential=node_with_prompts.credential,
|
||||
**example_prompts)
|
||||
|
||||
def test_reject_some_node_prompts(self, node_with_prompts):
|
||||
node_with_prompts.unified_job_template.ask_inventory_on_launch = False
|
||||
node_with_prompts.unified_job_template.ask_job_type_on_launch = False
|
||||
expect_kwargs = dict(inventory=node_with_prompts.inventory,
|
||||
credential=node_with_prompts.credential,
|
||||
**example_prompts)
|
||||
expect_kwargs.pop('inventory')
|
||||
expect_kwargs.pop('job_type')
|
||||
assert node_with_prompts.get_job_kwargs() == expect_kwargs
|
||||
|
||||
def test_no_accepted_project_node_prompts(self, node_with_prompts, project_unit):
|
||||
node_with_prompts.unified_job_template = project_unit
|
||||
assert node_with_prompts.get_job_kwargs() == {}
|
||||
|
||||
|
||||
class TestWorkflowWarnings:
|
||||
"""
|
||||
Tests of warnings that show user errors in the construction of a workflow
|
||||
"""
|
||||
|
||||
def test_warn_project_node_no_prompts(self, node_no_prompts, project_unit):
|
||||
node_no_prompts.unified_job_template = project_unit
|
||||
assert node_no_prompts.get_prompts_warnings() == {}
|
||||
|
||||
def test_warn_project_node_reject_all_prompts(self, node_with_prompts, project_unit):
|
||||
node_with_prompts.unified_job_template = project_unit
|
||||
assert 'ignored' in node_with_prompts.get_prompts_warnings()
|
||||
assert 'all' in node_with_prompts.get_prompts_warnings()['ignored']
|
||||
|
||||
def test_warn_reject_some_prompts(self, node_with_prompts):
|
||||
node_with_prompts.unified_job_template.ask_credential_on_launch = False
|
||||
node_with_prompts.unified_job_template.ask_job_type_on_launch = False
|
||||
assert 'ignored' in node_with_prompts.get_prompts_warnings()
|
||||
assert 'job_type' in node_with_prompts.get_prompts_warnings()['ignored']
|
||||
assert 'credential' in node_with_prompts.get_prompts_warnings()['ignored']
|
||||
assert len(node_with_prompts.get_prompts_warnings()['ignored']) == 2
|
||||
|
||||
def test_warn_missing_fields(self, node_no_prompts):
|
||||
node_no_prompts.inventory = None
|
||||
assert 'missing' in node_no_prompts.get_prompts_warnings()
|
||||
assert 'inventory' in node_no_prompts.get_prompts_warnings()['missing']
|
||||
assert 'credential' in node_no_prompts.get_prompts_warnings()['missing']
|
||||
assert len(node_no_prompts.get_prompts_warnings()['missing']) == 2
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user