From bdd0b9e4d92903f3a69636c92f39124803ec78ed Mon Sep 17 00:00:00 2001 From: AlanCoding Date: Thu, 12 Mar 2020 15:44:35 -0400 Subject: [PATCH] Add more tests for recent WFJT module issues --- .../test/awx/test_workflow_template.py | 99 ++++++++++++++++++- 1 file changed, 98 insertions(+), 1 deletion(-) diff --git a/awx_collection/test/awx/test_workflow_template.py b/awx_collection/test/awx/test_workflow_template.py index 47679ac21e..e58a5d5d92 100644 --- a/awx_collection/test/awx/test_workflow_template.py +++ b/awx_collection/test/awx/test_workflow_template.py @@ -3,7 +3,10 @@ __metaclass__ = type import pytest -from awx.main.models import WorkflowJobTemplate +from awx.main.models import ( + WorkflowJobTemplate, JobTemplate, Project, InventorySource, + Inventory, WorkflowJobTemplateNode +) @pytest.mark.django_db @@ -33,3 +36,97 @@ def test_create_workflow_job_template(run_module, admin_user, organization): } assert wfjt.organization_id == organization.id + + +@pytest.mark.django_db +def test_with_nested_workflow(run_module, admin_user, organization): + wfjt1 = WorkflowJobTemplate.objects.create(name='first', organization=organization) + + result = run_module('tower_workflow_template', { + 'name': 'foo-workflow', + 'organization': organization.name, + 'schema': [ + {'workflow': wfjt1.name} + ], + 'state': 'present' + }, admin_user) + assert not result.get('failed', False), result.get('msg', result) + + wfjt = WorkflowJobTemplate.objects.get(name='foo-workflow') + node = wfjt.workflow_nodes.first() + assert node is not None + assert node.unified_job_template == wfjt1 + + +@pytest.mark.django_db +def test_schema_with_branches(run_module, admin_user, organization): + + proj = Project.objects.create(organization=organization, name='Ansible Examples') + inv = Inventory.objects.create(organization=organization, name='test-inv') + jt = JobTemplate.objects.create( + project=proj, + playbook='helloworld.yml', + inventory=inv, + name='Hello world' + ) + inv_src = InventorySource.objects.create( + inventory=inv, + name='AWS servers' + ) + + result = run_module('tower_workflow_template', { + 'name': 'foo-workflow', + 'organization': organization.name, + 'schema': [ + { + 'job_template': 'Hello world', + 'failure': [ + { + 'inventory_source': 'AWS servers', + 'success': [ + { + 'project': 'Ansible Examples', + 'always': [ + { + 'job_template': "Hello world" + } + ] + } + ] + } + ] + } + ], + 'state': 'present' + }, admin_user) + assert not result.get('failed', False), result.get('msg', result) + + wfjt = WorkflowJobTemplate.objects.get(name='foo-workflow') + root_nodes = wfjt.workflow_nodes.filter(**{ + '%ss_success__isnull' % WorkflowJobTemplateNode.__name__.lower(): True, + '%ss_failure__isnull' % WorkflowJobTemplateNode.__name__.lower(): True, + '%ss_always__isnull' % WorkflowJobTemplateNode.__name__.lower(): True, + }) + assert len(root_nodes) == 1 + node = root_nodes[0] + assert node.unified_job_template == jt + second = node.failure_nodes.first() + assert second.unified_job_template == inv_src + third = second.success_nodes.first() + assert third.unified_job_template == proj + fourth = third.always_nodes.first() + assert fourth.unified_job_template == jt + + +@pytest.mark.django_db +def test_with_missing_ujt(run_module, admin_user, organization): + result = run_module('tower_workflow_template', { + 'name': 'foo-workflow', + 'organization': organization.name, + 'schema': [ + {'foo': 'bar'} + ], + 'state': 'present' + }, admin_user) + assert result.get('failed', False), result + assert 'You should provide exactly one of the attributes job_template,' in result['msg']