Function check_nodes

tests/integration/util.py:11–31

Helper function for checking nodes in cartography integration tests. Returns the result of a neo4j match query on the given node label and the given list of attributes as a set of tuples.

(
    neo4j_session: neo4j.Session,
    node_label: str,
    attrs: List[str],
)

def check_nodes(
    neo4j_session: neo4j.Session,
    node_label: str,
    attrs: List[str],
) -> Optional[Set[Tuple[Any, ...]]]:
    """
    Helper function for checking nodes in cartography integration tests.
    Returns the result of a neo4j match query on the given node label and the given list of attributes as a set of
    tuples.
    """
    if not attrs:
        raise ValueError(
            "`attrs` passed to check_nodes() must have at least one element.",
        )

    attrs = ", ".join(f"n.{attr}" for attr in attrs)
    query_template = Template("MATCH (n:$NodeLabel) RETURN $Attrs")
    result = neo4j_session.run(
        query_template.safe_substitute(NodeLabel=node_label, Attrs=attrs),
    )
    return {tuple(row.values()) for row in result}


def check_rels(