| 123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225 |
- import pickle
- import pytest
- import networkx as nx
- sp = pytest.importorskip("scipy")
- pytest.importorskip("numpy")
- @nx._dispatchable(implemented_by_nx=False)
- def _stub_func(G):
- raise NotImplementedError("_stub_func is a stub")
- def test_dispatch_kwds_vs_args():
- G = nx.path_graph(4)
- nx.pagerank(G)
- nx.pagerank(G=G)
- with pytest.raises(TypeError):
- nx.pagerank()
- def test_pickle():
- count = 0
- for name, func in nx.utils.backends._registered_algorithms.items():
- pickled = pickle.dumps(func.__wrapped__)
- assert pickle.loads(pickled) is func.__wrapped__
- try:
- # Some functions can't be pickled, but it's not b/c of _dispatchable
- pickled = pickle.dumps(func)
- except pickle.PicklingError:
- continue
- assert pickle.loads(pickled) is func
- count += 1
- assert count > 0
- assert pickle.loads(pickle.dumps(nx.inverse_line_graph)) is nx.inverse_line_graph
- @pytest.mark.skipif(
- "not nx.config.backend_priority.algos "
- "or nx.config.backend_priority.algos[0] != 'nx_loopback'"
- )
- def test_graph_converter_needs_backend():
- # When testing, `nx.from_scipy_sparse_array` will *always* call the backend
- # implementation if it's implemented. If `backend=` isn't given, then the result
- # will be converted back to NetworkX via `convert_to_nx`.
- # If not testing, then calling `nx.from_scipy_sparse_array` w/o `backend=` will
- # always call the original version. `backend=` is *required* to call the backend.
- from networkx.classes.tests.dispatch_interface import (
- LoopbackBackendInterface,
- LoopbackGraph,
- )
- A = sp.sparse.coo_array([[0, 3, 2], [3, 0, 1], [2, 1, 0]])
- side_effects = []
- def from_scipy_sparse_array(self, *args, **kwargs):
- side_effects.append(1) # Just to prove this was called
- return self.convert_from_nx(
- self.__getattr__("from_scipy_sparse_array")(*args, **kwargs),
- preserve_edge_attrs=True,
- preserve_node_attrs=True,
- preserve_graph_attrs=True,
- )
- @staticmethod
- def convert_to_nx(obj, *, name=None):
- if type(obj) is nx.Graph:
- return obj
- return nx.Graph(obj)
- # *This mutates LoopbackBackendInterface!*
- orig_convert_to_nx = LoopbackBackendInterface.convert_to_nx
- LoopbackBackendInterface.convert_to_nx = convert_to_nx
- LoopbackBackendInterface.from_scipy_sparse_array = from_scipy_sparse_array
- try:
- assert side_effects == []
- assert type(nx.from_scipy_sparse_array(A)) is nx.Graph
- assert side_effects == [1]
- assert (
- type(nx.from_scipy_sparse_array(A, backend="nx_loopback")) is LoopbackGraph
- )
- assert side_effects == [1, 1]
- # backend="networkx" is default implementation
- assert type(nx.from_scipy_sparse_array(A, backend="networkx")) is nx.Graph
- assert side_effects == [1, 1]
- finally:
- LoopbackBackendInterface.convert_to_nx = staticmethod(orig_convert_to_nx)
- del LoopbackBackendInterface.from_scipy_sparse_array
- with pytest.raises(ImportError, match="backend is not installed"):
- nx.from_scipy_sparse_array(A, backend="bad-backend-name")
- @pytest.mark.skipif(
- "not nx.config.backend_priority.algos "
- "or nx.config.backend_priority.algos[0] != 'nx_loopback'"
- )
- def test_networkx_backend():
- """Test using `backend="networkx"` in a dispatchable function."""
- # (Implementing this test is harder than it should be)
- from networkx.classes.tests.dispatch_interface import (
- LoopbackBackendInterface,
- LoopbackGraph,
- )
- G = LoopbackGraph()
- G.add_edges_from([(0, 1), (1, 2), (1, 3), (2, 4)])
- @staticmethod
- def convert_to_nx(obj, *, name=None):
- if isinstance(obj, LoopbackGraph):
- new_graph = nx.Graph()
- new_graph.__dict__.update(obj.__dict__)
- return new_graph
- return obj
- # *This mutates LoopbackBackendInterface!*
- # This uses the same trick as in the previous test.
- orig_convert_to_nx = LoopbackBackendInterface.convert_to_nx
- LoopbackBackendInterface.convert_to_nx = convert_to_nx
- try:
- G2 = nx.ego_graph(G, 0, backend="networkx")
- assert type(G2) is nx.Graph
- finally:
- LoopbackBackendInterface.convert_to_nx = staticmethod(orig_convert_to_nx)
- def test_dispatchable_are_functions():
- assert type(nx.pagerank) is type(nx.pagerank.orig_func)
- @pytest.mark.skipif("not nx.utils.backends.backends")
- def test_mixing_backend_graphs():
- from networkx.classes.tests import dispatch_interface
- G = nx.Graph()
- G.add_edge(1, 2)
- G.add_edge(2, 3)
- H = nx.Graph()
- H.add_edge(2, 3)
- rv = nx.intersection(G, H)
- assert set(nx.intersection(G, H)) == {2, 3}
- G2 = dispatch_interface.convert(G)
- H2 = dispatch_interface.convert(H)
- if "nx_loopback" in nx.config.backend_priority:
- # Auto-convert
- assert set(nx.intersection(G2, H)) == {2, 3}
- assert set(nx.intersection(G, H2)) == {2, 3}
- elif not nx.config.backend_priority and "nx_loopback" not in nx.config.backends:
- # G2 and H2 are backend objects for a backend that is not registered!
- with pytest.raises(ImportError, match="backend is not installed"):
- nx.intersection(G2, H)
- with pytest.raises(ImportError, match="backend is not installed"):
- nx.intersection(G, H2)
- # It would be nice to test passing graphs from *different* backends,
- # but we are not set up to do this yet.
- def test_bad_backend_name():
- """Using `backend=` raises with unknown backend even if there are no backends."""
- with pytest.raises(
- ImportError, match="'this_backend_does_not_exist' backend is not installed"
- ):
- nx.null_graph(backend="this_backend_does_not_exist")
- def test_not_implemented_by_nx():
- assert "networkx" in nx.pagerank.backends
- assert "networkx" not in _stub_func.backends
- if "nx_loopback" in nx.config.backends:
- from networkx.classes.tests.dispatch_interface import LoopbackBackendInterface
- def stub_func_implementation(G):
- return True
- LoopbackBackendInterface._stub_func = staticmethod(stub_func_implementation)
- try:
- assert _stub_func(nx.Graph()) is True
- finally:
- del LoopbackBackendInterface._stub_func
- with pytest.raises(NotImplementedError):
- _stub_func(nx.Graph())
- @pytest.mark.skipif(
- "not nx.config.backend_priority.algos "
- "or nx.config.backend_priority.algos[0] != 'nx_loopback'"
- )
- def test_dispatch_graph_new():
- from networkx.classes.tests.dispatch_interface import LoopbackGraph
- G = nx.Graph()
- assert not isinstance(G, LoopbackGraph)
- # `backend=` argument that gets passed to __init__ is ignored.
- # Best practice is that it should not be in the `.graph` dict.
- G = nx.Graph(backend="networkx")
- assert type(G) is nx.Graph
- assert "backend" not in G.graph
- G = nx.Graph(backend="nx_loopback")
- assert isinstance(G, LoopbackGraph)
- assert "backend" not in G.graph
- # Args are passed
- G1 = nx.Graph([(0, 1), (1, 2)])
- assert not isinstance(G1, LoopbackGraph)
- G2 = nx.Graph([(0, 1), (1, 2)], backend="nx_loopback")
- assert isinstance(G2, LoopbackGraph)
- assert nx.utils.misc.graphs_equal(G1, G2)
- # Test config for automatic usage
- with nx.config.backend_priority(classes=["nx_loopback"]):
- G = nx.Graph()
- assert isinstance(G, LoopbackGraph)
- # LoopbackDiGraph __new__ is not implemented
- G = nx.DiGraph()
- assert not isinstance(G, LoopbackGraph)
- G = nx.Graph()
- assert not isinstance(G, LoopbackGraph)
|