1097 lines
31 KiB
Python
1097 lines
31 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
|
|
|
|
|
|
# ======================= Semantic model and errors =======================
|
|
|
|
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 applied to the concrete interface.
|
|
mtu: [Optional] MTU in bytes applied to the concrete interface.
|
|
|
|
Notes:
|
|
MAC and MTU are independent of the addressing mode.
|
|
The generated program applies them using Linux ip link commands.
|
|
"""
|
|
|
|
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.
|
|
Keping 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
|
|
|
|
# Mutable global
|
|
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:
|
|
All four settings are implemented through Mininet TCLink parameters
|
|
in the generated program.
|
|
Bandwidth maps to bw, loss to loss, and delay and jitter to millisecond strings.
|
|
Additional provisioning settings require corresponding parser,
|
|
validator, planner, and renderer support before they can be used.
|
|
"""
|
|
|
|
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
|
|
|
|
|
|
# Mutable global
|
|
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.
|
|
|
|
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)
|
|
|
|
|
|
# ============================ Parsing filters ============================
|
|
|
|
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 "", # don't forget your English grammar ;)
|
|
", ".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
|
|
# bool is a subclass of int, but JSON booleans are not numeric settings.
|
|
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
|
|
# Reject booleans before accepting Python's integer and float types.
|
|
if isinstance(value, bool) or not isinstance(value, (int, float)):
|
|
raise TopologyParseError("%s must be a number" % context)
|
|
return float(value)
|
|
|
|
# ============================= Entity parsing =============================
|
|
|
|
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.
|
|
|
|
Example:
|
|
Input (diagnostic context omitted):
|
|
name = "lan"
|
|
raw = {
|
|
"addressing": "static",
|
|
"address": "192.168.1.1/24",
|
|
}
|
|
|
|
Output:
|
|
Interface(
|
|
name="lan",
|
|
addressing=AddressingMode.STATIC,
|
|
address="192.168.1.1/24",
|
|
gateway=None,
|
|
mac=None,
|
|
mtu=None,
|
|
)
|
|
|
|
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 (dynamic), otherwise
|
|
- a dictionary of logical interface tags to Interface objects.
|
|
|
|
Example:
|
|
Each case shows the value passed as raw_node and the returned value,
|
|
the diagnostic context argument is omitted.
|
|
raw_node is the node object, not just its interfaces member.
|
|
Unrelated node fields are omitted from these examples.
|
|
|
|
Interface set raw_node["interfaces"] return
|
|
--------------------------------------------------------
|
|
Omitted: {} None
|
|
Explicitly empty: {"interfaces": {}} {}
|
|
One declared: {"interfaces": {"lan": {}}} {"lan": Interface(name="lan")}
|
|
|
|
Omission preserves a dynamic interface set. An empty object preserves
|
|
a fixed pool containing zero interfaces.
|
|
Interface(name="lan") uses the model defaults, including NONE
|
|
addressing and no address, gateway, MAC, or MTU.
|
|
|
|
Raises:
|
|
TopologyParseError: If the interfaces field has an invalid shape.
|
|
"""
|
|
|
|
# Preserve the distinction between dynamic allocation and a fixed empty pool.
|
|
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.
|
|
|
|
Example:
|
|
Input (Python values after JSON loading, diagnostic context omitted):
|
|
node_type = NodeType.ROUTER
|
|
raw = {"ip_forward": True, "rp_filter": False}
|
|
|
|
Output:
|
|
RouterSettings(ip_forward=True, rp_filter=False)
|
|
|
|
For node_type = NodeType.HOST and raw = {}, the output is
|
|
HostSettings().
|
|
|
|
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.
|
|
|
|
Example:
|
|
Input raw (diagnostic index omitted):
|
|
{
|
|
"name": "h1",
|
|
"type": "host",
|
|
"interfaces": {"net0": {"addressing": "dhcp"}},
|
|
}
|
|
|
|
Output:
|
|
Node(
|
|
name="h1",
|
|
type=NodeType.HOST,
|
|
interfaces={
|
|
"net0": Interface(
|
|
name="net0", addressing=AddressingMode.DHCP,
|
|
),
|
|
},
|
|
settings=HostSettings(),
|
|
)
|
|
|
|
Unspecified Interface fields use their model defaults.
|
|
|
|
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.
|
|
|
|
Example:
|
|
Input raw (diagnostic index omitted):
|
|
{
|
|
"endpoints": [
|
|
{"node": "h1", "interface": "net0"},
|
|
{"node": "s1", "interface": "auto"},
|
|
],
|
|
}
|
|
|
|
Output:
|
|
Link(
|
|
endpoints=(
|
|
LinkEndpoint(node="h1", interface="net0"),
|
|
LinkEndpoint(
|
|
node="s1", interface=LinkInterfaceSelector.AUTO,
|
|
),
|
|
),
|
|
)
|
|
Note:
|
|
- Omitted name and settings use the Link defaults.
|
|
- Parsing converts the endpoint list to a tuple and recognizes AUTO.
|
|
- Validation checks whether the referenced nodes and interfaces exist.
|
|
|
|
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.
|
|
|
|
Example:
|
|
Input raw (diagnostic context omitted):
|
|
{
|
|
"range": {
|
|
"start": "192.168.1.100",
|
|
"end": "192.168.1.200",
|
|
},
|
|
"gateway": "192.168.1.1",
|
|
}
|
|
|
|
Output:
|
|
DHCPServerSettings(
|
|
address_range=DHCPRange(
|
|
start="192.168.1.100", end="192.168.1.200",
|
|
),
|
|
gateway="192.168.1.1",
|
|
)
|
|
|
|
The JSON range object becomes the typed address_range field.
|
|
|
|
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.
|
|
|
|
Example:
|
|
Input raw (diagnostic index omitted):
|
|
{
|
|
"type": "dhcp-server",
|
|
"node": "r0",
|
|
"interface": "lan",
|
|
"settings": {
|
|
"range": {
|
|
"start": "192.168.1.100",
|
|
"end": "192.168.1.200",
|
|
},
|
|
"gateway": "192.168.1.1",
|
|
},
|
|
}
|
|
|
|
Output:
|
|
Service(
|
|
type=ServiceType.DHCP_SERVER,
|
|
node="r0",
|
|
interface="lan",
|
|
settings=DHCPServerSettings(
|
|
address_range=DHCPRange(
|
|
start="192.168.1.100", end="192.168.1.200",
|
|
),
|
|
gateway="192.168.1.1",
|
|
),
|
|
)
|
|
|
|
Binding fields remain separate from service settings.
|
|
Parsing preserves the references. Validation checks their validity.
|
|
|
|
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,
|
|
)
|
|
|
|
|
|
|
|
|
|
# ============================ Public interface ============================
|
|
|
|
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)
|
|
|
|
|
|
|
|
# =============================== Self-test ===============================
|
|
|
|
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")
|