.venv/Lib/site-packages/networkx/classes/tests/dispatch_interface.py

# This file contains utilities for testing the dispatching feature

# A full test of all dispatchable algorithms is performed by
# modifying the pytest invocation and setting an environment variable
# NETWORKX_TEST_BACKEND=nx_loopback pytest
# This is comprehensive, but only tests the `test_override_dispatch`
# function in networkx.classes.backends.

# To test the `_dispatchable` function directly, several tests scattered throughout
# NetworkX have been augmented to test normal and dispatch mode.
# Searching for `dispatch_interface` should locate the specific tests.

import networkx as nx
from networkx import DiGraph, Graph, MultiDiGraph, MultiGraph, PlanarEmbedding
from networkx.classes.reportviews import NodeView


class LoopbackGraph(Graph):
    __networkx_backend__ = "nx_loopback"


class LoopbackDiGraph(DiGraph):
    __networkx_backend__ = "nx_loopback"


class LoopbackMultiGraph(MultiGraph):
    __networkx_backend__ = "nx_loopback"


class LoopbackMultiDiGraph(MultiDiGraph):
    __networkx_backend__ = "nx_loopback"


class LoopbackPlanarEmbedding(PlanarEmbedding):
    __networkx_backend__ = "nx_loopback"


def convert(graph):
    if isinstance(graph, PlanarEmbedding):
        return LoopbackPlanarEmbedding(graph)
    if isinstance(graph, MultiDiGraph):
        return LoopbackMultiDiGraph(graph)
    if isinstance(graph, MultiGraph):
        return LoopbackMultiGraph(graph)
    if isinstance(graph, DiGraph):
        return LoopbackDiGraph(graph)
    if isinstance(graph, Graph):
        return LoopbackGraph(graph)
    raise TypeError(f"Unsupported type of graph: {type(graph)}")


class LoopbackBackendInterface:
    def __getattr__(self, item):
        try:
            return nx.utils.backends._registered_algorithms[item].orig_func
        except KeyError:
            raise AttributeError(item) from None

    @staticmethod
    def convert_from_nx(
        graph,
        *,
        edge_attrs=None,
        node_attrs=None,
        preserve_edge_attrs=None,
        preserve_node_attrs=None,
        preserve_graph_attrs=None,
        name=None,
        graph_name=None,
    ):
        if name in {
            # Raise if input graph changes. See test_dag.py::test_topological_sort6
            "lexicographical_topological_sort",
            "topological_generations",
            "topological_sort",
            # Would be nice to some day avoid these cutoffs of full testing
        }:
            return graph
        if isinstance(graph, NodeView):
            # Convert to a Graph with only nodes (no edges)
            new_graph = Graph()
            new_graph.add_nodes_from(graph.items())
            graph = new_graph
            G = LoopbackGraph()
        elif not isinstance(graph, Graph):
            raise TypeError(
                f"Bad type for graph argument {graph_name} in {name}: {type(graph)}"
            )
        elif graph.__class__ in {Graph, LoopbackGraph}:
            G = LoopbackGraph()
        elif graph.__class__ in {DiGraph, LoopbackDiGraph}:
            G = LoopbackDiGraph()
        elif graph.__class__ in {MultiGraph, LoopbackMultiGraph}:
            G = LoopbackMultiGraph()
        elif graph.__class__ in {MultiDiGraph, LoopbackMultiDiGraph}:
            G = LoopbackMultiDiGraph()
        elif graph.__class__ in {PlanarEmbedding, LoopbackPlanarEmbedding}:
            G = LoopbackDiGraph()  # or LoopbackPlanarEmbedding
        else:
            # Would be nice to handle these better some day
            # nx.algorithms.approximation.kcomponents._AntiGraph
            # nx.classes.tests.test_multidigraph.MultiDiGraphSubClass
            # nx.classes.tests.test_multigraph.MultiGraphSubClass
            G = graph.__class__()

        if preserve_graph_attrs:
            G.graph.update(graph.graph)

        # add nodes
        G.add_nodes_from(graph)
        if preserve_node_attrs:
            for n, dd in G._node.items():
                dd.update(graph.nodes[n])
        elif node_attrs:
            for n, dd in G._node.items():
                dd.update(
                    (attr, graph._node[n].get(attr, default))
                    for attr, default in node_attrs.items()
                    if default is not None or attr in graph._node[n]
                )

        # tools to build datadict and keydict
        if preserve_edge_attrs:

            def G_new_datadict(old_dd):
                return G.edge_attr_dict_factory(old_dd)
        elif edge_attrs:

            def G_new_datadict(old_dd):
                return G.edge_attr_dict_factory(
                    (attr, old_dd.get(attr, default))
                    for attr, default in edge_attrs.items()
                    if default is not None or attr in old_dd
                )
        else:

            def G_new_datadict(old_dd):
                return G.edge_attr_dict_factory()

        if G.is_multigraph():

            def G_new_inner(keydict):
                kd = G.adjlist_inner_dict_factory(
                    (k, G_new_datadict(dd)) for k, dd in keydict.items()
                )
                return kd
        else:
            G_new_inner = G_new_datadict

        # add edges keeping the same order in _adj and _pred
        G_adj = G._adj
        if G.is_directed():
            for n, nbrs in graph._adj.items():
                G_adj[n].update((nbr, G_new_inner(dd)) for nbr, dd in nbrs.items())
            # ensure same datadict for pred and adj; and pred order of graph._pred
            G_pred = G._pred
            for n, nbrs in graph._pred.items():
                G_pred[n].update((nbr, G_adj[nbr][n]) for nbr in nbrs)
        else:  # undirected
            for n, nbrs in graph._adj.items():
                # ensure same datadict for both ways; and adj order of graph._adj
                G_adj[n].update(
                    (nbr, G_adj[nbr][n] if n in G_adj[nbr] else G_new_inner(dd))
                    for nbr, dd in nbrs.items()
                )

        return G

    @staticmethod
    def convert_to_nx(obj, *, name=None):
        return obj

    @staticmethod
    def on_start_tests(items):
        # Verify that items can be xfailed
        for item in items:
            assert hasattr(item, "add_marker")

    def can_run(self, name, args, kwargs):
        # It is unnecessary to define this function if algorithms are fully supported.
        # We include it for illustration purposes.
        return hasattr(self, name)


backend_interface = LoopbackBackendInterface()