def sprint_adj_db_delta()

in openr/py/openr/cli/utils/utils.py [0:0]

• 35 lines of code
• 12 McCabe index (conditional complexity)

def sprint_adj_db_delta(new_adj_db, old_adj_db):
    """given serialized adjacency database, print neighbors delta as
        compared to the supplied global state

    :param new_adj_db openr_types.AdjacencyDatabase: latest from kv store
    :param old_adj_db openr_types.AdjacencyDatabase: last one we had

    :return [str]: list of string to be printed
    """

    # check for deltas between old and new
    # first check for changes in the adjacencies lists
    adj_list_deltas = find_adj_list_deltas(
        old_adj_db.adjacencies, new_adj_db.adjacencies
    )

    strs = []

    for change_type, old_adj, new_adj in adj_list_deltas:
        if change_type == "NEIGHBOR_DOWN":
            strs.append(
                "{}: {} via {}".format(
                    change_type, old_adj.otherNodeName, old_adj.ifName
                )
            )
        if change_type == "NEIGHBOR_UP" or change_type == "NEIGHBOR_UPDATE":
            strs.append(
                "{}: {} via {}\n{}".format(
                    change_type,
                    new_adj.otherNodeName,
                    new_adj.ifName,
                    sprint_adj_delta(old_adj, new_adj),
                )
            )

    # check for other adjDB changes
    old_db_dict = copy.copy(old_adj_db).__dict__
    old_db_dict.pop("adjacencies", None)
    old_db_dict.pop("perfEvents", None)
    new_db_dict = copy.copy(new_adj_db).__dict__
    new_db_dict.pop("adjacencies", None)
    new_db_dict.pop("perfEvents", None)
    if new_db_dict != old_db_dict:
        rows = []
        strs.append("ADJ_DB_UPDATE: {}".format(new_adj_db.thisNodeName))
        for k in sorted(new_db_dict.keys()):
            if old_db_dict.get(k) != new_db_dict.get(k):
                rows.append([k, old_db_dict.get(k), "-->", new_db_dict.get(k)])
        strs.append(printing.render_horizontal_table(rows, tablefmt="plain"))

    return strs