selector.py 文件源码

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