Merge pull request #2824 from chrismeyersfsu/workflow-convergence_enforce2

enforce 1 edge between 2 nodes constraint

Reviewed-by: https://github.com/softwarefactory-project-zuul[bot]

awx/api/views/__init__.py

@@ -9,6 +9,7 @@ import time
 import socket
 import sys
 import requests
+import functools
 from base64 import b64encode
 from collections import OrderedDict, Iterable
 import six
@@ -2957,6 +2958,18 @@ class WorkflowJobTemplateNodeChildrenBaseList(WorkflowsEnforcementMixin, Enforce
         if parent.id == sub.id:
             return {"Error": _("Cycle detected.")}
 
+        '''
+        Look for parent->child connection in all relationships except the relationship that is
+        attempting to be added; because it's ok to re-add the relationship
+        '''
+        relationships = ['success_nodes', 'failure_nodes', 'always_nodes']
+        relationships.remove(self.relationship)
+        qs = functools.reduce(lambda x, y: (x | y),
+                              (Q(**{'{}__in'.format(rel): [sub.id]}) for rel in relationships))
+
+        if WorkflowJobTemplateNode.objects.filter(Q(pk=parent.id) & qs).exists():
+            return {"Error": _("Relationship not allowed.")}
+
         parent_node_type_relationship = getattr(parent, self.relationship)
         parent_node_type_relationship.add(sub)
 

awx/main/tests/functional/api/test_workflow_node.py

@@ -1,4 +1,5 @@
 import pytest
+import json
 
 from awx.api.versioning import reverse
 
@@ -75,6 +76,52 @@ def test_node_accepts_prompted_fields(inventory, project, workflow_job_template,
          user=admin_user, expect=201)
 
 
+@pytest.mark.django_db
+class TestExclusiveRelationshipEnforcement():
+    @pytest.fixture
+    def n1(self, workflow_job_template):
+        return WorkflowJobTemplateNode.objects.create(workflow_job_template=workflow_job_template)
+
+    @pytest.fixture
+    def n2(self, workflow_job_template):
+        return WorkflowJobTemplateNode.objects.create(workflow_job_template=workflow_job_template)
+
+    def generate_url(self, relationship, id):
+        return reverse('api:workflow_job_template_node_{}_nodes_list'.format(relationship),
+                       kwargs={'pk': id})
+
+    relationship_permutations = [
+        ['success', 'failure', 'always'],
+        ['success', 'always', 'failure'],
+        ['failure', 'always', 'success'],
+        ['failure', 'success', 'always'],
+        ['always', 'success', 'failure'],
+        ['always', 'failure', 'success'],
+    ]
+
+    @pytest.mark.parametrize("relationships", relationship_permutations, ids=["-".join(item) for item in relationship_permutations])
+    def test_multi_connections_same_parent_disallowed(self, post, admin_user, n1, n2, relationships):
+        for index, relationship in enumerate(relationships):
+            r = post(self.generate_url(relationship, n1.id),
+                     data={'associate': True, 'id': n2.id},
+                     user=admin_user,
+                     expect=204 if index == 0 else 400)
+
+            if index != 0:
+                assert {'Error': 'Relationship not allowed.'} == json.loads(r.content)
+
+    @pytest.mark.parametrize("relationship", ['success', 'failure', 'always'])
+    def test_existing_relationship_allowed(self, post, admin_user, n1, n2, relationship):
+        post(self.generate_url(relationship, n1.id),
+             data={'associate': True, 'id': n2.id},
+             user=admin_user,
+             expect=204)
+        post(self.generate_url(relationship, n1.id),
+             data={'associate': True, 'id': n2.id},
+             user=admin_user,
+             expect=204)
+
+
 @pytest.mark.django_db
 class TestNodeCredentials:
     '''