def get_nodes_from_spec(graph, spec):
select_parents = spec['select_parents']
select_children = spec['select_children']
qualified_node_name = spec['qualified_node_name']
selected_nodes = set(get_nodes_by_qualified_name(graph,
qualified_node_name))
additional_nodes = set()
test_nodes = set()
if select_parents:
for node in selected_nodes:
parent_nodes = nx.ancestors(graph, node)
additional_nodes.update(parent_nodes)
if select_children:
for node in selected_nodes:
child_nodes = nx.descendants(graph, node)
additional_nodes.update(child_nodes)
model_nodes = selected_nodes | additional_nodes
for node in model_nodes:
# include tests that depend on this node. if we aren't running tests,
# they'll be filtered out later.
child_tests = [n for n in graph.successors(node)
if graph.node.get(n).get('resource_type') ==
NodeType.Test]
test_nodes.update(child_tests)
return model_nodes | test_nodes
评论列表
文章目录