Feature: saved launchtime configurations

Consolidate prompts accept/reject logic in unified models
Break out accept/reject logic for variables
Surface new promptable fields on WFJT nodes, schedules

Make schedules and workflows accurately reject variables
  that are not allowed by the prompting
  rules or the survey rules on the template

Validate against unallowed extra_data in system job schedules
Prevent schedule or WFJT node POST/PATCH with unprompted data
Move system job days validation to new mechanism
Add new psuedo-field for WFJT node credential
Add validation for node related credentials
Add related config model to unified job
Use JobLaunchConfig model for launch RBAC check

Support credential overwrite behavior with multi-creds
  change modern manual launch to use merge behavior
Refactor JobLaunchSerializer, self.instance=None
Modularize job launch view to create "modern" data
Auto-create config object with every job
Add create schedule endpoint for jobs
This commit is contained in:
AlanCoding
2017-11-01 12:08:50 -04:00
parent cd8a4b4669
commit 34a8e0a9b6
49 changed files with 2343 additions and 884 deletions
@@ -9,6 +9,17 @@ from awx.main.models.workflow import (
import mock
@pytest.fixture
def credential():
ssh_type = CredentialType.defaults['ssh']()
return Credential(
id=43,
name='example-cred',
credential_type=ssh_type,
inputs={'username': 'asdf', 'password': 'asdf'}
)
class TestWorkflowJobInheritNodesMixin():
class TestCreateWorkflowJobNodes():
@pytest.fixture
@@ -123,55 +134,60 @@ def job_node_no_prompts(workflow_job_unit, jt_ask):
@pytest.fixture
def job_node_with_prompts(job_node_no_prompts):
def job_node_with_prompts(job_node_no_prompts, mocker):
job_node_no_prompts.char_prompts = example_prompts
job_node_no_prompts.inventory = Inventory(name='example-inv')
ssh_type = CredentialType.defaults['ssh']()
job_node_no_prompts.credential = Credential(
name='example-inv',
credential_type=ssh_type,
inputs={'username': 'asdf', 'password': 'asdf'}
)
job_node_no_prompts.inventory = Inventory(name='example-inv', id=45)
job_node_no_prompts.inventory_id = 45
return job_node_no_prompts
@pytest.fixture
def wfjt_node_no_prompts(workflow_job_template_unit, jt_ask):
return WorkflowJobTemplateNode(workflow_job_template=workflow_job_template_unit, unified_job_template=jt_ask)
node = WorkflowJobTemplateNode(
workflow_job_template=workflow_job_template_unit,
unified_job_template=jt_ask
)
return node
@pytest.fixture
def wfjt_node_with_prompts(wfjt_node_no_prompts):
def wfjt_node_with_prompts(wfjt_node_no_prompts, mocker):
wfjt_node_no_prompts.char_prompts = example_prompts
wfjt_node_no_prompts.inventory = Inventory(name='example-inv')
ssh_type = CredentialType.defaults['ssh']()
wfjt_node_no_prompts.credential = Credential(
name='example-inv',
credential_type=ssh_type,
inputs={'username': 'asdf', 'password': 'asdf'}
)
return wfjt_node_no_prompts
def test_node_getter_and_setters():
node = WorkflowJobTemplateNode()
node.job_type = 'check'
assert node.char_prompts['job_type'] == 'check'
assert node.job_type == 'check'
class TestWorkflowJobCreate:
def test_create_no_prompts(self, wfjt_node_no_prompts, workflow_job_unit, mocker):
mock_create = mocker.MagicMock()
with mocker.patch('awx.main.models.WorkflowJobNode.objects.create', mock_create):
wfjt_node_no_prompts.create_workflow_job_node(workflow_job=workflow_job_unit)
mock_create.assert_called_once_with(
extra_data={},
survey_passwords={},
char_prompts=wfjt_node_no_prompts.char_prompts,
inventory=None, credential=None,
inventory=None,
unified_job_template=wfjt_node_no_prompts.unified_job_template,
workflow_job=workflow_job_unit)
def test_create_with_prompts(self, wfjt_node_with_prompts, workflow_job_unit, mocker):
def test_create_with_prompts(self, wfjt_node_with_prompts, workflow_job_unit, credential, mocker):
mock_create = mocker.MagicMock()
with mocker.patch('awx.main.models.WorkflowJobNode.objects.create', mock_create):
wfjt_node_with_prompts.create_workflow_job_node(workflow_job=workflow_job_unit)
wfjt_node_with_prompts.create_workflow_job_node(
workflow_job=workflow_job_unit
)
mock_create.assert_called_once_with(
extra_data={},
survey_passwords={},
char_prompts=wfjt_node_with_prompts.char_prompts,
inventory=wfjt_node_with_prompts.inventory,
credential=wfjt_node_with_prompts.credential,
unified_job_template=wfjt_node_with_prompts.unified_job_template,
workflow_job=workflow_job_unit)
@@ -196,7 +212,7 @@ class TestWorkflowJobNodeJobKWARGS:
def test_char_prompts_and_res_node_prompts(self, job_node_with_prompts):
# TBD: properly handle multicred credential assignment
expect_kwargs = dict(
inventory=job_node_with_prompts.inventory.pk,
inventory=job_node_with_prompts.inventory,
**example_prompts)
expect_kwargs.update(self.kwargs_base)
assert job_node_with_prompts.get_job_kwargs() == expect_kwargs
@@ -205,7 +221,7 @@ class TestWorkflowJobNodeJobKWARGS:
# TBD: properly handle multicred credential assignment
job_node_with_prompts.unified_job_template.ask_inventory_on_launch = False
job_node_with_prompts.unified_job_template.ask_job_type_on_launch = False
expect_kwargs = dict(inventory=job_node_with_prompts.inventory.pk,
expect_kwargs = dict(inventory=job_node_with_prompts.inventory,
**example_prompts)
expect_kwargs.update(self.kwargs_base)
expect_kwargs.pop('inventory')
@@ -217,27 +233,5 @@ class TestWorkflowJobNodeJobKWARGS:
assert job_node_no_prompts.get_job_kwargs() == self.kwargs_base
class TestWorkflowWarnings:
"""
Tests of warnings that show user errors in the construction of a workflow
"""
def test_no_warn_project_node_no_prompts(self, job_node_no_prompts, project_unit):
job_node_no_prompts.unified_job_template = project_unit
assert job_node_no_prompts.get_prompts_warnings() == {}
def test_warn_project_node_reject_all_prompts(self, job_node_with_prompts, project_unit):
job_node_with_prompts.unified_job_template = project_unit
assert 'ignored' in job_node_with_prompts.get_prompts_warnings()
assert 'all' in job_node_with_prompts.get_prompts_warnings()['ignored']
def test_no_warn_accept_all_prompts(self, job_node_with_prompts):
assert job_node_with_prompts.get_prompts_warnings() == {}
def test_warn_reject_some_prompts(self, job_node_with_prompts):
job_node_with_prompts.unified_job_template.ask_credential_on_launch = False
job_node_with_prompts.unified_job_template.ask_job_type_on_launch = False
assert 'ignored' in job_node_with_prompts.get_prompts_warnings()
assert 'job_type' in job_node_with_prompts.get_prompts_warnings()['ignored']
assert len(job_node_with_prompts.get_prompts_warnings()['ignored']) == 1
def test_ask_mapping_integrity():
assert WorkflowJobTemplate.ask_mapping.keys() == ['extra_vars']