two version of R2R are hereHEADmaster

1 files changed, 185 insertions, 0 deletions

diff --git a/.venv/lib/python3.12/site-packages/networkx/classes/tests/dispatch_interface.py b/.venv/lib/python3.12/site-packages/networkx/classes/tests/dispatch_interface.py
new file mode 100644
index 00000000..5cc908d7
--- /dev/null
+++ b/.venv/lib/python3.12/site-packages/networkx/classes/tests/dispatch_interface.py
@@ -0,0 +1,185 @@
+# 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()