External-Mirrors/awx
2
0
Fork 0
mirror of https://github.com/ansible/awx.git synced 2026-05-23 08:37:48 -02:30
Code Issues Packages Projects Releases Wiki Activity

do not deny existing workflow node relationships

Browse Source
This commit is contained in:
chris meyers
2018-11-29 13:29:16 -05:00
parent 916c92ffc7
commit d347a06e3d
2 changed files with 26 additions and 22 deletions
Download Patch File Download Diff File Expand all files Collapse all files

14
awx/api/views/__init__.py
View File

@@ -2957,12 +2957,16 @@ class WorkflowJobTemplateNodeChildrenBaseList(WorkflowsEnforcementMixin, Enforce
if parent.id == sub.id: if parent.id == sub.id:
return {"Error": _("Cycle detected.")} return {"Error": _("Cycle detected.")}
if parent.id == sub.id '''
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 = reduce(lambda x, y: (x | y),
(Q(**{'{}__in'.format(rel): [sub.id]}) for rel in relationships))
if WorkflowJobTemplateNode.objects.filter(Q(pk=parent.id) & if WorkflowJobTemplateNode.objects.filter(Q(pk=parent.id) & qs).exists():
Q(success_nodes__in=[sub.id]) |
Q(failure_nodes__in=[sub.id]) |
Q(always_nodes__in=[sub.id])).exists():
return {"Error": _("Relationship not allowed.")} return {"Error": _("Relationship not allowed.")}
parent_node_type_relationship = getattr(parent, self.relationship) parent_node_type_relationship = getattr(parent, self.relationship)

34
awx/main/tests/functional/api/test_workflow_node.py
View File

@@ -77,7 +77,7 @@ def test_node_accepts_prompted_fields(inventory, project, workflow_job_template,
@pytest.mark.django_db @pytest.mark.django_db
class TestExclusivePathEnforcement(): class TestExclusiveRelationshipEnforcement():
@pytest.fixture @pytest.fixture
def n1(self, workflow_job_template): def n1(self, workflow_job_template):
return WorkflowJobTemplateNode.objects.create(workflow_job_template=workflow_job_template) return WorkflowJobTemplateNode.objects.create(workflow_job_template=workflow_job_template)
@@ -86,11 +86,11 @@ class TestExclusivePathEnforcement():
def n2(self, workflow_job_template): def n2(self, workflow_job_template):
return WorkflowJobTemplateNode.objects.create(workflow_job_template=workflow_job_template) return WorkflowJobTemplateNode.objects.create(workflow_job_template=workflow_job_template)
def generate_url(self, path, id): def generate_url(self, relationship, id):
return reverse('api:workflow_job_template_node_{}_nodes_list'.format(path), return reverse('api:workflow_job_template_node_{}_nodes_list'.format(relationship),
kwargs={'pk': id}) kwargs={'pk': id})
path_permutations = [ relationship_permutations = [
['success', 'failure', 'always'], ['success', 'failure', 'always'],
['success', 'always', 'failure'], ['success', 'always', 'failure'],
['failure', 'always', 'success'], ['failure', 'always', 'success'],
@@ -99,10 +99,10 @@ class TestExclusivePathEnforcement():
['always', 'failure', 'success'], ['always', 'failure', 'success'],
] ]
@pytest.mark.parametrize("paths", path_permutations, ids=["-".join(item) for item in path_permutations]) @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, paths): def test_multi_connections_same_parent_disallowed(self, post, admin_user, n1, n2, relationships):
for index, path in enumerate(paths): for index, relationship in enumerate(relationships):
r = post(self.generate_url(path, n1.id), r = post(self.generate_url(relationship, n1.id),
data={'associate': True, 'id': n2.id}, data={'associate': True, 'id': n2.id},
user=admin_user, user=admin_user,
expect=204 if index == 0 else 400) expect=204 if index == 0 else 400)
@@ -110,16 +110,16 @@ class TestExclusivePathEnforcement():
if index != 0: if index != 0:
assert {'Error': 'Relationship not allowed.'} == json.loads(r.content) assert {'Error': 'Relationship not allowed.'} == json.loads(r.content)
@pytest.mark.parametrize("relationship", ['success', 'failure', 'always']): @pytest.mark.parametrize("relationship", ['success', 'failure', 'always'])
def test_existing_relationship_allowed(self, post, admin_user, n1, n2, relationship): def test_existing_relationship_allowed(self, post, admin_user, n1, n2, relationship):
r = post(self.generate_url(path, n1.id), post(self.generate_url(relationship, n1.id),
data={'associate': True, 'id': n2.id}, data={'associate': True, 'id': n2.id},
user=admin_user, user=admin_user,
expect=204) expect=204)
r = post(self.generate_url(path, n1.id), post(self.generate_url(relationship, n1.id),
data={'associate': True, 'id': n2.id}, data={'associate': True, 'id': n2.id},
user=admin_user, user=admin_user,
expect=200) expect=204)
@pytest.mark.django_db @pytest.mark.django_db