931 lines
25 KiB
Python
931 lines
25 KiB
Python
"""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 <cchoutou@ece.auth.gr>
|
|
"""
|
|
|
|
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")
|