add some tests

Author: arashbm

tests/network_strategies.py

@@ -0,0 +1,147 @@
+from hypothesis import strategies as st
+import reticula as ret
+
+# Shared integer vertex type
+VERT = ret.int64
+TIME = ret.double
+
+
+@st.composite
+def undirected_network(draw: st.DrawFn,
+                       min_verts: int = 0, max_verts: int = 10,
+                       self_loops: bool = True):
+    if min_verts < 0:
+        raise ValueError("min_verts cannot be negative")
+    if max_verts < min_verts:
+        raise ValueError("min_verts should not exceed max_verts")
+    verts = list(
+        range(draw(st.integers(min_value=min_verts, max_value=max_verts))))
+    max_edges = len(verts)*(len(verts)-1)//2
+    if self_loops:
+        max_edges += len(verts)
+    edges = st.lists(
+        st.tuples(
+            st.integers(min_value=0, max_value=len(verts)),
+            st.integers(min_value=0, max_value=len(verts))).filter(
+                lambda e: e[0] != e[1] or self_loops),
+        min_size=0, max_size=max_edges)
+    return ret.undirected_network[VERT](draw(edges), verts)
+
+
+@st.composite
+def directed_network(draw: st.DrawFn,
+                     min_verts: int = 0, max_verts: int = 10,
+                     self_loops: bool = True):
+    if min_verts < 0:
+        raise ValueError("min_verts cannot be negative")
+    if max_verts < min_verts:
+        raise ValueError("min_verts should not exceed max_verts")
+    verts = list(
+        range(draw(st.integers(min_value=min_verts, max_value=max_verts))))
+    max_edges = len(verts)*(len(verts)-1)
+    if self_loops:
+        max_edges += len(verts)
+    edges = st.lists(
+        st.tuples(
+            st.integers(min_value=0, max_value=len(verts)),
+            st.integers(min_value=0, max_value=len(verts))).filter(
+                lambda e: e[0] != e[1] or self_loops),
+        min_size=0, max_size=max_edges)
+    return ret.directed_network[VERT](draw(edges), verts)
+
+
+@st.composite
+def undirected_hypernetwork(draw: st.DrawFn,
+                            min_verts: int = 0, max_verts: int = 10):
+    if min_verts < 0:
+        raise ValueError("min_verts cannot be negative")
+    if max_verts < min_verts:
+        raise ValueError("min_verts should not exceed max_verts")
+    verts = list(
+        range(draw(st.integers(min_value=min_verts, max_value=max_verts))))
+    max_edges = len(verts)*2
+    hyperedge = st.lists(st.integers(min_value=0, max_value=len(
+        verts)), min_size=0, max_size=len(verts))
+    edges = st.lists(hyperedge, min_size=0, max_size=max_edges)
+    return ret.undirected_hypernetwork[VERT](draw(edges), verts)
+
+
+@st.composite
+def directed_hypernetwork(draw: st.DrawFn,
+                          min_verts: int = 0, max_verts: int = 10):
+    if min_verts < 0:
+        raise ValueError("min_verts cannot be negative")
+    if max_verts < min_verts:
+        raise ValueError("min_verts should not exceed max_verts")
+    verts = list(
+        range(draw(st.integers(min_value=min_verts, max_value=max_verts))))
+    max_edges = len(verts)*2
+    vertex_list = st.lists(st.integers(
+        min_value=0, max_value=len(verts)), min_size=0, max_size=len(verts))
+    edge = st.tuples(vertex_list, vertex_list)
+    edges = st.lists(edge, min_size=0, max_size=max_edges)
+    return ret.directed_hypernetwork[VERT](draw(edges), verts)
+
+
+@st.composite
+def undirected_temporal_network(draw: st.DrawFn,
+                                min_verts: int = 0, max_verts: int = 10,
+                                self_loops: bool = True):
+    verts = list(
+        range(draw(st.integers(min_value=min_verts, max_value=max_verts))))
+    max_edges = len(verts)*(len(verts)-1)//2
+    if self_loops:
+        max_edges += len(verts)
+    edge = st.tuples(
+        st.integers(min_value=0, max_value=len(verts)),
+        st.integers(min_value=0, max_value=len(verts)),
+        st.floats(min_value=0, max_value=10)).filter(
+            lambda e: e[0] != e[1] or self_loops)
+    edges = st.lists(edge, min_size=0, max_size=max_edges)
+    return ret.undirected_temporal_network[VERT, TIME](draw(edges), verts)
+
+
+@st.composite
+def directed_temporal_network(draw: st.DrawFn,
+                              min_verts: int = 0, max_verts: int = 10,
+                              self_loops: bool = True):
+    verts = list(
+        range(draw(st.integers(min_value=min_verts, max_value=max_verts))))
+    max_edges = len(verts)*(len(verts)-1)
+    if self_loops:
+        max_edges += len(verts)
+    edge = st.tuples(
+        st.integers(min_value=0, max_value=len(verts)),
+        st.integers(min_value=0, max_value=len(verts)),
+        st.floats(min_value=0, max_value=10)).filter(
+            lambda e: e[0] != e[1] or self_loops)
+    edges = st.lists(edge, min_size=0, max_size=max_edges)
+    return ret.directed_temporal_network[VERT, TIME](draw(edges), verts)
+
+
+@st.composite
+def undirected_temporal_hypernetwork(draw: st.DrawFn,
+                                     min_verts: int = 0, max_verts: int = 10):
+    verts = list(
+        range(draw(st.integers(min_value=min_verts, max_value=max_verts))))
+    max_edges = len(verts)*2
+    hyperedge = st.tuples(
+        st.lists(st.integers(min_value=0, max_value=len(verts)),
+                 min_size=0, max_size=len(verts)),
+        st.floats(min_value=0, max_value=10))
+    edges = st.lists(hyperedge, min_size=0, max_size=max_edges)
+    return ret.undirected_temporal_hypernetwork[VERT, TIME](draw(edges), verts)
+
+
+@st.composite
+def directed_temporal_hypernetwork(draw: st.DrawFn,
+                                   min_verts: int = 0, max_verts: int = 10):
+    verts = list(
+        range(draw(st.integers(min_value=min_verts, max_value=max_verts))))
+    max_edges = len(verts)*2
+    vertex_list = st.lists(st.integers(
+        min_value=0, max_value=len(verts)), min_size=0, max_size=len(verts))
+    hyperedge = st.tuples(vertex_list, vertex_list,
+                          st.floats(min_value=0, max_value=10))
+    edges = st.lists(hyperedge, min_size=0, max_size=max_edges)
+    return ret.directed_temporal_hypernetwork[VERT, TIME](draw(edges), verts)

