[FEATURE]: python unit tests for easygraph functions and classes

easygraph/classes/operation.py

@@ -260,8 +260,8 @@ def set_node_attributes(G, values, name=None):
 def topological_generations(G):
     if not G.is_directed():
         raise AssertionError("Topological sort not defined on undirected graphs.")
-    indegree_map = {v: d for v, d in G.in_degree() if d > 0}
-    zero_indegree = [v for v, d in G.in_degree() if d == 0]
+    indegree_map = {v: d for v, d in G.in_degree().items() if d > 0}
+    zero_indegree = [v for v, d in G.in_degree().items() if d == 0]
     while zero_indegree:
         this_generation = zero_indegree
         zero_indegree = []
@@ -283,7 +283,7 @@ def topological_generations(G):
 
 
 def topological_sort(G):
-    for generation in eg.topological_generations(G):
+    for generation in topological_generations(G):
         yield from generation
 
 

easygraph/classes/tests/test_base.py

@@ -0,0 +1,145 @@
+import sys
+
+import pytest
+
+
+np = pytest.importorskip("numpy")
+pd = pytest.importorskip("pandas")
+sp = pytest.importorskip("scipy")
+
+import easygraph as eg
+
+from easygraph.utils.misc import *
+
+
+class TestConvertNumpyArray:
+    def setup_method(self):
+        self.G1 = eg.complete_graph(5)
+
+    def assert_equal(self, G1, G2):
+        assert nodes_equal(G1.nodes, G2.nodes)
+        assert edges_equal(G1.edges, G2.edges, need_data=False)
+
+    def identity_conversion(self, G, A, create_using):
+        assert A.sum() > 0
+        GG = eg.from_numpy_array(A, create_using=create_using)
+        self.assert_equal(G, GG)
+        GW = eg.to_easygraph_graph(A, create_using=create_using)
+        self.assert_equal(G, GW)
+
+    def test_identity_graph_array(self):
+        A = eg.to_numpy_array(self.G1)
+        self.identity_conversion(self.G1, A, eg.Graph())
+
+
+class TestConvertPandas:
+    def setup_method(self):
+        self.rng = np.random.RandomState(seed=5)
+        ints = self.rng.randint(1, 11, size=(3, 2))
+        a = ["A", "B", "C"]
+        b = ["D", "A", "E"]
+        df = pd.DataFrame(ints, columns=["weight", "cost"])
+        df[0] = a
+        df["b"] = b
+        self.df = df
+
+        mdf = pd.DataFrame([[4, 16, "A", "D"]], columns=["weight", "cost", 0, "b"])
+        self.mdf = pd.concat([df, mdf])
+
+    def assert_equal(self, G1, G2):
+        assert nodes_equal(G1.nodes, G2.nodes)
+        assert edges_equal(G1.edges, G2.edges, need_data=False)
+
+    def test_from_edgelist_multi_attr(self):
+        Gtrue = eg.Graph(
+            [
+                ("E", "C", {"cost": 9, "weight": 10}),
+                ("B", "A", {"cost": 1, "weight": 7}),
+                ("A", "D", {"cost": 7, "weight": 4}),
+            ]
+        )
+        G = eg.from_pandas_edgelist(self.df, 0, "b", ["weight", "cost"])
+        self.assert_equal(G, Gtrue)
+
+    def test_from_adjacency(self):
+        Gtrue = eg.DiGraph([("A", "B"), ("B", "C")])
+        data = {
+            "A": {"A": 0, "B": 0, "C": 0},
+            "B": {"A": 1, "B": 0, "C": 0},
+            "C": {"A": 0, "B": 1, "C": 0},
+        }
+        dftrue = pd.DataFrame(data, dtype=np.intp)
+        df = dftrue[["A", "C", "B"]]
+        G = eg.from_pandas_adjacency(df, create_using=eg.DiGraph())
+        self.assert_equal(G, Gtrue)
+
+
+class TestConvertScipy:
+    def setup_method(self):
+        self.G1 = eg.complete_graph(3)
+
+    def assert_equal(self, G1, G2):
+        assert nodes_equal(G1.nodes, G2.nodes)
+        assert edges_equal(G1.edges, G2.edges, need_data=False)
+
+    @pytest.mark.skipif(
+        sys.version_info < (3, 8), reason="requires python3.8 or higher"
+    )
+    def test_from_scipy(self):
+        data = sp.sparse.csr_matrix([[0, 1, 1], [1, 0, 1], [1, 1, 0]])
+        G = eg.from_scipy_sparse_matrix(data)
+        self.assert_equal(self.G1, G)
+
+
+def test_from_edgelist():
+    edgelist = [(0, 1), (1, 2)]
+    G = eg.from_edgelist(edgelist)
+    assert sorted((u, v) for u, v, _ in G.edges) == [(0, 1), (1, 2)]
+
+
+def test_from_dict_of_lists():
+    d = {0: [1], 1: [2]}
+    G = eg.to_easygraph_graph(d)
+    assert sorted((u, v) for u, v, _ in G.edges) == [(0, 1), (1, 2)]
+
+
+def test_from_dict_of_dicts():
+    d = {0: {1: {}}, 1: {2: {}}}
+    G = eg.to_easygraph_graph(d)
+    assert sorted((u, v) for u, v, _ in G.edges) == [(0, 1), (1, 2)]
+
+
+def test_from_numpy_array():
+    G = eg.complete_graph(3)
+    A = eg.to_numpy_array(G)
+    G2 = eg.from_numpy_array(A)
+    assert sorted((u, v) for u, v, _ in G.edges) == sorted(
+        (u, v) for u, v, _ in G2.edges
+    )
+
+
+def test_from_pandas_edgelist():
+    df = pd.DataFrame({"source": [0, 1], "target": [1, 2], "weight": [0.5, 0.7]})
+    G = eg.from_pandas_edgelist(df, source="source", target="target", edge_attr=True)
+    assert sorted((u, v) for u, v, _ in G.edges) == [(0, 1), (1, 2)]
+
+
+def test_from_pandas_adjacency():
+    df = pd.DataFrame([[0, 1], [1, 0]], columns=["A", "B"], index=["A", "B"])
+    G = eg.from_pandas_adjacency(df)
+    assert sorted((u, v) for u, v, _ in G.edges) == [("A", "B")]
+
+
+def test_from_scipy_sparse_matrix():
+    mat = sp.sparse.csr_matrix([[0, 1, 0], [1, 0, 1], [0, 1, 0]])
+    G = eg.from_scipy_sparse_matrix(mat)
+    expected_edges = [(0, 1), (1, 2)]
+    assert sorted((u, v) for u, v, _ in G.edges) == expected_edges
+
+
+def test_invalid_dict_type():
+    class NotGraph:
+        pass
+
+    with pytest.raises(eg.EasyGraphError):
+        eg.to_easygraph_graph(NotGraph())

