-
Notifications
You must be signed in to change notification settings - Fork 42
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
tests: add Python resourcegraph tests
Problem: there are no tests for the Python resourcegraph class. Add some basic tests.
- Loading branch information
1 parent
911fe43
commit 24ab265
Showing
3 changed files
with
81 additions
and
1 deletion.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,78 @@ | ||
#!/usr/bin/env python3 | ||
|
||
############################################################### | ||
# Copyright 2023 Lawrence Livermore National Security, LLC | ||
# (c.f. AUTHORS, NOTICE.LLNS, COPYING) | ||
# | ||
# This file is part of the Flux resource manager framework. | ||
# For details, see https://github.com/flux-framework. | ||
# | ||
# SPDX-License-Identifier: LGPL-3.0 | ||
|
||
import unittest | ||
import json | ||
|
||
from fluxion.resourcegraph.V1 import ( | ||
FluxionResourceGraphV1, | ||
FluxionResourcePoolV1, | ||
FluxionResourceRelationshipV1, | ||
) | ||
|
||
from pycotap import TAPTestRunner | ||
|
||
RV1 = { | ||
"version": 1, | ||
"execution": { | ||
"R_lite": [{"rank": "0", "children": {"core": "0-4"}}], | ||
"starttime": 0.0, | ||
"expiration": 0.0, | ||
"nodelist": ["compute01"], | ||
}, | ||
} | ||
|
||
RV1_2 = { | ||
"version": 1, | ||
"execution": { | ||
"R_lite": [{"rank": "0-10", "children": {"gpu": "0-1", "core": "0-7"}}], | ||
"starttime": 0.0, | ||
"expiration": 0.0, | ||
"nodelist": ["compute[0-10]"], | ||
}, | ||
} | ||
|
||
|
||
class TestResourceGraph(unittest.TestCase): | ||
"""Test for the ResourceGraph class.""" | ||
|
||
def _check_metadata(self, metadata): | ||
if metadata["type"] in ("node", "core", "gpu", "cluster"): | ||
self.assertEqual(metadata["unit"], "") | ||
self.assertEqual(metadata["size"], 1) | ||
self.assertEqual(metadata["properties"], []) | ||
else: | ||
raise ValueError(metadata["type"]) | ||
|
||
def test_basic(self): | ||
graph = FluxionResourceGraphV1(RV1) | ||
self.assertTrue(graph.is_directed()) | ||
j = graph.to_JSON() | ||
json.dumps(j) # make sure it doesn't throw an error | ||
self.assertTrue(j["graph"]["directed"]) | ||
self.assertEqual(len(j["graph"]["nodes"]), len(graph.get_nodes())) | ||
self.assertEqual(len(j["graph"]["edges"]), len(graph.get_edges())) | ||
for node in graph.get_nodes(): | ||
self._check_metadata(node.get_metadata()) | ||
|
||
def test_basic_2(self): | ||
graph = FluxionResourceGraphV1(RV1_2) | ||
self.assertTrue(graph.is_directed()) | ||
j = graph.to_JSON() | ||
json.dumps(j) | ||
self.assertTrue(j["graph"]["directed"]) | ||
self.assertEqual(len(j["graph"]["nodes"]), len(graph.get_nodes())) | ||
self.assertEqual(len(j["graph"]["edges"]), len(graph.get_edges())) | ||
for node in graph.get_nodes(): | ||
self._check_metadata(node.get_metadata()) | ||
|
||
|
||
unittest.main(testRunner=TAPTestRunner()) |