118 lines
2.9 KiB
Python
118 lines
2.9 KiB
Python
"""Load topology JSON files and expose the high-level loading API.
|
|
|
|
The loading stage has two public entry points:
|
|
|
|
load_topology(path)
|
|
Read JSON from disk and return the raw dictionary.
|
|
|
|
load(path)
|
|
Read JSON from disk and immediately parse it into the typed Topology
|
|
model by calling topology_parser.parse_topology().
|
|
|
|
No Mininet code is imported or executed in this module.
|
|
"""
|
|
|
|
import json
|
|
from pathlib import Path
|
|
from tempfile import NamedTemporaryFile
|
|
from typing import Any, Dict, Union
|
|
|
|
from topology_parser import Topology, parse_topology
|
|
|
|
|
|
PathLike = Union[str, Path]
|
|
|
|
|
|
class TopologyLoadError(ValueError):
|
|
"""Raised when a topology file cannot be loaded as a JSON object."""
|
|
|
|
|
|
def load_topology(path: PathLike) -> Dict[str, Any]:
|
|
"""Load a topology JSON file into a raw dictionary.
|
|
|
|
Args:
|
|
path: Path to the topology JSON file.
|
|
|
|
Returns:
|
|
Raw JSON dictionary exactly as loaded from disk.
|
|
|
|
Raises:
|
|
TopologyLoadError: If the file cannot be opened, contains invalid JSON,
|
|
or has a non-object JSON value at its root.
|
|
"""
|
|
|
|
source = Path(path)
|
|
|
|
try:
|
|
with source.open("r", encoding="utf-8") as handle:
|
|
raw = json.load(handle)
|
|
except OSError as exc:
|
|
raise TopologyLoadError(
|
|
"failed to read topology file %s: %s" % (source, exc)
|
|
) from exc
|
|
except json.JSONDecodeError as exc:
|
|
raise TopologyLoadError(
|
|
"invalid JSON in topology file %s at line %d column %d: %s"
|
|
% (source, exc.lineno, exc.colno, exc.msg)
|
|
) from exc
|
|
|
|
if not isinstance(raw, dict):
|
|
raise TopologyLoadError(
|
|
"topology root in %s must be a JSON object" % source
|
|
)
|
|
|
|
return raw
|
|
|
|
|
|
def load(path: PathLike) -> Topology:
|
|
"""Load and parse a topology file in one convenience call.
|
|
|
|
Args:
|
|
path: Path to the topology JSON file.
|
|
|
|
Returns:
|
|
Parsed Topology model.
|
|
|
|
Raises:
|
|
TopologyLoadError: If the JSON file cannot be loaded.
|
|
topology_parser.TopologyParseError: If the loaded JSON cannot be
|
|
represented by the topology grammar.
|
|
"""
|
|
|
|
return parse_topology(load_topology(path))
|
|
|
|
|
|
def _self_test() -> None:
|
|
"""Run a small smoke test against load_topology() and load()."""
|
|
|
|
sample = {
|
|
"nodes": [
|
|
{
|
|
"name": "s1",
|
|
"type": "switch",
|
|
}
|
|
],
|
|
"links": [],
|
|
"services": [],
|
|
}
|
|
|
|
with NamedTemporaryFile(
|
|
mode="w",
|
|
suffix=".json",
|
|
encoding="utf-8",
|
|
delete=True,
|
|
) as handle:
|
|
json.dump(sample, handle)
|
|
handle.flush()
|
|
|
|
raw = load_topology(handle.name)
|
|
topology = load(handle.name)
|
|
|
|
assert raw["nodes"][0]["name"] == "s1"
|
|
assert topology.nodes[0].name == "s1"
|
|
|
|
|
|
if __name__ == "__main__":
|
|
_self_test()
|
|
print("topology_loader: self-test passed")
|