easygraph/classes/tests/test_directed_graph.py

@@ -0,0 +1,97 @@
+import os
+import unittest
+
+from easygraph import DiGraph
+
+
+class TestDiGraph(unittest.TestCase):
+    def setUp(self):
+        self.G = DiGraph()
+
+    def test_add_node_and_exists(self):
+        self.G.add_node("A")
+        self.assertTrue(self.G.has_node("A"))
+        self.assertIn("A", self.G.nodes)
+
+    def test_add_nodes_with_attrs(self):
+        self.G.add_nodes(["B", "C"], nodes_attr=[{"age": 30}, {"age": 40}])
+        self.assertEqual(self.G.nodes["B"]["age"], 30)
+        self.assertEqual(self.G.nodes["C"]["age"], 40)
+
+    def test_add_edge_and_attrs(self):
+        self.G.add_edge("A", "B", weight=5)
+        self.assertTrue(self.G.has_edge("A", "B"))
+        self.assertEqual(self.G.adj["A"]["B"]["weight"], 5)
+
+    def test_add_edges_with_attrs(self):
+        self.G.add_edges([("B", "C"), ("C", "D")], edges_attr=[{"w": 1}, {"w": 2}])
+        self.assertEqual(self.G.adj["B"]["C"]["w"], 1)
+        self.assertEqual(self.G.adj["C"]["D"]["w"], 2)
+
+    def test_remove_node_and_edges(self):
+        self.G.add_edges([("X", "Y"), ("Y", "Z")])
+        self.G.remove_node("Y")
+        self.assertFalse("Y" in self.G.nodes)
+        self.assertFalse(self.G.has_edge("Y", "Z"))
+
+    def test_remove_edge(self):
+        self.G.add_edge("M", "N")
+        self.G.remove_edge("M", "N")
+        self.assertFalse(self.G.has_edge("M", "N"))
+
+    def test_degrees(self):
+        self.G.add_edges(
+            [("A", "B"), ("C", "B")], edges_attr=[{"weight": 3}, {"weight": 2}]
+        )
+
+        in_degrees = self.G.in_degree(weight="weight")
+        out_degrees = self.G.out_degree(weight="weight")
+        degrees = self.G.degree(weight="weight")
+
+        self.assertEqual(in_degrees["B"], 5)
+        self.assertEqual(out_degrees["A"], 3)
+        self.assertEqual(degrees["B"], 5)
+
+    def test_neighbors_and_preds(self):
+        self.G.add_edges([("P", "Q"), ("R", "P")])
+        self.assertIn("Q", list(self.G.neighbors("P")))
+        self.assertIn("R", list(self.G.predecessors("P")))
+        all_n = list(self.G.all_neighbors("P"))
+        self.assertIn("Q", all_n)
+        self.assertIn("R", all_n)
+
+    def test_size_and_num_edges_nodes(self):
+        self.G.add_edges([("X", "Y"), ("Y", "Z")])
+        self.assertEqual(self.G.size(), 2)
+        self.assertEqual(self.G.number_of_edges(), 2)
+        self.assertEqual(self.G.number_of_nodes(), 3)
+
+    def test_subgraph_and_ego(self):
+        self.G.add_edges([("A", "B"), ("B", "C"), ("C", "D")])
+        sub = self.G.nodes_subgraph(["A", "B", "C"])
+        self.assertTrue(sub.has_edge("A", "B"))
+        self.assertFalse(sub.has_edge("C", "D"))
+        ego = self.G.ego_subgraph("B")
+        self.assertIn("A", ego.nodes or [])
+        self.assertIn("C", ego.nodes or [])
+
+    def test_to_index_node_graph(self):
+        self.G.add_edges([("foo", "bar"), ("bar", "baz")])
+        G2, node2idx, idx2node = self.G.to_index_node_graph()
+        self.assertEqual(len(G2.nodes), 3)
+        self.assertEqual(node2idx["foo"], 0)
+        self.assertEqual(idx2node[0], "foo")
+
+    def test_copy(self):
+        self.G.add_edge("copyA", "copyB", weight=42)
+        G_copy = self.G.copy()
+        self.assertEqual(G_copy.adj["copyA"]["copyB"]["weight"], 42)
+
+    def test_file_add_edges(self):
+        fname = "temp_edges.txt"
+        with open(fname, "w") as f:
+            f.write("1 2 3.5\n2 3 4.5\n")
+        self.G.add_edges_from_file(fname, weighted=True)
+        os.remove(fname)
+        self.assertEqual(self.G.adj["1"]["2"]["weight"], 3.5)
+        self.assertEqual(self.G.adj["2"]["3"]["weight"], 4.5)

