From 8b23ff71b4854aaf63221e92894bb63ad1e6f746 Mon Sep 17 00:00:00 2001 From: beeankha Date: Thu, 22 Aug 2019 15:54:51 -0400 Subject: [PATCH] Update/add more functional tests --- awx/main/signals.py | 2 +- .../functional/api/test_workflow_node.py | 96 ++++++++++++++++--- 2 files changed, 85 insertions(+), 13 deletions(-) diff --git a/awx/main/signals.py b/awx/main/signals.py index a655c19b3b..9846b11dd3 100644 --- a/awx/main/signals.py +++ b/awx/main/signals.py @@ -658,7 +658,7 @@ def delete_approval_node_type_change(sender, instance, **kwargs): old.unified_job_template.delete() -@receiver(post_delete, sender=WorkflowApprovalTemplate) +@receiver(pre_delete, sender=WorkflowApprovalTemplate) def deny_orphaned_approvals(sender, instance, **kwargs): for approval in WorkflowApproval.objects.filter(workflow_approval_template=instance, status='pending'): approval.deny() diff --git a/awx/main/tests/functional/api/test_workflow_node.py b/awx/main/tests/functional/api/test_workflow_node.py index 4c366bfe19..5c5e7de2e0 100644 --- a/awx/main/tests/functional/api/test_workflow_node.py +++ b/awx/main/tests/functional/api/test_workflow_node.py @@ -7,10 +7,12 @@ from awx.main.models.activity_stream import ActivityStream from awx.main.models.jobs import JobTemplate from awx.main.models.workflow import ( WorkflowApprovalTemplate, + WorkflowJob, WorkflowJobTemplate, WorkflowJobTemplateNode, ) from awx.main.models.credential import Credential +from awx.main.scheduler import TaskManager @pytest.fixture @@ -83,18 +85,19 @@ class TestApprovalNodes(): assert approval_node.unified_job_template.timeout==0 def test_approval_node_creation_failure(self, post, approval_node, admin_user): + # This test leaves off a required param to assert that user will get a 400. url = reverse('api:workflow_job_template_node_create_approval', kwargs={'pk': approval_node.pk, 'version': 'v2'}) - post(url, {'name': '', 'description': 'Approval Node', 'timeout': 0}, - user=admin_user, expect=400) - # Leave off a required param to assert that you get a 400 + r = post(url, {'name': '', 'description': 'Approval Node', 'timeout': 0}, + user=admin_user, expect=400) approval_node = WorkflowJobTemplateNode.objects.get(pk=approval_node.pk) assert isinstance(approval_node.unified_job_template, WorkflowApprovalTemplate) is False + assert {'name': ['This field may not be blank.']} == json.loads(r.content) @pytest.mark.parametrize("is_admin, is_org_admin, status", [ - [True, False, 200], - [False, False, 403], - [False, True, 200], + [True, False, 200], # if they're a WFJT admin, they get a 200 + [False, False, 403], # if they're not a WFJT *nor* org admin, they get a 403 + [False, True, 200], # if they're an organization admin, they get a 200 ]) def test_approval_node_creation_rbac(self, post, approval_node, alice, is_admin, is_org_admin, status): url = reverse('api:workflow_job_template_node_create_approval', @@ -107,7 +110,7 @@ class TestApprovalNodes(): user=alice, expect=status) @pytest.mark.django_db - def test_approval_node_exists(self, post, approval_node, admin_user, get): + def test_approval_node_exists(self, post, admin_user, get): workflow_job_template = WorkflowJobTemplate.objects.create() approval_node = WorkflowJobTemplateNode.objects.create( workflow_job_template=workflow_job_template @@ -132,11 +135,41 @@ class TestApprovalNodes(): qs2 = ActivityStream.objects.filter(organization__isnull=True) assert qs2.count() == 5 - assert qs2[0].operation == 'create' - assert qs2[1].operation == 'create' - assert qs2[2].operation == 'create' - assert qs2[3].operation == 'create' - assert qs2[4].operation == 'update' + assert list(qs2.values_list('operation', 'object1')) == [('create', 'user'), + ('create', 'workflow_job_template'), + ('create', 'workflow_job_template_node'), + ('create', 'workflow_approval_template'), + ('update', 'workflow_job_template_node'), + ] + + @pytest.mark.django_db + def test_approval_node_actions(self, post, admin_user, job_template): + # This test ensures that a user (with permissions to do so) can approve/deny + # workflow approvals. Also asserts that trying to approve/deny approvals + # that have already been dealt with will throw an error. + wfjt = WorkflowJobTemplate.objects.create(name='foobar') + node = wfjt.workflow_nodes.create(unified_job_template=job_template) + url = reverse('api:workflow_job_template_node_create_approval', + kwargs={'pk': node.pk, 'version': 'v2'}) + post(url, {'name': 'Approve/Deny Test', 'description': '', 'timeout': 0}, + user=admin_user, expect=200) + post(reverse('api:workflow_job_template_launch', kwargs={'pk': wfjt.pk}), + user=admin_user, expect=201) + wf_job = WorkflowJob.objects.first() + TaskManager().schedule() + TaskManager().schedule() + wfj_node = wf_job.workflow_nodes.first() + approval = wfj_node.job + assert approval.name == 'Approve/Deny Test' + post(reverse('api:workflow_approval_approve', kwargs={'pk': approval.pk}), + user=admin_user, expect=204) + # Test that there is an activity stream entry that was created for the "approve" action. + qs = ActivityStream.objects.order_by('-timestamp').first() + assert qs.object1 == 'workflow_approval' + assert qs.changes == '{"status": ["pending", "successful"]}' + assert qs.operation == 'update' + post(reverse('api:workflow_approval_approve', kwargs={'pk': approval.pk}), + user=admin_user, expect=403) def test_approval_node_cleanup(self, post, approval_node, admin_user, get): workflow_job_template = WorkflowJobTemplate.objects.create() @@ -148,9 +181,48 @@ class TestApprovalNodes(): post(url, {'name': 'URL Test', 'description': 'An approval', 'timeout': 0}, user=admin_user) + assert WorkflowApprovalTemplate.objects.count() == 1 workflow_job_template.delete() + assert WorkflowApprovalTemplate.objects.count() == 0 get(url, admin_user, expect=404) + def test_changed_approval_deletion(self, post, approval_node, admin_user, workflow_job_template, job_template): + # This test verifies that when an approval node changes into something else + # (in this case, a job template), then the previously-set WorkflowApprovalTemplate + # is automatically deleted. + workflow_job_template = WorkflowJobTemplate.objects.create() + approval_node = WorkflowJobTemplateNode.objects.create( + workflow_job_template=workflow_job_template + ) + url = reverse('api:workflow_job_template_node_create_approval', + kwargs={'pk': approval_node.pk, 'version': 'v2'}) + post(url, {'name': 'URL Test', 'description': 'An approval', 'timeout': 0}, + user=admin_user) + assert WorkflowApprovalTemplate.objects.count() == 1 + approval_node.unified_job_template = job_template + approval_node.save() + assert WorkflowApprovalTemplate.objects.count() == 0 + + def test_deleted_approval_denial(self, post, approval_node, admin_user, workflow_job_template): + # Verifying that when a WorkflowApprovalTemplate is deleted, any/all of + # its pending approvals are auto-denied (vs left in 'pending' state). + workflow_job_template = WorkflowJobTemplate.objects.create() + approval_node = WorkflowJobTemplateNode.objects.create( + workflow_job_template=workflow_job_template + ) + url = reverse('api:workflow_job_template_node_create_approval', + kwargs={'pk': approval_node.pk, 'version': 'v2'}) + post(url, {'name': 'URL Test', 'description': 'An approval', 'timeout': 0}, + user=admin_user) + assert WorkflowApprovalTemplate.objects.count() == 1 + approval_template = WorkflowApprovalTemplate.objects.first() + approval = approval_template.create_unified_job() + approval.status = 'pending' + approval.save() + approval_template.delete() + approval.refresh_from_db() + assert approval.status == 'failed' + @pytest.mark.django_db class TestExclusiveRelationshipEnforcement():