()
| 36 | |
| 37 | @unittest.skipIf(F._default_context_str == "gpu", reason="GPU not implemented") |
| 38 | def test_graph_serialize_with_feature(): |
| 39 | num_graphs = 100 |
| 40 | |
| 41 | t0 = time.time() |
| 42 | |
| 43 | g_list = construct_graph(num_graphs) |
| 44 | |
| 45 | t1 = time.time() |
| 46 | |
| 47 | # create a temporary file and immediately release it so DGL can open it. |
| 48 | f = tempfile.NamedTemporaryFile(delete=False) |
| 49 | path = f.name |
| 50 | f.close() |
| 51 | |
| 52 | dgl.save_graphs(path, g_list) |
| 53 | |
| 54 | t2 = time.time() |
| 55 | idx_list = np.random.permutation(np.arange(num_graphs)).tolist() |
| 56 | loadg_list, _ = dgl.load_graphs(path, idx_list) |
| 57 | |
| 58 | t3 = time.time() |
| 59 | idx = idx_list[0] |
| 60 | load_g = loadg_list[0] |
| 61 | print("Save time: {} s".format(t2 - t1)) |
| 62 | print("Load time: {} s".format(t3 - t2)) |
| 63 | print("Graph Construction time: {} s".format(t1 - t0)) |
| 64 | |
| 65 | assert F.allclose(load_g.nodes(), g_list[idx].nodes()) |
| 66 | |
| 67 | load_edges = load_g.all_edges("uv", "eid") |
| 68 | g_edges = g_list[idx].all_edges("uv", "eid") |
| 69 | assert F.allclose(load_edges[0], g_edges[0]) |
| 70 | assert F.allclose(load_edges[1], g_edges[1]) |
| 71 | assert F.allclose(load_g.edata["e1"], g_list[idx].edata["e1"]) |
| 72 | assert F.allclose(load_g.edata["e2"], g_list[idx].edata["e2"]) |
| 73 | assert F.allclose(load_g.ndata["n1"], g_list[idx].ndata["n1"]) |
| 74 | |
| 75 | os.unlink(path) |
| 76 | |
| 77 | |
| 78 | @unittest.skipIf(F._default_context_str == "gpu", reason="GPU not implemented") |
nothing calls this directly
no test coverage detected