tests/test_algorithms.py

@@ -0,0 +1,181 @@
+import math
+
+import pytest
+import reticula as ret
+from hypothesis import given
+from network_strategies import undirected_network, directed_network
+
+
+def test_density():
+    g = ret.complete_graph[ret.int64](size=4)
+    assert ret.density(g) == 1.0
+
+    g = ret.undirected_network[ret.int64](edges=[], verts=range(4))
+    assert ret.density(g) == 0.0
+
+    g = ret.path_graph[ret.int64](size=4)
+    expected_density = 3 / 6
+    assert ret.density(g) == pytest.approx(expected_density)
+
+
+def test_is_connected():
+    g = ret.path_graph[ret.int64](size=4)
+    assert ret.is_connected(g)
+
+    g = ret.undirected_network[ret.int64]([(0, 1), (2, 3)])
+    assert not ret.is_connected(g)
+
+    g = ret.undirected_network[ret.int64]([], [0])
+    assert ret.is_connected(g)
+
+    g = ret.undirected_network[ret.int64]()
+    assert ret.is_connected(g)
+
+
+def test_is_weakly_connected():
+    g = ret.directed_network[ret.int64]([(0, 1), (1, 2)])
+    assert ret.is_weakly_connected(g)
+
+    g = ret.directed_network[ret.int64]([(0, 1), (2, 3)])
+    assert not ret.is_weakly_connected(g)
+
+
+def test_is_acyclic():
+    g = ret.directed_network[ret.int64]([(0, 1), (0, 2), (1, 3), (2, 3)])
+    assert ret.is_acyclic(g)
+
+    g = ret.directed_network[ret.int64]([(0, 1), (1, 2), (2, 0)])
+    assert not ret.is_acyclic(g)
+
+
+def test_topological_order():
+    g = ret.directed_network[ret.int64]([(0, 1), (0, 2), (1, 3), (2, 3)])
+    order = ret.topological_order(g)
+    assert len(order) == 4
+    assert order.index(0) < order.index(1)
+    assert order.index(0) < order.index(2)
+    assert order.index(1) < order.index(3)
+    assert order.index(2) < order.index(3)
+
+    g = ret.directed_network[ret.int64]([(0, 1), (1, 2), (2, 0)])
+    pytest.raises(ValueError, ret.topological_order, g)
+
+
+def test_shortest_path_lengths():
+    g = ret.path_graph[ret.int64](size=4)
+
+    distances_from = ret.shortest_path_lengths_from(g, 0)
+    assert distances_from[0] == 0
+    assert distances_from[1] == 1
+    assert distances_from[2] == 2
+    assert distances_from[3] == 3
+    assert ret.shortest_path_lengths_from(g, 12) == {12: 0}
+
+    distances_to = ret.shortest_path_lengths_to(g, 3)
+    assert distances_to[0] == 3
+    assert distances_to[1] == 2
+    assert distances_to[2] == 1
+    assert distances_to[3] == 0
+    assert ret.shortest_path_lengths_to(g, 12) == {12: 0}
+
+
+def test_is_reachable():
+    g = ret.path_graph[ret.int64](size=4)
+    assert ret.is_reachable(g, 0, 3)
+    assert ret.is_reachable(g, 3, 0)
+
+    g = ret.undirected_network[ret.int64]([(0, 1), (2, 3)])
+    assert not ret.is_reachable(g, 0, 2)
+
+
+def test_component_functions():
+    g = ret.undirected_network[ret.int64]([(0, 1), (2, 3), (4, 5), (5, 6)])
+
+    comp = ret.connected_component(g, 0)
+    assert set(comp) == {0, 1}
+    assert len(comp) == 2
+
+    comp = ret.largest_connected_component(g)
+    assert set(comp) == {4, 5, 6}
+
+    all_comps = ret.connected_components(g)
+    assert len(all_comps) == 3
+
+
+def test_weakly_connected_components():
+    g = ret.directed_network[ret.int64]([(0, 1), (2, 3)])
+    wccs = ret.weakly_connected_components(g)
+    assert len(wccs) == 2
+
+    wcc_sets = [set(comp) for comp in wccs]
+    assert {0, 1} in wcc_sets
+    assert {2, 3} in wcc_sets
+
+    lwcc = ret.largest_weakly_connected_component(g)
+    assert len(lwcc) == 2
+
+
+def test_in_out_components():
+    g = ret.directed_network[ret.int64]([(0, 1), (1, 2), (3, 1)])
+
+    in_comp = ret.in_component(g, 1)
+    assert 0 in in_comp
+    assert 3 in in_comp
+
+    out_comp = ret.out_component(g, 1)
+    assert 2 in out_comp
+
+    in_sizes = ret.in_component_sizes(g)
+    assert len(in_sizes) == len(g.vertices())
+
+    out_sizes = ret.out_component_sizes(g)
+    assert len(out_sizes) == len(g.vertices())
+
+
+def test_degree_sequence():
+    g = ret.undirected_network[ret.int64]([(0, 1), (1, 2), (2, 3)])
+
+    deg_seq = ret.degree_sequence(g)
+    assert set(deg_seq) == {1, 2, 2, 1}
+
+    g = ret.directed_network[ret.int64]([(0, 1), (0, 2), (1, 2)])
+    in_deg_seq = ret.in_degree_sequence(g)
+    assert set(in_deg_seq) == {0, 1, 2}
+
+    out_deg_seq = ret.out_degree_sequence(g)
+    assert set(out_deg_seq) == {2, 1, 0}
+
+    in_out_deg_seq = ret.in_out_degree_pair_sequence(g)
+    assert set(in_out_deg_seq) == {(0, 2), (1, 1), (2, 0)}
+
+    inc_deg_seq = ret.incident_degree_sequence(g)
+    assert set(inc_deg_seq) == {2, 2, 2}
+
+
+@given(undirected_network())
+def test_degree_properties(net):
+    for v in net.vertices():
+        deg = ret.degree(net, v)
+        assert deg >= 0
+        assert deg == len([e for e in net.edges() if v in e.incident_verts()])
+
+
+@given(directed_network())
+def test_directed_degree_properties(net):
+    for v in net.vertices():
+        in_deg = ret.in_degree(net, v)
+        out_deg = ret.out_degree(net, v)
+        assert in_deg >= 0
+        assert out_deg >= 0
+
+
+def test_attribute_assortativity():
+    g = ret.undirected_network[ret.int64]([(0, 1), (1, 2), (2, 3)])
+    attrs = {0: 1.0, 1: 2.0, 2: 2.0, 3: 1.0}
+
+    r = ret.attribute_assortativity(g, attrs, 0.0)
+    assert isinstance(r, float)
+    assert r == pytest.approx(-0.5)
+
+    r = ret.attribute_assortativity(g, {}, 1.0)
+    assert math.isnan(r)

