External-Mirrors/awx
2
0
Fork 0
mirror of https://github.com/ansible/awx.git synced 2026-02-16 18:50:04 -03:30
Code Issues Packages Projects Releases Wiki Activity

handle edge case ring cycle

Browse Source
This commit is contained in:
chris meyers
2018-11-01 12:45:43 -04:00
parent ea80e244b9
commit 7b751ea2ff
2 changed files with 14 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
awx/main/scheduler/dag_simple.py
View File

@@ -127,6 +127,9 @@ class SimpleDAG(object):
path = set([]) path = set([])
stack = node_objs stack = node_objs
if len(self.nodes) != 0 and len(node_objs) == 0:
return True
while stack: while stack:
node_obj = stack.pop() node_obj = stack.pop()

18
awx/main/tests/functional/models/test_workflow.py
View File

@@ -3,11 +3,17 @@
import pytest import pytest
# AWX # AWX
from awx.main.models.workflow import WorkflowJob, WorkflowJobNode, WorkflowJobTemplateNode, WorkflowJobTemplate from awx.main.models.workflow import (
WorkflowJob,
WorkflowJobNode,
WorkflowJobTemplateNode,
WorkflowJobTemplate,
)
from awx.main.models.jobs import JobTemplate, Job from awx.main.models.jobs import JobTemplate, Job
from awx.main.models.projects import ProjectUpdate from awx.main.models.projects import ProjectUpdate
from awx.main.scheduler.dag_workflow import WorkflowDAG from awx.main.scheduler.dag_workflow import WorkflowDAG
from awx.api.versioning import reverse from awx.api.versioning import reverse
from awx.api.views import WorkflowJobTemplateNodeSuccessNodesList
# Django # Django
from django.test import TransactionTestCase from django.test import TransactionTestCase
@@ -237,15 +243,11 @@ class TestWorkflowJobTemplate:
assert parent_qs[0] == wfjt.workflow_job_template_nodes.all()[1] assert parent_qs[0] == wfjt.workflow_job_template_nodes.all()[1]
def test_topology_validator(self, wfjt): def test_topology_validator(self, wfjt):
from awx.api.views import WorkflowJobTemplateNodeChildrenBaseList test_view = WorkflowJobTemplateNodeSuccessNodesList()
test_view = WorkflowJobTemplateNodeChildrenBaseList()
nodes = wfjt.workflow_job_template_nodes.all() nodes = wfjt.workflow_job_template_nodes.all()
node_assoc = WorkflowJobTemplateNode.objects.create(workflow_job_template=wfjt)
nodes[2].always_nodes.add(node_assoc)
# test cycle validation # test cycle validation
assert test_view.is_valid_relation(node_assoc, nodes[0]) == {'Error': 'Cycle detected.'} print(nodes[0].success_nodes.get(id=nodes[1].id).failure_nodes.get(id=nodes[2].id))
# test mutex validation assert test_view.is_valid_relation(nodes[2], nodes[0]) == {'Error': 'Cycle detected.'}
test_view.relationship = 'failure_nodes'
def test_always_success_failure_creation(self, wfjt, admin, get): def test_always_success_failure_creation(self, wfjt, admin, get):
wfjt_node = wfjt.workflow_job_template_nodes.all()[1] wfjt_node = wfjt.workflow_job_template_nodes.all()[1]