"""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. Author: Christos Choutouridis """ 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")