tests/test_clustering_operations.py

@@ -0,0 +1,57 @@
+import pytest
+
+import reticula as ret
+
+
+def test_temporal_clusters():
+    edges = [(0, 1, 1.0), (1, 2, 1.5), (2, 3, 2.0), (3, 4, 3.0)]
+    net = ret.directed_temporal_network[ret.int64, ret.double](edges)
+    adj = ret.temporal_adjacency.simple[net.edge_type()]()
+
+    e = net.edge_type()(1, 2, 1.5)
+
+    in_clusters = ret.in_clusters(net, adj)
+    assert dict(in_clusters)[e].volume() == 2
+
+    out_clusters = ret.out_clusters(net, adj)
+    assert dict(out_clusters)[e].volume() == 3
+
+
+def test_cluster_functions():
+    edges = [(0, 1, 1.0), (1, 2, 1.5), (2, 3, 2.0), (3, 4, 3.0)]
+    net = ret.directed_temporal_network[ret.int64, ret.double](edges)
+    adj = ret.temporal_adjacency.simple[net.edge_type()]()
+
+    in_cluster = ret.in_cluster(net, adj, 2, 1.5)
+    assert in_cluster.volume() == 1
+
+    out_cluster = ret.out_cluster(net, adj, 1, 1.5)
+    assert out_cluster.volume() == 1
+
+
+def test_cluster_sizes():
+    edges = [(0, 1, 1.0), (1, 2, 1.5), (2, 3, 2.0), (3, 4, 3.0)]
+    net = ret.directed_temporal_network[ret.int64, ret.double](edges)
+    adj = ret.temporal_adjacency.simple[net.edge_type()]()
+
+    e = net.edge_type()(1, 2, 1.5)
+
+    in_sizes = ret.in_cluster_sizes(net, adj)
+    assert dict(in_sizes)[e].volume() == 2
+
+    out_sizes = ret.out_cluster_sizes(net, adj)
+    assert dict(out_sizes)[e].volume() == 3
+
+
+def test_cluster_size_estimates():
+    edges = [(0, 1, 1.0), (1, 2, 1.5), (2, 3, 2.0), (3, 4, 3.0)]
+    net = ret.directed_temporal_network[ret.int64, ret.double](edges)
+    adj = ret.temporal_adjacency.simple[net.edge_type()]()
+
+    e = net.edge_type()(1, 2, 1.5)
+
+    in_sizes = ret.in_cluster_sizes(net, adj)
+    assert dict(in_sizes)[e].volume() == pytest.approx(2, rel=0.1)
+
+    out_sizes = ret.out_cluster_sizes(net, adj)
+    assert dict(out_sizes)[e].volume() == pytest.approx(3, rel=0.1)