"""Validate the semantic consistency of a parsed topology. This module implements the validation stage of the topology pipeline. The parser answers: "Can this JSON be represented by the topology grammar?" The validator answers: "Does the parsed topology describe a coherent network?" Validation is intentionally independent of Mininet. It does not create interfaces, execute commands, print diagnostics, or modify network state. Instead, it collects semantic issues and raises a TopologyValidationError when validation fails. Author: Christos Choutouridis """ from dataclasses import dataclass import ipaddress import math import re from typing import Dict, List, Optional, Sequence, Set, Tuple from topology_parser import ( AddressingMode, DHCPServerSettings, HostSettings, Interface, Link, LinkInterfaceSelector, LinkSettings, Node, NodeType, RouterSettings, Service, ServiceInterfaceSelector, ServiceType, SwitchSettings, Topology, ) # ========================== Validation constants ========================== _RESERVED_INTERFACE_TAGS = { LinkInterfaceSelector.AUTO.value, ServiceInterfaceSelector.ALL.value, } _MAC_ADDRESS_RE = re.compile( r"^(?:[0-9A-Fa-f]{2}:){5}[0-9A-Fa-f]{2}$" ) # ====================== Diagnostics and lookup state ====================== @dataclass(frozen=True) class ValidationIssue: """Describe one semantic topology validation failure. Args: location: Logical location of the invalid declaration. message: Human-readable description of the semantic failure. """ location: str message: str class TopologyValidationError(ValueError): """Report one or more semantic errors discovered during validation. Args: issues: Sequence of semantic validation failures. Attributes: issues: Immutable tuple containing all collected validation issues. """ def __init__(self, issues: Sequence[ValidationIssue]) -> None: """Initialize the aggregated validation error. Args: issues: Sequence of semantic validation failures. """ self.issues: Tuple[ValidationIssue, ...] = tuple(issues) super().__init__(self._format_message()) def _format_message(self) -> str: """Format all validation issues into one exception message. Returns: Multi-line human-readable validation summary. """ count = len(self.issues) header = "%d topology validation error%s" % ( count, "" if count == 1 else "s", # ;) ) lines = [header] for issue in self.issues: lines.append("- %s: %s" % (issue.location, issue.message)) return "\n".join(lines) @dataclass class _NodeIndex: """Hold node lookup information prepared for validation. _build_node_index() retains the first declaration for each nonempty name and records repeated names separately while reporting duplicate issues. Link and service checks consult duplicate_names before resolving a known node, avoiding secondary diagnostics based on an ambiguous declaration. Args: nodes: First declaration for each distinct nonempty node name. duplicate_names: Node names that were declared more than once. Example: Suppose the source list contains three Node objects: topology.nodes = [first_h1, s1, second_h1] first_h1.name = "h1" s1.name = "s1" second_h1.name = "h1" The resulting lookup has this shape: node_index = _NodeIndex( nodes={"h1": first_h1, "s1": s1}, duplicate_names={"h1"}, ) Dictionary keys are logical node names, and values reference the original Node objects, not copies or positions in the source list. node_index.nodes["s1"] therefore gives the s1 object directly. The duplicate set contains names only; it is empty for unique names. Membership in nodes means a declaration exists, not that it is unique. Although nodes["h1"] retains first_h1, duplicate_names marks that reference as ambiguous so link and service checks skip resolving it. The original topology.nodes list still contains both h1 declarations. Empty names are reported as issues and excluded from both collections. """ nodes: Dict[str, Node] duplicate_names: Set[str] # =========================== Validation helpers =========================== def _add_issue( issues: List[ValidationIssue], location: str, message: str, ) -> None: """Append one validation issue. Args: issues: Mutable list receiving validation failures. location: Logical location associated with the failure. message: Human-readable description of the failure. """ issues.append(ValidationIssue(location=location, message=message)) def _build_node_index( topology: Topology, issues: List[ValidationIssue], ) -> _NodeIndex: """Build a node lookup table while reporting duplicate names. Args: topology: Parsed topology being validated. issues: Mutable list receiving validation failures. Returns: Node lookup information for later validation passes. Notes: The returned _NodeIndex holds name-to-object lookup and ambiguity information; validation issues are appended to the supplied list. first_index is a temporary name-to-source-position map used to point duplicate diagnostics back to the first declaration. It is not part of the returned index, whose shape is illustrated in the _NodeIndex class docstring. """ nodes: Dict[str, Node] = {} first_index: Dict[str, int] = {} duplicate_names: Set[str] = set() for index, node in enumerate(topology.nodes): location = "nodes[%d].name" % index if not node.name: _add_issue(issues, location, "node name must not be empty") continue if node.name in nodes: duplicate_names.add(node.name) _add_issue( issues, location, "duplicate node name %r; first declared at nodes[%d]" % (node.name, first_index[node.name]), ) continue nodes[node.name] = node first_index[node.name] = index return _NodeIndex( nodes=nodes, duplicate_names=duplicate_names, ) # ============================ Validation core ============================= def _validate_node_settings( node: Node, location: str, issues: List[ValidationIssue], ) -> None: """Validate that a node carries settings matching its semantic type. Args: node: Node whose settings object is checked. location: Logical location of the node (where node data came from). issues: Mutable list receiving validation failures. """ expected_type = { NodeType.HOST: HostSettings, NodeType.SWITCH: SwitchSettings, NodeType.ROUTER: RouterSettings, }[node.type] if not isinstance(node.settings, expected_type): _add_issue( issues, location + ".settings", "settings object does not match node type %r" % node.type.value, ) def _validate_ipv4_interface( value: str, location: str, issues: List[ValidationIssue], ) -> Optional[ipaddress.IPv4Interface]: """Validate one IPv4 CIDR interface address. Require an explicit prefix and convert the string to an IPv4Interface for subsequent subnet checks. Invalid syntax or an unsupported address family records a validation issue at the supplied location instead of raising a parsing exception. Args: value: Address string expected to contain an IPv4 address and prefix. location: Logical location associated with the address(where address string--value-- came from). issues: Mutable list receiving validation failures. Returns: IPv4Interface on success, otherwise None after recording an issue. Example: Input: value = "192.168.1.10/24" location = "nodes[0].interfaces.lan.address" issues = [] Output (result names the return value): result = ipaddress.IPv4Interface("192.168.1.10/24") issues = [] With value = "192.168.1.10" and a fresh empty issues list: result = None issues = [ ValidationIssue( location="nodes[0].interfaces.lan.address", message="static address must use CIDR notation", ), ] Failure appends to the supplied list rather than raising an exception. """ if "/" not in value: _add_issue( issues, location, "static address must use CIDR notation", ) return None try: parsed = ipaddress.ip_interface(value) except ValueError: _add_issue(issues, location, "invalid IP interface address %r" % value) return None if not isinstance(parsed, ipaddress.IPv4Interface): _add_issue(issues, location, "only IPv4 addresses are supported") return None return parsed def _validate_ipv4_address( value: str, location: str, issues: List[ValidationIssue], ) -> Optional[ipaddress.IPv4Address]: """Validate one IPv4 address without a prefix. Convert the string to an IPv4Address for subsequent semantic checks. Invalid syntax or an unsupported address family records a validation issue at the supplied location instead of raising a parsing exception. Args: value: Address string expected to contain one IPv4 address. location: Logical location associated with the address (where value string came from). issues: Mutable list receiving validation failures. Returns: IPv4Address on success, otherwise None after recording an issue. Example: Input: value = "192.168.1.1" location = "nodes[0].interfaces.lan.gateway" issues = [] Output (result names the return value): result = ipaddress.IPv4Address("192.168.1.1") issues = [] With value = "::1" and a fresh empty issues list: result = None issues = [ ValidationIssue( location="nodes[0].interfaces.lan.gateway", message="only IPv4 addresses are supported", ), ] A prefixed value such as "192.168.1.1/24" also returns None and appends an invalid-IPv4-address issue, since this helper expects an address without a prefix. """ try: parsed = ipaddress.ip_address(value) except ValueError: _add_issue(issues, location, "invalid IPv4 address %r" % value) return None if not isinstance(parsed, ipaddress.IPv4Address): _add_issue(issues, location, "only IPv4 addresses are supported") return None return parsed def _validate_interface( node: Node, interface: Interface, location: str, issues: List[ValidationIssue], ) -> None: """Validate one declared logical interface. Args: node: Node that owns the interface. interface: Interface being validated. location: Logical location of the interface (where the interface data came from). issues: Mutable list receiving validation failures. """ if not interface.name: _add_issue(issues, location, "interface tag must not be empty") if interface.name in _RESERVED_INTERFACE_TAGS: _add_issue( issues, location, "interface tag %r is reserved by the grammar" % interface.name, ) if node.type is NodeType.SWITCH and interface.addressing is not AddressingMode.NONE: _add_issue( issues, location + ".addressing", "Layer-2 switch interfaces must not request L3 addressing", ) parsed_address: Optional[ipaddress.IPv4Interface] = None if interface.addressing is AddressingMode.STATIC: if interface.address is None: _add_issue( issues, location + ".address", "static addressing requires an address", ) else: parsed_address = _validate_ipv4_interface( interface.address, location + ".address", issues, ) # For subnets larger than two(2) addresses, reject assigning the network or broadcast # address to an interface. if parsed_address is not None: network = parsed_address.network if ( network.num_addresses > 2 # /30 and up and parsed_address.ip in {network.network_address, network.broadcast_address} ): _add_issue( issues, location + ".address", "static address must not be the network or broadcast address", ) if interface.gateway is not None: parsed_gateway = _validate_ipv4_address( interface.gateway, location + ".gateway", issues, ) if parsed_address is not None and parsed_gateway is not None: if parsed_gateway not in parsed_address.network: _add_issue( issues, location + ".gateway", "gateway must belong to the interface subnet", ) elif parsed_gateway == parsed_address.ip: _add_issue( issues, location + ".gateway", "gateway must not be the interface's own address", ) elif ( # Same here, /30 and up parsed_address.network.num_addresses > 2 and parsed_gateway in { parsed_address.network.network_address, parsed_address.network.broadcast_address, } ): _add_issue( issues, location + ".gateway", "gateway must not be the network or broadcast address", ) elif interface.addressing in {AddressingMode.DHCP, AddressingMode.NONE}: if interface.address is not None: _add_issue( issues, location + ".address", "address is only valid with static addressing", ) if interface.gateway is not None: _add_issue( issues, location + ".gateway", "gateway is only valid with static addressing", ) if interface.mac is not None and _MAC_ADDRESS_RE.fullmatch(interface.mac) is None: _add_issue( issues, location + ".mac", "MAC address must contain six hexadecimal octets separated by ':'", ) if interface.mtu is not None and interface.mtu <= 0: _add_issue( issues, location + ".mtu", "MTU must be greater than zero", ) def _validate_nodes( topology: Topology, issues: List[ValidationIssue], ) -> _NodeIndex: """Validate node declarations and their local interface configuration. Args: topology: Parsed topology being validated. issues: Mutable list receiving validation failures. Returns: Node index prepared for later cross-reference checks. """ node_index = _build_node_index(topology, issues) for node_number, node in enumerate(topology.nodes): # keep node's location for error reporting node_location = "nodes[%d]" % node_number _validate_node_settings(node, node_location, issues) if node.interfaces is None: continue gateway_locations: List[str] = [] for interface_name, interface in node.interfaces.items(): # keep interface's location for error reporting interface_location = "%s.interfaces.%s" % ( node_location, interface_name, ) if interface.name != interface_name: # yes I have name inside the interface _add_issue( issues, interface_location, "interface object name %r does not match its declaration tag %r" % (interface.name, interface_name), ) _validate_interface( node, interface, interface_location, issues ) if interface.gateway is not None: gateway_locations.append(interface_location + ".gateway") if len(gateway_locations) > 1: for gateway_location in gateway_locations: _add_issue( issues, gateway_location, "only one static default gateway is supported per node", ) return node_index # ============================ Link validation ============================ def _validate_link_settings( settings: LinkSettings, location: str, issues: List[ValidationIssue], ) -> None: """Validate optional link provisioning settings. Args: settings: Link settings being validated. location: Logical location of the settings object (where the settings data came from). issues: Mutable list receiving validation failures. """ numeric_settings = { "bandwidth_mbps": settings.bandwidth_mbps, "delay_ms": settings.delay_ms, "loss_percent": settings.loss_percent, "jitter_ms": settings.jitter_ms, } for name, value in numeric_settings.items(): if value is not None and not math.isfinite(value): _add_issue( issues, "%s.%s" % (location, name), "value must be finite", ) if settings.bandwidth_mbps is not None and settings.bandwidth_mbps <= 0: _add_issue( issues, location + ".bandwidth_mbps", "bandwidth must be greater than zero", ) if settings.delay_ms is not None and settings.delay_ms < 0: _add_issue( issues, location + ".delay_ms", "delay must not be negative", ) if settings.jitter_ms is not None and settings.jitter_ms < 0: _add_issue( issues, location + ".jitter_ms", "jitter must not be negative", ) if ( settings.loss_percent is not None and not 0 <= settings.loss_percent <= 100 ): _add_issue( issues, location + ".loss_percent", "loss percentage must be between 0 and 100", ) def _validate_link_names( topology: Topology, issues: List[ValidationIssue], ) -> None: """Validate optional logical link names. Args: topology: Parsed topology being validated. issues: Mutable list receiving validation failures. """ first_index: Dict[str, int] = {} for index, link in enumerate(topology.links): if link.name is None: continue location = "links[%d].name" % index if not link.name: _add_issue(issues, location, "link name must not be empty") continue if link.name in first_index: _add_issue( issues, location, "duplicate link name %r; first declared at links[%d]" % (link.name, first_index[link.name]), ) continue first_index[link.name] = index def _validate_links( topology: Topology, node_index: _NodeIndex, issues: List[ValidationIssue], ) -> None: """Validate links, endpoint references, and interface allocation capacity. First collect valid explicit interface usage and AUTO requests while checking endpoint references and rejecting explicit reuse. Then compare each fixed pool's AUTO demand with its remaining capacity after all explicit reservations are known. This checks allocation feasibility without selecting concrete interfaces; dynamic pools need no capacity check. Args: topology: Parsed topology being validated. node_index: Node lookup information prepared by node validation. issues: Mutable list receiving validation failures. Example: For otherwise valid links involving a fixed pool on an example switch s1: Declared interfaces: p1, p2 Explicit reservation: p2 AUTO requests: 2 Remaining capacity: 1 Result: None; one capacity issue appended With only one AUTO request, no capacity issue is appended. Neither case consumes the pool or assigns an interface to AUTO. """ _validate_link_names(topology, issues) used_explicit: Dict[Tuple[str, str], str] = {} auto_requests: Dict[str, List[str]] = {} for link_index, link in enumerate(topology.links): _validate_link_settings( link.settings, "links[%d].settings" % link_index, issues, ) for endpoint_index, endpoint in enumerate(link.endpoints): # keep endpoint's location for error reporting endpoint_location = "links[%d].endpoints[%d]" % ( link_index, endpoint_index, ) if endpoint.node not in node_index.nodes: _add_issue( issues, endpoint_location + ".node", "unknown node %r" % endpoint.node, ) continue if endpoint.node in node_index.duplicate_names: # The duplicate-name error has already been reported. # Cross-reference resolution would be ambiguous, so avoid # generating secondary diagnostics from an arbitrary instance. continue node = node_index.nodes[endpoint.node] if endpoint.interface is LinkInterfaceSelector.AUTO: if node.interfaces is not None: auto_requests.setdefault(endpoint.node, []).append( endpoint_location + ".interface" ) continue interface_name = endpoint.interface if not interface_name: _add_issue( issues, endpoint_location + ".interface", "interface tag must not be empty", ) continue if node.interfaces is None: _add_issue( issues, endpoint_location + ".interface", "explicit interface %r cannot be referenced because node %r " "does not declare an interface set" % (interface_name, endpoint.node), ) continue if interface_name not in node.interfaces: _add_issue( issues, endpoint_location + ".interface", "node %r has no declared interface %r" % (endpoint.node, interface_name), ) continue key = (endpoint.node, interface_name) if key in used_explicit: _add_issue( issues, endpoint_location + ".interface", "interface %s:%s is already used by %s" % ( endpoint.node, interface_name, used_explicit[key], ), ) continue used_explicit[key] = endpoint_location # Count AUTO capacity only after collecting every explicit reservation. for node_name, request_locations in auto_requests.items(): node = node_index.nodes[node_name] # Dynamic check if node.interfaces is None: continue used_count = sum( 1 for used_node, _ in used_explicit if used_node == node_name ) free_count = len(node.interfaces) - used_count if len(request_locations) > free_count: shortage = len(request_locations) - free_count _add_issue( issues, "links", "node %r has %d automatic interface request%s but only %d " "unused declared interface%s remain%s" % ( node_name, len(request_locations), "" if len(request_locations) == 1 else "s", free_count, "" if free_count == 1 else "s", "; %d request%s cannot be satisfied" % ( shortage, "" if shortage == 1 else "s", ), ), # Ok, laugh as you want, but I like my plurals! ) # =========================== Service validation =========================== def _resolve_service_interface( service: Service, service_location: str, node: Node, issues: List[ValidationIssue], ) -> Optional[Interface]: """Resolve a service binding to one declared interface when possible. Explicit tags select a declared model directly; AUTO requires exactly one declared interface with static addressing. Service AUTO never allocates an interface or consumes a link's free pool. ALL is left unresolved for the service-specific validator to diagnose. Args: service: Service whose interface binding is resolved. service_location: Logical location of the service (whre service data came from). node: Node on which the service runs. issues: Mutable list receiving validation failures. Returns: Resolved Interface when the binding identifies exactly one valid interface, otherwise None. Example: With service.interface = ServiceInterfaceSelector.AUTO and issues = []: Declared static ifaces: lan Result: node.interfaces["lan"] Issues: unchanged With two declared static interfaces, lan and wan: Result: None Issues: one ambiguity issue appended at the service interface These scenarios describe selection only; address validity and link attachment are checked elsewhere in the pipeline. """ selector = service.interface if isinstance(selector, str): if node.interfaces is None: _add_issue( issues, service_location + ".interface", "explicit interface %r cannot be referenced because node %r " "does not declare an interface set" % (selector, node.name), ) return None if selector not in node.interfaces: _add_issue( issues, service_location + ".interface", "node %r has no declared interface %r" % (node.name, selector), ) return None return node.interfaces[selector] if selector is ServiceInterfaceSelector.ALL: return None if selector is ServiceInterfaceSelector.AUTO: if node.interfaces is None: _add_issue( issues, service_location + ".interface", "automatic service binding requires declared interfaces", ) return None candidates = [ interface for interface in node.interfaces.values() if interface.addressing is AddressingMode.STATIC ] if not candidates: _add_issue( issues, service_location + ".interface", "automatic service binding found no statically addressed interface", ) return None if len(candidates) > 1: _add_issue( issues, service_location + ".interface", "automatic service binding is ambiguous across %d statically " "addressed interfaces" % len(candidates), ) return None return candidates[0] return None def _validate_dhcp_server( service: Service, service_location: str, interface: Optional[Interface], issues: List[ValidationIssue], ) -> None: """Validate DHCP-server-specific semantics. Check address syntax and range ordering independently of the binding, so those diagnostics remain available even when interface resolution fails. With a usable static binding, check subnet membership and reserved subnet addresses, then exclude the server and advertised gateway from the pool. Skip checks whose required address or binding could not be parsed. Args: service: DHCP server service being validated. service_location: Logical location of the service (where service data came from). interface: Resolved service interface, if available. issues: Mutable list receiving validation failures. Example: For a valid static binding and otherwise valid service settings: Server interface: 192.168.1.1/24 Advertised gateway: omitted Pool: 192.168.1.100 through 192.168.1.200 Result: None; no issues appended Changing the pool to 192.168.1.1 through 192.168.1.200 still keeps it inside the subnet, but includes the server's own address. The result remains None, with one server-address exclusion issue appended to the supplied issues list. """ if service.interface is ServiceInterfaceSelector.ALL: _add_issue( issues, service_location + ".interface", "DHCP server does not support the 'all' interface selector", ) if not isinstance(service.settings, DHCPServerSettings): _add_issue( issues, service_location + ".settings", "DHCP server has an incompatible settings object", ) return start = _validate_ipv4_address( service.settings.address_range.start, service_location + ".settings.range.start", issues, ) end = _validate_ipv4_address( service.settings.address_range.end, service_location + ".settings.range.end", issues, ) if start is not None and end is not None and int(start) > int(end): _add_issue( issues, service_location + ".settings.range", "DHCP range start must not be greater than range end", ) gateway: Optional[ipaddress.IPv4Address] = None if service.settings.gateway is not None: gateway = _validate_ipv4_address( service.settings.gateway, service_location + ".settings.gateway", issues, ) if interface is None: return if interface.addressing is not AddressingMode.STATIC: _add_issue( issues, service_location + ".interface", "DHCP server must bind to a statically addressed interface", ) return if interface.address is None: # The interface validator reports this as well. return parsed_interface = _validate_ipv4_interface( interface.address, service_location + ".interface", issues, ) if parsed_interface is None: return network = parsed_interface.network for value, location, label in ( (start, service_location + ".settings.range.start", "range start"), (end, service_location + ".settings.range.end", "range end"), ): if value is None: continue if value not in network: _add_issue( issues, location, "%s must belong to DHCP interface subnet %s" % (label, network), ) continue if ( network.num_addresses > 2 and value in {network.network_address, network.broadcast_address} ): _add_issue( issues, location, "%s must not be the network or broadcast address" % label, ) if gateway is not None: if gateway not in network: _add_issue( issues, service_location + ".settings.gateway", "DHCP gateway must belong to DHCP interface subnet %s" % network, ) elif ( network.num_addresses > 2 and gateway in {network.network_address, network.broadcast_address} ): _add_issue( issues, service_location + ".settings.gateway", "DHCP gateway must not be the network or broadcast address", ) if ( start is not None and end is not None and int(start) <= int(end) ): server_address = parsed_interface.ip if int(start) <= int(server_address) <= int(end): _add_issue( issues, service_location + ".settings.range", "DHCP range must not include the server interface address %s" % server_address, ) if ( gateway is not None and int(start) <= int(gateway) <= int(end) ): _add_issue( issues, service_location + ".settings.range", "DHCP range must not include the advertised gateway %s" % gateway, ) def _validate_services( topology: Topology, node_index: _NodeIndex, issues: List[ValidationIssue], ) -> None: """Validate service references and service-specific semantics. Args: topology: Parsed topology being validated. node_index: Node lookup information prepared by node validation. issues: Mutable list receiving validation failures. """ for index, service in enumerate(topology.services): location = "services[%d]" % index if service.node not in node_index.nodes: _add_issue( issues, location + ".node", "unknown node %r" % service.node, ) continue if service.node in node_index.duplicate_names: continue node = node_index.nodes[service.node] interface = _resolve_service_interface( service, location, node, issues, ) if service.type is ServiceType.DHCP_SERVER: _validate_dhcp_server( service, location, interface, issues, ) # ============================ Public interface ============================ def validate(topology: Topology) -> None: """Validate the semantic consistency of a parsed topology. Args: topology: Parsed topology produced by topology_parser.parse_topology(). Returns: None when no semantic issues are found. Example: Input constructed through the public parser: topology = parse_topology({ "nodes": [{"name": "h1", "type": "host"}], }) Output: result = validate(topology) result is None Invalid input: topology = parse_topology({ "nodes": [ {"name": "h1", "type": "host"}, {"name": "h1", "type": "host"}, ], }) validate(topology) raises TopologyValidationError instead of returning. The exception's issues tuple contains one ValidationIssue with location="nodes[1].name" and a duplicate-node-name message. When multiple issues are discovered, they share one exception. Raises: TopologyValidationError: If one or more semantic consistency errors are discovered. Notes: Validation is non-destructive and does not invoke Mininet or execute Linux networking commands. All discovered semantic issues are collected before the exception is raised. """ issues: List[ValidationIssue] = [] node_index = _validate_nodes(topology, issues) _validate_links(topology, node_index, issues) _validate_services(topology, node_index, issues) if issues: raise TopologyValidationError(issues) # =============================== Self-test =============================== def _self_test() -> None: """Run smoke tests against the public validate() interface.""" from topology_parser import parse_topology valid_raw = { "nodes": [ { "name": "h1", "type": "host", "interfaces": { "net0": { "addressing": "dhcp", } }, }, { "name": "h3", "type": "host", "interfaces": { "net0": { "addressing": "static", "address": "192.168.2.10/24", "gateway": "192.168.2.1", } }, }, { "name": "s1", "type": "switch", }, { "name": "s2", "type": "switch", }, { "name": "r0", "type": "router", "interfaces": { "lan": { "addressing": "static", "address": "192.168.1.1/24", }, "wan": { "addressing": "static", "address": "192.168.2.1/24", }, }, "settings": { "ip_forward": True, "rp_filter": False, }, }, ], "links": [ { "endpoints": [ {"node": "h1", "interface": "net0"}, {"node": "s1", "interface": "auto"}, ] }, { "endpoints": [ {"node": "r0", "interface": "lan"}, {"node": "s1", "interface": "auto"}, ] }, { "endpoints": [ {"node": "r0", "interface": "wan"}, {"node": "s2", "interface": "auto"}, ] }, { "endpoints": [ {"node": "h3", "interface": "net0"}, {"node": "s2", "interface": "auto"}, ] }, ], "services": [ { "type": "dhcp-server", "node": "r0", "interface": "lan", "settings": { "range": { "start": "192.168.1.100", "end": "192.168.1.200", }, "gateway": "192.168.1.1", }, } ], } validate(parse_topology(valid_raw)) invalid_raw = { "nodes": [ { "name": "h1", "type": "host", "interfaces": { "net0": { "addressing": "dhcp", "gateway": "192.168.1.1", } }, }, { "name": "h1", "type": "host", }, { "name": "s1", "type": "switch", "interfaces": { "p1": {}, }, }, ], "links": [ { "endpoints": [ {"node": "missing", "interface": "auto"}, {"node": "s1", "interface": "p1"}, ] }, { "endpoints": [ {"node": "s1", "interface": "p1"}, {"node": "s1", "interface": "auto"}, ] }, ], "services": [], } try: validate(parse_topology(invalid_raw)) except TopologyValidationError as exc: assert len(exc.issues) >= 4 assert any("duplicate node name" in issue.message for issue in exc.issues) assert any("unknown node" in issue.message for issue in exc.issues) assert any("already used" in issue.message for issue in exc.issues) assert any("only valid with static addressing" in issue.message for issue in exc.issues) else: raise AssertionError( "invalid topology must raise TopologyValidationError" ) if __name__ == "__main__": _self_test() print("topology_validator: self-test passed")