diff options
Diffstat (limited to '.venv/lib/python3.12/site-packages/networkx/classes/tests/dispatch_interface.py')
| -rw-r--r-- | .venv/lib/python3.12/site-packages/networkx/classes/tests/dispatch_interface.py | 185 |
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() |