Coverage for tests/core/test_graph_generation.py: 100%

1import sys 

2from os.path import dirname as up 

3from builtins import * 

4import pytest 

5 

6import networkx as nx 

7import dantro 

8from dantro.containers import XrDataContainer 

9import numpy as np 

10 

11from dantro._import_tools import import_module_from_path 

12from pkg_resources import resource_filename 

13import h5py as h5 

14from utopya.yaml import load_yml 

15 

16sys.path.insert(0, up(up(up(__file__)))) 

17 

18graph_gen = import_module_from_path( 

19 mod_path=up(up(up(__file__))), mod_str="include.graph" 

20) 

21 

22# Load the test config 

23CFG_FILENAME = resource_filename("tests", "cfgs/graph_generation.yml") 

24test_cfg = load_yml(CFG_FILENAME) 

25 

26 

27def test_graph_generation(): 

28 for _, config in test_cfg.items(): 

29 

30 _raises = config.pop("_raises", False) 

31 _exp_exc = Exception if not isinstance(_raises, str) else globals()[_raises] 

32 _warns = config.pop("_warns", False) 

33 _exp_warning = UserWarning if not isinstance(_warns, str) else globals()[_warns] 

34 _match = config.pop("_match", " ") 

35 

36 nw_cfg = config.pop("network") 

37 

38 if _raises: 

39 with pytest.raises(_exp_exc, match=_match): 

40 graph_gen.generate_graph(**config, **nw_cfg) 

41 continue 

42 

43 # Generate the graph from the configuration 

44 G = graph_gen.generate_graph(**config, **nw_cfg) 

45 

46 # check the graph is non-empty 

47 assert G != nx.empty_graph() 

48 

49 # check the graph has no isolated vertices 

50 assert not list(nx.isolates(G)) 

51 

52 # check the edge weights were set, if specified 

53 if nw_cfg["graph_props"]["is_weighted"]: 

54 assert [0 <= e[-1]["weight"] <= 1 for e in G.edges(data=True)] 

55 else: 

56 assert [not e[-1] for e in G.edges(data=True)] 

57 

58 # Check Barabasi-Albert and BollobasRiordan graphs are undirected and directed respectively 

59 if nw_cfg["type"] == "BarabasiAlbert": 

60 assert not G.is_directed() 

61 elif nw_cfg["type"] == "BollobasRiordan": 

62 assert G.is_directed() 

63 

64def test_graph_saving(tmpdir): 

65 

66 # Create an h5File in the temporary directory for the 

67 h5dir = tmpdir.mkdir("hdf5_data") 

68 h5file = h5.File(h5dir.join(f"test_file.h5"), "w") 

69 nw_group = h5file.create_group("graph") 

70 nw_group.attrs["content"] = "graph" 

71 nw_group.attrs["allows_parallel"] = False 

72 nw_group.attrs["is_directed"] = True 

73 

74 # Test saving a static graph 

75 G1 = graph_gen.generate_graph(N=16, mean_degree=4, type='random', graph_props=dict(is_directed=True, is_weighted=True)) 

76 

77 graph_gen.save_nw(G1, nw_group, write_adjacency_matrix=True) 

78 

79 # Load the network 

80 GG = dantro.groups.GraphGroup( 

81 name="true_network", 

82 attrs=dict( 

83 directed=h5file["graph"].attrs["is_directed"], 

84 parallel=h5file["graph"].attrs["allows_parallel"], 

85 ), 

86 ) 

87 GG.new_container( 

88 "nodes", 

89 Cls=XrDataContainer, 

90 data=np.array(h5file["graph"]["_vertices"]), 

91 ) 

92 GG.new_container("edges", Cls=XrDataContainer, data=[]) 

93 

94 edges = np.array(h5file["graph"]["_edges"]) 

95 edge_weights = np.expand_dims( 

96 np.array(h5file["graph"]["_edge_weights"]), -1 

97 ) 

98 weighted_edges = np.squeeze(np.concatenate((edges, edge_weights), axis=-1)) 

99 

100 A = np.array(h5file["graph"]["_adjacency_matrix"]) 

101 

102 G2 = GG.create_graph() 

103 G2.add_weighted_edges_from(weighted_edges, "weight") 

104 h5file.close() 

105 

106 A1 = nx.adjacency_matrix(G1).toarray() 

107 A2 = nx.adjacency_matrix(G2).toarray() 

108 

109 # Assert the adjacency matrices are all equal 

110 assert (np.all([a == pytest.approx(0, abs=1e-6) for a in A1-A2])) 

111 assert (np.all([a == pytest.approx(0, abs=1e-6) for a in A1-A])) 

112 

113