|
| 1 | +#!/usr/bin/env python3 |
| 2 | + |
| 3 | +############################################################### |
| 4 | +# Copyright 2023 Lawrence Livermore National Security, LLC |
| 5 | +# (c.f. AUTHORS, NOTICE.LLNS, COPYING) |
| 6 | +# |
| 7 | +# This file is part of the Flux resource manager framework. |
| 8 | +# For details, see https://github.com/flux-framework. |
| 9 | +# |
| 10 | +# SPDX-License-Identifier: LGPL-3.0 |
| 11 | + |
| 12 | +import unittest |
| 13 | +import json |
| 14 | +import sys |
| 15 | +import pathlib |
| 16 | + |
| 17 | +from pycotap import TAPTestRunner |
| 18 | + |
| 19 | +# add fluxion to sys.path |
| 20 | +sys.path.insert(0, str(pathlib.Path(__file__).absolute().parents[2] / "src" / "python")) |
| 21 | + |
| 22 | +from fluxion.resourcegraph.V1 import ( |
| 23 | + FluxionResourceGraphV1, |
| 24 | + FluxionResourcePoolV1, |
| 25 | + FluxionResourceRelationshipV1, |
| 26 | +) |
| 27 | + |
| 28 | +RV1 = { |
| 29 | + "version": 1, |
| 30 | + "execution": { |
| 31 | + "R_lite": [{"rank": "0", "children": {"core": "0-4"}}], |
| 32 | + "starttime": 0.0, |
| 33 | + "expiration": 0.0, |
| 34 | + "nodelist": ["compute01"], |
| 35 | + }, |
| 36 | +} |
| 37 | + |
| 38 | +RV1_2 = { |
| 39 | + "version": 1, |
| 40 | + "execution": { |
| 41 | + "R_lite": [{"rank": "0-10", "children": {"gpu": "0-1", "core": "0-7"}}], |
| 42 | + "starttime": 0.0, |
| 43 | + "expiration": 0.0, |
| 44 | + "nodelist": ["compute[0-10]"], |
| 45 | + }, |
| 46 | +} |
| 47 | + |
| 48 | + |
| 49 | +class TestResourceGraph(unittest.TestCase): |
| 50 | + """Test for the ResourceGraph class.""" |
| 51 | + |
| 52 | + def _check_metadata(self, metadata): |
| 53 | + if metadata["type"] in ("node", "core", "gpu", "cluster"): |
| 54 | + self.assertEqual(metadata["unit"], "") |
| 55 | + self.assertEqual(metadata["size"], 1) |
| 56 | + self.assertEqual(metadata["properties"], []) |
| 57 | + else: |
| 58 | + raise ValueError(metadata["type"]) |
| 59 | + |
| 60 | + def test_basic(self): |
| 61 | + graph = FluxionResourceGraphV1(RV1) |
| 62 | + self.assertTrue(graph.is_directed()) |
| 63 | + j = graph.to_JSON() |
| 64 | + json.dumps(j) # make sure it doesn't throw an error |
| 65 | + self.assertTrue(j["graph"]["directed"]) |
| 66 | + self.assertEqual(len(j["graph"]["nodes"]), len(graph.get_nodes())) |
| 67 | + self.assertEqual(len(j["graph"]["edges"]), len(graph.get_edges())) |
| 68 | + for node in graph.get_nodes(): |
| 69 | + self._check_metadata(node.get_metadata()) |
| 70 | + |
| 71 | + def test_basic_2(self): |
| 72 | + graph = FluxionResourceGraphV1(RV1_2) |
| 73 | + self.assertTrue(graph.is_directed()) |
| 74 | + j = graph.to_JSON() |
| 75 | + json.dumps(j) |
| 76 | + self.assertTrue(j["graph"]["directed"]) |
| 77 | + self.assertEqual(len(j["graph"]["nodes"]), len(graph.get_nodes())) |
| 78 | + self.assertEqual(len(j["graph"]["edges"]), len(graph.get_edges())) |
| 79 | + for node in graph.get_nodes(): |
| 80 | + self._check_metadata(node.get_metadata()) |
| 81 | + |
| 82 | + |
| 83 | +unittest.main(testRunner=TAPTestRunner()) |
0 commit comments