commit eb5dc0afea3bdf10feb5cd219ce78fde6dae33ac Author: Christos Choutouridis Date: Sat Sep 26 17:15:43 2026 +0300 Init commit: First try to load and parse diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..5925ab4 --- /dev/null +++ b/.gitignore @@ -0,0 +1,9 @@ +# IDEs +.idea +.vscode + +# Python specific +__pycache__/ + + + diff --git a/Computer Networks II - Programming task.pdf b/Computer Networks II - Programming task.pdf new file mode 100644 index 0000000..bd3248c Binary files /dev/null and b/Computer Networks II - Programming task.pdf differ diff --git a/LICENSE b/LICENSE new file mode 100644 index 0000000..7d9c8af --- /dev/null +++ b/LICENSE @@ -0,0 +1,21 @@ +MIT License + +Copyright (c) 2026 Christos Choutouridis + +Permission is hereby granted, free of charge, to any person obtaining a copy +of this software and associated documentation files (the "Software"), to deal +in the Software without restriction, including without limitation the rights +to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +copies of the Software, and to permit persons to whom the Software is +furnished to do so, subject to the following conditions: + +The above copyright notice and this permission notice shall be included in all +copies or substantial portions of the Software. + +THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE +SOFTWARE. diff --git a/source/topology_loader.py b/source/topology_loader.py new file mode 100644 index 0000000..456b571 --- /dev/null +++ b/source/topology_loader.py @@ -0,0 +1,117 @@ +"""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") diff --git a/source/topology_parser.py b/source/topology_parser.py new file mode 100644 index 0000000..4d40d74 --- /dev/null +++ b/source/topology_parser.py @@ -0,0 +1,930 @@ +"""Parse the declarative topology JSON representation into typed Python objects. + +This module contains the in-memory semantic model used by the rest of the +pipeline and the parser that converts a raw JSON dictionary into that model. + +The parser is intentionally independent of Mininet. Its job is only to turn +JSON-shaped data into typed Python objects. Cross-reference and semantic +validation belong to the validation stage that follows parsing. + +Author: Christos Choutouridis +""" + +from dataclasses import dataclass, field +from enum import Enum +from typing import Any, Dict, List, Mapping, Optional, Set, Tuple, Union + + +JSONMapping = Mapping[str, Any] + + +class TopologyParseError(ValueError): + """Raised when raw JSON data cannot be represented by the topology model.""" + + +class NodeType(Enum): + """Supported logical node types.""" + + HOST = "host" + SWITCH = "switch" + ROUTER = "router" + + +class AddressingMode(Enum): + """Supported Layer-3 addressing modes for an interface.""" + + NONE = "none" + STATIC = "static" + DHCP = "dhcp" + + +class LinkInterfaceSelector(Enum): + """Special interface selectors that are valid on a link endpoint.""" + + AUTO = "auto" + + +class ServiceInterfaceSelector(Enum): + """Special interface selectors that are valid for a service binding.""" + + AUTO = "auto" + ALL = "all" + + +class ServiceType(Enum): + """Supported service types.""" + + DHCP_SERVER = "dhcp-server" + + +@dataclass +class Interface: + """Describe one logical network interface owned by a node. + + Args: + name: Logical interface tag used by the topology description. + addressing: Layer-3 addressing strategy. Omitting addressing in JSON + is represented as AddressingMode.NONE. + address: Static CIDR address, valid when addressing is STATIC. + gateway: Optional default gateway, valid when addressing is STATIC. + mac: Optional explicit MAC address. Reserved for current/future use. + mtu: Optional MTU value. Reserved for current/future use. + """ + + name: str + addressing: AddressingMode = AddressingMode.NONE + address: Optional[str] = None + gateway: Optional[str] = None + mac: Optional[str] = None + mtu: Optional[int] = None + + +@dataclass +class HostSettings: + """Contain host-specific settings. + + The current grammar does not require host-specific settings yet. Keeping + a dedicated type gives the model a stable extension point. + """ + + +@dataclass +class SwitchSettings: + """Contain switch-specific settings. + + The current grammar does not require switch-specific settings yet. Keeping + a dedicated type gives the model a stable extension point. + """ + + +@dataclass +class RouterSettings: + """Contain router-specific settings. + + Args: + ip_forward: Desired IPv4 forwarding state, or None when unspecified. + rp_filter: Desired reverse-path-filter state, or None when unspecified. + """ + + ip_forward: Optional[bool] = None + rp_filter: Optional[bool] = None + + +NodeSettings = Union[HostSettings, SwitchSettings, RouterSettings] + + +@dataclass +class Node: + """Describe one logical topology node. + + Args: + name: Unique logical node name. + type: Semantic node type. + interfaces: Declared logical interfaces. None means that the node did + not declare an interface set and therefore allows dynamic interface + allocation. An empty dictionary means that an explicit, fixed + interface set was declared and contains zero interfaces. + settings: Type-specific node settings. + """ + + name: str + type: NodeType + interfaces: Optional[Dict[str, Interface]] + settings: NodeSettings + + +@dataclass +class LinkEndpoint: + """Describe one endpoint of a point-to-point logical link. + + Args: + node: Name of the node referenced by the endpoint. + interface: Explicit logical interface tag or the AUTO selector. + """ + + node: str + interface: Union[str, LinkInterfaceSelector] + + +@dataclass +class LinkSettings: + """Contain optional link-specific provisioning settings. + + Args: + bandwidth_mbps: Optional link bandwidth limit in megabits per second. + delay_ms: Optional one-way propagation delay in milliseconds. + loss_percent: Optional packet-loss percentage. + jitter_ms: Optional delay variation in milliseconds. + + Notes: + These fields are reserved as extension points for later Mininet link + provisioning, for example through traffic-controlled link classes. + They are parsed now so the semantic model already has a stable place + for link-level configuration. + """ + + bandwidth_mbps: Optional[float] = None + delay_ms: Optional[float] = None + loss_percent: Optional[float] = None + jitter_ms: Optional[float] = None + + +@dataclass +class Link: + """Describe a link between exactly two endpoints. + + Args: + endpoints: Ordered pair of link endpoints. + name: Optional logical link name used for diagnostics or extensions. + settings: Link-specific provisioning settings. + """ + + endpoints: Tuple[LinkEndpoint, LinkEndpoint] + name: Optional[str] = None + settings: LinkSettings = field(default_factory=LinkSettings) + + +@dataclass +class DHCPRange: + """Describe the address range offered by a DHCP server. + + Args: + start: First address in the DHCP allocation range. + end: Last address in the DHCP allocation range. + """ + + start: str + end: str + + +@dataclass +class DHCPServerSettings: + """Contain DHCP-server-specific settings. + + Args: + address_range: Address range offered to DHCP clients. + gateway: Optional default gateway advertised to DHCP clients. + """ + + address_range: DHCPRange + gateway: Optional[str] = None + + +ServiceSettings = DHCPServerSettings + + +@dataclass +class Service: + """Describe a service attached to a topology node. + + Args: + type: Service implementation type. + node: Name of the node on which the service runs. + interface: Explicit logical interface tag or a supported special + selector such as AUTO or ALL. + settings: Type-specific service settings. + """ + + type: ServiceType + node: str + interface: Union[str, ServiceInterfaceSelector] + settings: ServiceSettings + + +@dataclass +class Topology: + """Contain the complete parsed topology model. + + Args: + nodes: Nodes in declaration order. Keeping nodes as a list preserves + the parsed source faithfully, including duplicate names, so that + the separate validation stage can diagnose them instead of losing + information during parsing. + links: Links in declaration order. Order is preserved because it can + make automatic interface allocation deterministic. + services: Services in declaration order. Order may later be useful for + deterministic startup and shutdown. + """ + + nodes: List[Node] = field(default_factory=list) + links: List[Link] = field(default_factory=list) + services: List[Service] = field(default_factory=list) + + +def _mapping(value: Any, context: str) -> JSONMapping: + """Return a value as a mapping or raise a parse error. + + Args: + value: Value expected to be a JSON object. + context: Human-readable location used in the error message. + + Returns: + The value as a mapping. + + Raises: + TopologyParseError: If value is not a mapping. + """ + + if not isinstance(value, Mapping): + raise TopologyParseError("%s must be a JSON object" % context) + return value + + +def _reject_unknown_fields( + data: JSONMapping, + allowed: Set[str], + context: str, +) -> None: + """Reject object members that are not part of the grammar. + + Args: + data: Parsed JSON object whose member names are checked. + allowed: Set of field names accepted at this grammar location. + context: Human-readable location used in the error message. + + Raises: + TopologyParseError: If one or more unknown fields are present. + """ + + unknown = sorted(set(data.keys()) - allowed) + if unknown: + raise TopologyParseError( + "%s contains unknown field%s: %s" + % ( + context, + "s" if len(unknown) != 1 else "", + ", ".join(repr(name) for name in unknown), + ) + ) + + +def _list(value: Any, context: str) -> List[Any]: + """Return a value as a list or raise a parse error. + + Args: + value: Value expected to be a JSON array. + context: Human-readable location used in the error message. + + Returns: + The value as a list. + + Raises: + TopologyParseError: If value is not a list. + """ + + if not isinstance(value, list): + raise TopologyParseError("%s must be a JSON array" % context) + return value + + +def _string(value: Any, context: str) -> str: + """Return a value as a string or raise a parse error. + + Args: + value: Value expected to be a string. + context: Human-readable location used in the error message. + + Returns: + The string value. + + Raises: + TopologyParseError: If value is not a string. + """ + + if not isinstance(value, str): + raise TopologyParseError("%s must be a string" % context) + return value + + +def _optional_string(value: Any, context: str) -> Optional[str]: + """Return an optional string or raise a parse error. + + Args: + value: None or a value expected to be a string. + context: Human-readable location used in the error message. + + Returns: + None or the string value. + + Raises: + TopologyParseError: If value is neither None nor a string. + """ + + if value is None: + return None + return _string(value, context) + + +def _optional_bool(value: Any, context: str) -> Optional[bool]: + """Return an optional boolean or raise a parse error. + + Args: + value: None or a value expected to be a boolean. + context: Human-readable location used in the error message. + + Returns: + None or the boolean value. + + Raises: + TopologyParseError: If value is neither None nor a boolean. + """ + + if value is None: + return None + if not isinstance(value, bool): + raise TopologyParseError("%s must be a boolean" % context) + return value + + +def _optional_int(value: Any, context: str) -> Optional[int]: + """Return an optional integer or raise a parse error. + + Args: + value: None or a value expected to be an integer. + context: Human-readable location used in the error message. + + Returns: + None or the integer value. + + Raises: + TopologyParseError: If value is neither None nor an integer. + """ + + if value is None: + return None + if isinstance(value, bool) or not isinstance(value, int): + raise TopologyParseError("%s must be an integer" % context) + return value + + +def _optional_float(value: Any, context: str) -> Optional[float]: + """Return an optional numeric value as float or raise a parse error. + + Args: + value: None or a value expected to be an integer or floating-point + number. + context: Human-readable location used in the error message. + + Returns: + None or the numeric value converted to float. + + Raises: + TopologyParseError: If value is neither None nor a numeric value. + """ + + if value is None: + return None + if isinstance(value, bool) or not isinstance(value, (int, float)): + raise TopologyParseError("%s must be a number" % context) + return float(value) + + +def _parse_interface(name: str, raw: Any, context: str) -> Interface: + """Parse one interface definition. + + Args: + name: Logical interface tag taken from the JSON object key. + raw: Raw JSON value containing interface properties. + context: Human-readable location used in error messages. + + Returns: + Parsed Interface object. + + Raises: + TopologyParseError: If the interface cannot be represented. + """ + + data = _mapping(raw, context) + _reject_unknown_fields( + data, + {"addressing", "address", "gateway", "mac", "mtu"}, + context, + ) + + raw_addressing = data.get("addressing", AddressingMode.NONE.value) + try: + addressing = AddressingMode(_string(raw_addressing, context + ".addressing")) + except ValueError as exc: + raise TopologyParseError( + "%s.addressing has unsupported value %r" % (context, raw_addressing) + ) from exc + + return Interface( + name=name, + addressing=addressing, + address=_optional_string(data.get("address"), context + ".address"), + gateway=_optional_string(data.get("gateway"), context + ".gateway"), + mac=_optional_string(data.get("mac"), context + ".mac"), + mtu=_optional_int(data.get("mtu"), context + ".mtu"), + ) + + +def _parse_interfaces(raw_node: JSONMapping, context: str) -> Optional[Dict[str, Interface]]: + """Parse the optional interface collection of a node. + + Args: + raw_node: Raw JSON node object. + context: Human-readable node location used in error messages. + + Returns: + None when the interfaces field is omitted, otherwise a dictionary of + logical interface tags to Interface objects. + + Raises: + TopologyParseError: If the interfaces field has an invalid shape. + """ + + if "interfaces" not in raw_node: + return None + + raw_interfaces = _mapping(raw_node["interfaces"], context + ".interfaces") + interfaces: Dict[str, Interface] = {} + + for name, raw_interface in raw_interfaces.items(): + interface_name = _string(name, context + ".interfaces key") + interfaces[interface_name] = _parse_interface( + interface_name, + raw_interface, + "%s.interfaces.%s" % (context, interface_name), + ) + + return interfaces + + +def _parse_node_settings(node_type: NodeType, raw: Any, context: str) -> NodeSettings: + """Parse type-specific node settings. + + Args: + node_type: Parsed semantic node type. + raw: Raw JSON settings object or None. + context: Human-readable node location used in error messages. + + Returns: + Type-specific node settings object. + + Raises: + TopologyParseError: If the settings object has an invalid shape or + contains values that cannot be represented by the selected type. + """ + + data = {} if raw is None else dict(_mapping(raw, context + ".settings")) + + allowed_fields = { + NodeType.HOST: set(), + NodeType.SWITCH: set(), + NodeType.ROUTER: {"ip_forward", "rp_filter"}, + }[node_type] + _reject_unknown_fields(data, allowed_fields, context + ".settings") + + if node_type is NodeType.ROUTER: + return RouterSettings( + ip_forward=_optional_bool( + data.get("ip_forward"), + context + ".settings.ip_forward", + ), + rp_filter=_optional_bool( + data.get("rp_filter"), + context + ".settings.rp_filter", + ), + ) + + if node_type is NodeType.SWITCH: + return SwitchSettings() + + return HostSettings() + + +def _parse_node(raw: Any, index: int) -> Node: + """Parse one node declaration. + + Args: + raw: Raw JSON node object. + index: Node index in the source array, used in diagnostics. + + Returns: + Parsed Node object. + + Raises: + TopologyParseError: If the node cannot be represented. + """ + + context = "nodes[%d]" % index + data = _mapping(raw, context) + _reject_unknown_fields( + data, + {"name", "type", "interfaces", "settings"}, + context, + ) + + name = _string(data.get("name"), context + ".name") + raw_type = _string(data.get("type"), context + ".type") + + try: + node_type = NodeType(raw_type) + except ValueError as exc: + raise TopologyParseError( + "%s.type has unsupported value %r" % (context, raw_type) + ) from exc + + return Node( + name=name, + type=node_type, + interfaces=_parse_interfaces(data, context), + settings=_parse_node_settings(node_type, data.get("settings"), context), + ) + + +def _parse_link_interface(value: Any, context: str) -> Union[str, LinkInterfaceSelector]: + """Parse a link endpoint interface reference. + + Args: + value: Raw interface reference. + context: Human-readable location used in error messages. + + Returns: + Explicit interface tag or LinkInterfaceSelector.AUTO. + + Raises: + TopologyParseError: If the reference is not a string. + """ + + text = _string(value, context) + if text == LinkInterfaceSelector.AUTO.value: + return LinkInterfaceSelector.AUTO + return text + + +def _parse_link_endpoint(raw: Any, context: str) -> LinkEndpoint: + """Parse one link endpoint. + + Args: + raw: Raw JSON endpoint object. + context: Human-readable location used in error messages. + + Returns: + Parsed LinkEndpoint object. + + Raises: + TopologyParseError: If the endpoint cannot be represented. + """ + + data = _mapping(raw, context) + _reject_unknown_fields(data, {"node", "interface"}, context) + + return LinkEndpoint( + node=_string(data.get("node"), context + ".node"), + interface=_parse_link_interface( + data.get("interface"), + context + ".interface", + ), + ) + + +def _parse_link_settings(raw: Any, context: str) -> LinkSettings: + """Parse optional link-specific provisioning settings. + + Args: + raw: Raw JSON settings object or None. + context: Human-readable link location used in error messages. + + Returns: + Parsed LinkSettings object. + + Raises: + TopologyParseError: If a known setting has an incompatible JSON type. + """ + + data = {} if raw is None else dict(_mapping(raw, context)) + _reject_unknown_fields( + data, + {"bandwidth_mbps", "delay_ms", "loss_percent", "jitter_ms"}, + context, + ) + + return LinkSettings( + bandwidth_mbps=_optional_float( + data.get("bandwidth_mbps"), + context + ".bandwidth_mbps", + ), + delay_ms=_optional_float( + data.get("delay_ms"), + context + ".delay_ms", + ), + loss_percent=_optional_float( + data.get("loss_percent"), + context + ".loss_percent", + ), + jitter_ms=_optional_float( + data.get("jitter_ms"), + context + ".jitter_ms", + ), + ) + + +def _parse_link(raw: Any, index: int) -> Link: + """Parse one link declaration. + + Args: + raw: Raw JSON link object. + index: Link index in the source array, used in diagnostics. + + Returns: + Parsed Link object. + + Raises: + TopologyParseError: If the link cannot be represented. + """ + + context = "links[%d]" % index + data = _mapping(raw, context) + _reject_unknown_fields( + data, + {"endpoints", "name", "settings"}, + context, + ) + + raw_endpoints = _list(data.get("endpoints"), context + ".endpoints") + + if len(raw_endpoints) != 2: + raise TopologyParseError( + "%s.endpoints must contain exactly two endpoints" % context + ) + + endpoints = ( + _parse_link_endpoint(raw_endpoints[0], context + ".endpoints[0]"), + _parse_link_endpoint(raw_endpoints[1], context + ".endpoints[1]"), + ) + + return Link( + endpoints=endpoints, + name=_optional_string(data.get("name"), context + ".name"), + settings=_parse_link_settings( + data.get("settings"), + context + ".settings", + ), + ) + + +def _parse_service_interface( + value: Any, + context: str, +) -> Union[str, ServiceInterfaceSelector]: + """Parse a service interface binding. + + Args: + value: Raw service interface reference. + context: Human-readable location used in error messages. + + Returns: + Explicit interface tag, AUTO, or ALL. + + Raises: + TopologyParseError: If the reference is not a string. + """ + + text = _string(value, context) + + for selector in ServiceInterfaceSelector: + if text == selector.value: + return selector + + return text + + +def _parse_dhcp_server_settings(raw: Any, context: str) -> DHCPServerSettings: + """Parse DHCP-server-specific settings. + + Args: + raw: Raw JSON DHCP server settings object. + context: Human-readable location used in error messages. + + Returns: + Parsed DHCPServerSettings object. + + Raises: + TopologyParseError: If required structural fields are missing or have + incompatible JSON types. + """ + + data = _mapping(raw, context) + _reject_unknown_fields(data, {"range", "gateway"}, context) + + raw_range = _mapping(data.get("range"), context + ".range") + _reject_unknown_fields(raw_range, {"start", "end"}, context + ".range") + + address_range = DHCPRange( + start=_string(raw_range.get("start"), context + ".range.start"), + end=_string(raw_range.get("end"), context + ".range.end"), + ) + + return DHCPServerSettings( + address_range=address_range, + gateway=_optional_string(data.get("gateway"), context + ".gateway"), + ) + + +def _parse_service(raw: Any, index: int) -> Service: + """Parse one service declaration. + + Args: + raw: Raw JSON service object. + index: Service index in the source array, used in diagnostics. + + Returns: + Parsed Service object. + + Raises: + TopologyParseError: If the service cannot be represented. + """ + + context = "services[%d]" % index + data = _mapping(raw, context) + _reject_unknown_fields( + data, + {"type", "node", "interface", "settings"}, + context, + ) + + raw_type = _string(data.get("type"), context + ".type") + try: + service_type = ServiceType(raw_type) + except ValueError as exc: + raise TopologyParseError( + "%s.type has unsupported value %r" % (context, raw_type) + ) from exc + + if service_type is ServiceType.DHCP_SERVER: + settings = _parse_dhcp_server_settings( + data.get("settings"), + context + ".settings", + ) + else: + raise TopologyParseError( + "%s.type has no parser implementation" % context + ) + + return Service( + type=service_type, + node=_string(data.get("node"), context + ".node"), + interface=_parse_service_interface( + data.get("interface"), + context + ".interface", + ), + settings=settings, + ) + + +def parse_topology(raw: Mapping[str, Any]) -> Topology: + """Parse a raw JSON dictionary into the typed topology model. + + Args: + raw: Mapping produced by the loading stage. + + Returns: + Topology object containing ordered nodes, links, and services. + + Raises: + TopologyParseError: If the raw structure cannot be represented by the + topology grammar. + + Notes: + This function intentionally does not validate cross references such as + whether a link references an existing node. Those checks belong to + the separate validation stage. + """ + + data = _mapping(raw, "topology") + _reject_unknown_fields(data, {"nodes", "links", "services"}, "topology") + + raw_nodes = _list(data.get("nodes", []), "nodes") + raw_links = _list(data.get("links", []), "links") + raw_services = _list(data.get("services", []), "services") + + nodes = [ + _parse_node(raw_node, index) + for index, raw_node in enumerate(raw_nodes) + ] + + links = [ + _parse_link(raw_link, index) + for index, raw_link in enumerate(raw_links) + ] + + services = [ + _parse_service(raw_service, index) + for index, raw_service in enumerate(raw_services) + ] + + return Topology(nodes=nodes, links=links, services=services) + + +def _self_test() -> None: + """Run a small smoke test against the public parse_topology() interface.""" + + raw = { + "nodes": [ + { + "name": "h1", + "type": "host", + "interfaces": { + "net0": { + "addressing": "dhcp", + } + }, + }, + { + "name": "s1", + "type": "switch", + }, + ], + "links": [ + { + "endpoints": [ + {"node": "h1", "interface": "net0"}, + {"node": "s1", "interface": "auto"}, + ], + "settings": { + "bandwidth_mbps": 100, + "delay_ms": 1.5, + }, + } + ], + "services": [], + } + + topology = parse_topology(raw) + + h1 = topology.nodes[0] + s1 = topology.nodes[1] + + assert h1.type is NodeType.HOST + assert h1.interfaces is not None + assert h1.interfaces["net0"].addressing is AddressingMode.DHCP + assert s1.interfaces is None + assert topology.links[0].endpoints[1].interface is LinkInterfaceSelector.AUTO + assert topology.links[0].settings.bandwidth_mbps == 100.0 + assert topology.links[0].settings.delay_ms == 1.5 + + invalid = { + "nodes": [ + { + "name": "r0", + "type": "router", + "settings": { + "ip_foward": True, + }, + } + ] + } + + try: + parse_topology(invalid) + except TopologyParseError as exc: + assert "ip_foward" in str(exc) + else: + raise AssertionError("unknown fields must raise TopologyParseError") + + +if __name__ == "__main__": + _self_test() + print("topology_parser: self-test passed")