easygraph/classes/tests/test_graphV2.py

@@ -0,0 +1,122 @@
+import unittest
+
+import easygraph as eg
+
+
+class TestEasyGraph(unittest.TestCase):
+    def setUp(self):
+        self.G = eg.Graph()
+
+    def test_add_single_node(self):
+        self.G.add_node(1)
+        self.assertIn(1, self.G.nodes)
+
+    def test_add_multiple_nodes(self):
+        self.G.add_nodes([2, 3, 4])
+        for node in [2, 3, 4]:
+            self.assertIn(node, self.G.nodes)
+
+    def test_add_node_with_attributes(self):
+        self.G.add_node("node", color="red")
+        self.assertEqual(self.G.nodes["node"]["color"], "red")
+
+    def test_add_single_edge(self):
+        self.G.add_edge(1, 2)
+        self.assertTrue(self.G.has_edge(1, 2))
+        self.assertTrue(self.G.has_edge(2, 1))
+
+    def test_add_edge_with_weight(self):
+        self.G.add_edge("a", "b", weight=10)
+        self.assertEqual(self.G["a"]["b"]["weight"], 10)
+
+    def test_add_edges(self):
+        self.G.add_edges([(1, 2), (2, 3)], edges_attr=[{"weight": 5}, {"weight": 6}])
+        self.assertEqual(self.G[1][2]["weight"], 5)
+        self.assertEqual(self.G[2][3]["weight"], 6)
+
+    def test_remove_node(self):
+        self.G.add_node(10)
+        self.G.remove_node(10)
+        self.assertNotIn(10, self.G.nodes)
+
+    def test_remove_edge(self):
+        self.G.add_edge(1, 2)
+        self.G.remove_edge(1, 2)
+        self.assertFalse(self.G.has_edge(1, 2))
+
+    def test_neighbors(self):
+        self.G.add_edges([(1, 2), (1, 3)])
+        neighbors = list(self.G.neighbors(1))
+        self.assertIn(2, neighbors)
+        self.assertIn(3, neighbors)
+
+    def test_subgraph(self):
+        self.G.add_edges([(1, 2), (2, 3), (3, 4)])
+        subG = self.G.nodes_subgraph([2, 3])
+        self.assertIn(2, subG.nodes)
+        self.assertIn(3, subG.nodes)
+        self.assertTrue(subG.has_edge(2, 3))
+        self.assertFalse(subG.has_edge(3, 4))
+
+    def test_ego_subgraph(self):
+        self.G.add_edges([(1, 2), (2, 3), (2, 4)])
+        ego = self.G.ego_subgraph(2)
+        self.assertIn(2, ego.nodes)
+        self.assertIn(1, ego.nodes)
+        self.assertIn(3, ego.nodes)
+        self.assertIn(4, ego.nodes)
+
+    def test_to_index_node_graph(self):
+        self.G.add_edges([("a", "b"), ("b", "c")])
+        G_index, index_of_node, node_of_index = self.G.to_index_node_graph()
+        self.assertEqual(len(G_index.nodes), 3)
+        self.assertTrue(all(isinstance(k, int) for k in G_index.nodes))
+
+    def test_directed_conversion(self):
+        self.G.add_edge(1, 2)
+        H = self.G.to_directed()
+        self.assertTrue(H.is_directed())
+        self.assertTrue(H.has_edge(1, 2))
+        self.assertTrue(H.has_edge(2, 1))
+
+    def test_clone_graph(self):
+        self.G.add_edges([(1, 2), (2, 3)])
+        G_clone = self.G.copy()
+        self.assertTrue(G_clone.has_edge(1, 2))
+        self.assertTrue(G_clone.has_edge(2, 3))
+
+    def test_degree(self):
+        self.G.add_edge(1, 2, weight=5)
+        deg = self.G.degree()
+        self.assertEqual(deg[1], 5)
+        self.assertEqual(deg[2], 5)
+
+    def test_size(self):
+        self.G.add_edges([(1, 2), (2, 3)])
+        self.assertEqual(self.G.size(), 2)
+
+    def test_edge_weight_default(self):
+        self.G.add_edge(4, 5)
+        self.assertEqual(self.G[4][5].get("weight", 1), 1)
+
+    def test_node_index_mappings(self):
+        self.G.add_nodes([10, 20, 30])
+        index2node = self.G.index2node
+        node_index = self.G.node_index
+        for i, node in index2node.items():
+            self.assertEqual(node_index[node], i)
+
+    def test_graph_order(self):
+        self.G.add_nodes([1, 2, 3])
+        self.assertEqual(self.G.order(), 3)
+
+    def test_graph_size_with_weight(self):
+        self.G.add_edges([(1, 2), (2, 3)], edges_attr=[{"weight": 4}, {"weight": 6}])
+        self.assertEqual(self.G.size(weight="weight"), 10.0)
+
+    def test_clear_cache(self):
+        self.G.add_edge(1, 2)
+        _ = self.G.edges
+        self.assertIn("edge", self.G.cache)
+        self.G._clear_cache()
+        self.assertEqual(len(self.G.cache), 0)