hoo2/Computer-Networks-II
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Documentation, comments and small changes

Browse Source
This commit is contained in:
hoo2
2026-09-27 16:18:35 +03:00
parent 0e43b943c4
commit e1e164bd65
8 changed files with 653 additions and 211 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.gitmodules
+3
View File
@@ -0,0 +1,3 @@
[submodule "report/AUThReport"]
path = report/AUThReport
url = ssh://git@git.hoo2.net:222/hoo2/AUThReport.git
report/AUThReport
Submodule
+1
Submodule report/AUThReport added at 74ec4b5f6c
source/build_network.py
Regular → Executable
+14 -4
View File
@@ -29,13 +29,15 @@ import subprocess
import sys
from typing import Optional, Sequence, Tuple, Union
from topology_loader import TopologyLoadError, load
from topology_parser import TopologyParseError
from topology_loader import TopologyLoadError, load_topology
from topology_parser import TopologyParseError, parse_topology
from topology_plan import ExecutionPlan, TopologyPlanError, plan
from topology_renderer import render_python
from topology_validator import TopologyValidationError, validate
# ============================ Types and errors ============================
PathLike = Union[str, Path]
@@ -76,6 +78,8 @@ class EnvironmentReport:
return all(item.available for item in self.requirements)
# =========================== Environment checks ===========================
def _check_python_module(name: str, install_hint: str) -> EnvironmentRequirement:
"""Check whether one Python module is importable.
@@ -193,6 +197,8 @@ def check_environment() -> EnvironmentReport:
return EnvironmentReport(requirements=tuple(requirements))
# =================== Compilation and artifact execution ===================
def compile_topology(path: PathLike) -> Tuple[ExecutionPlan, str]:
"""Compile one topology file into a plan and executable Python source.
@@ -212,7 +218,7 @@ def compile_topology(path: PathLike) -> Tuple[ExecutionPlan, str]:
complete execution plan.
"""
topology = load(path)
topology = parse_topology(load_topology(path))
validate(topology)
execution_plan = plan(topology)
source = render_python(execution_plan)
@@ -241,7 +247,7 @@ def write_generated_python(source: str, path: PathLike) -> Path:
path: Destination path for the generated artifact.
Returns:
Resolved destination Path.
Destination Path, remaining relative if the supplied path is relative.
Raises:
BuildNetworkError: If the artifact cannot be written.
@@ -294,6 +300,8 @@ def execute_generated_python(path: PathLike) -> int:
return completed.returncode
# ================== Command-line reporting and arguments ==================
def _print_environment_report(report: EnvironmentReport) -> None:
"""Print an environment report from the command-line layer.
@@ -350,6 +358,8 @@ def _build_argument_parser() -> argparse.ArgumentParser:
return parser
# ======================== Command-line entry point ========================
def main(argv: Optional[Sequence[str]] = None) -> int:
"""Run the coordinator command-line interface.
source/topology_loader.py
+9 -19
View File
@@ -15,13 +15,16 @@ Author: Christos Choutouridis <cchoutou@ece.auth.gr>
"""
import json
from pathlib import Path
from pathlib import Path
from tempfile import NamedTemporaryFile
from typing import Any, Dict, Union
from typing import Any, Dict, Union
from topology_parser import Topology, parse_topology
# ============================ Types and errors ============================
PathLike = Union[str, Path]
@@ -29,6 +32,8 @@ class TopologyLoadError(ValueError):
"""Raised when a topology file cannot be loaded as a JSON object."""
# ============================ Public interface ============================
def load_topology(path: PathLike) -> Dict[str, Any]:
"""Load a topology JSON file into a raw dictionary.
@@ -66,23 +71,8 @@ def load_topology(path: PathLike) -> Dict[str, Any]:
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))
# =============================== Self-test ===============================
def _self_test() -> None:
"""Run a small smoke test against load_topology() and load()."""
@@ -108,7 +98,7 @@ def _self_test() -> None:
handle.flush()
raw = load_topology(handle.name)
topology = load(handle.name)
topology = parse_topology(load_topology(handle.name))
assert raw["nodes"][0]["name"] == "s1"
assert topology.nodes[0].name == "s1"
source/topology_parser.py
+264 -98
View File
@@ -3,18 +3,20 @@
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
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
from enum import Enum
from typing import Any, Dict, List, Mapping, Optional, Set, Tuple, Union
# ======================= Semantic model and errors =======================
JSONMapping = Mapping[str, Any]
@@ -25,7 +27,7 @@ class TopologyParseError(ValueError):
class NodeType(Enum):
"""Supported logical node types."""
HOST = "host"
HOST = "host"
SWITCH = "switch"
ROUTER = "router"
@@ -33,9 +35,9 @@ class NodeType(Enum):
class AddressingMode(Enum):
"""Supported Layer-3 addressing modes for an interface."""
NONE = "none"
NONE = "none"
STATIC = "static"
DHCP = "dhcp"
DHCP = "dhcp"
class LinkInterfaceSelector(Enum):
@@ -48,7 +50,7 @@ class ServiceInterfaceSelector(Enum):
"""Special interface selectors that are valid for a service binding."""
AUTO = "auto"
ALL = "all"
ALL = "all"
class ServiceType(Enum):
@@ -62,29 +64,33 @@ 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: 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
name: str
addressing: AddressingMode = AddressingMode.NONE
address: Optional[str] = None
gateway: Optional[str] = None
mac: Optional[str] = None
mtu: Optional[int] = 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.
The current grammar does not require host-specific settings yet.
Keeping a dedicated type gives the model a stable extension point.
"""
@@ -92,8 +98,8 @@ class HostSettings:
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.
The current grammar does not require switch-specific settings yet.
Keping a dedicated type gives the model a stable extension point.
"""
@@ -107,9 +113,9 @@ class RouterSettings:
"""
ip_forward: Optional[bool] = None
rp_filter: Optional[bool] = None
rp_filter: Optional[bool] = None
# Mutable global
NodeSettings = Union[HostSettings, SwitchSettings, RouterSettings]
@@ -118,19 +124,21 @@ 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.
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
name: str
type: NodeType
interfaces: Optional[Dict[str, Interface]]
settings: NodeSettings
settings: NodeSettings
@dataclass
@@ -138,11 +146,11 @@ class LinkEndpoint:
"""Describe one endpoint of a point-to-point logical link.
Args:
node: Name of the node referenced by the endpoint.
node: Name of the node referenced by the endpoint.
interface: Explicit logical interface tag or the AUTO selector.
"""
node: str
node: str
interface: Union[str, LinkInterfaceSelector]
@@ -151,22 +159,23 @@ 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.
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.
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
delay_ms: Optional[float] = None
loss_percent: Optional[float] = None
jitter_ms: Optional[float] = None
@dataclass
@@ -175,13 +184,13 @@ class Link:
Args:
endpoints: Ordered pair of link endpoints.
name: Optional logical link name used for diagnostics or extensions.
settings: Link-specific provisioning settings.
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)
name: Optional[str] = None
settings: LinkSettings = field(default_factory=LinkSettings)
@dataclass
@@ -190,11 +199,11 @@ class DHCPRange:
Args:
start: First address in the DHCP allocation range.
end: Last address in the DHCP allocation range.
end: Last address in the DHCP allocation range.
"""
start: str
end: str
end: str
@dataclass
@@ -203,13 +212,14 @@ class DHCPServerSettings:
Args:
address_range: Address range offered to DHCP clients.
gateway: Optional default gateway advertised to DHCP clients.
gateway: [Optional] default gateway advertised to DHCP clients.
"""
address_range: DHCPRange
gateway: Optional[str] = None
gateway: Optional[str] = None
# Mutable global
ServiceSettings = DHCPServerSettings
@@ -218,44 +228,48 @@ 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.
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.
settings: Type-specific service settings.
"""
type: ServiceType
node: str
interface: Union[str, ServiceInterfaceSelector]
settings: ServiceSettings
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:
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)
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.
value: Value expected to be a JSON object.
context: Human-readable location used in the error message.
Returns:
@@ -278,7 +292,7 @@ def _reject_unknown_fields(
"""Reject object members that are not part of the grammar.
Args:
data: Parsed JSON object whose member names are checked.
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.
@@ -292,7 +306,7 @@ def _reject_unknown_fields(
"%s contains unknown field%s: %s"
% (
context,
"s" if len(unknown) != 1 else "",
"s" if len(unknown) != 1 else "", # don't forget your English grammar ;)
", ".join(repr(name) for name in unknown),
)
)
@@ -302,7 +316,7 @@ 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.
value: Value expected to be a JSON array.
context: Human-readable location used in the error message.
Returns:
@@ -321,7 +335,7 @@ 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.
value: Value expected to be a string.
context: Human-readable location used in the error message.
Returns:
@@ -340,7 +354,7 @@ 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.
value: None or a value expected to be a string.
context: Human-readable location used in the error message.
Returns:
@@ -359,7 +373,7 @@ 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.
value: None or a value expected to be a boolean.
context: Human-readable location used in the error message.
Returns:
@@ -380,7 +394,7 @@ 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.
value: None or a value expected to be an integer.
context: Human-readable location used in the error message.
Returns:
@@ -392,6 +406,7 @@ def _optional_int(value: Any, context: str) -> Optional[int]:
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
@@ -401,8 +416,7 @@ 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.
value: None or a value expected to be an integer or floating-point number.
context: Human-readable location used in the error message.
Returns:
@@ -414,22 +428,42 @@ def _optional_float(value: Any, context: str) -> Optional[float]:
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.
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.
"""
@@ -464,16 +498,34 @@ def _parse_interfaces(raw_node: JSONMapping, context: str) -> Optional[Dict[str,
Args:
raw_node: Raw JSON node object.
context: Human-readable node location used in error messages.
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.
- 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
@@ -496,12 +548,23 @@ def _parse_node_settings(node_type: NodeType, raw: Any, context: str) -> NodeSet
Args:
node_type: Parsed semantic node type.
raw: Raw JSON settings object or None.
context: Human-readable node location used in error messages.
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.
@@ -538,12 +601,34 @@ def _parse_node(raw: Any, index: int) -> Node:
"""Parse one node declaration.
Args:
raw: Raw JSON node object.
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.
"""
@@ -578,7 +663,7 @@ def _parse_link_interface(value: Any, context: str) -> Union[str, LinkInterfaceS
"""Parse a link endpoint interface reference.
Args:
value: Raw interface reference.
value: Raw interface reference.
context: Human-readable location used in error messages.
Returns:
@@ -598,7 +683,7 @@ def _parse_link_endpoint(raw: Any, context: str) -> LinkEndpoint:
"""Parse one link endpoint.
Args:
raw: Raw JSON endpoint object.
raw: Raw JSON endpoint object.
context: Human-readable location used in error messages.
Returns:
@@ -624,7 +709,7 @@ def _parse_link_settings(raw: Any, context: str) -> LinkSettings:
"""Parse optional link-specific provisioning settings.
Args:
raw: Raw JSON settings object or None.
raw: Raw JSON settings object or None.
context: Human-readable link location used in error messages.
Returns:
@@ -665,12 +750,35 @@ def _parse_link(raw: Any, index: int) -> Link:
"""Parse one link declaration.
Args:
raw: Raw JSON link object.
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.
"""
@@ -712,7 +820,7 @@ def _parse_service_interface(
"""Parse a service interface binding.
Args:
value: Raw service interface reference.
value: Raw service interface reference.
context: Human-readable location used in error messages.
Returns:
@@ -735,12 +843,32 @@ def _parse_dhcp_server_settings(raw: Any, context: str) -> DHCPServerSettings:
"""Parse DHCP-server-specific settings.
Args:
raw: Raw JSON DHCP server settings object.
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.
@@ -767,12 +895,43 @@ def _parse_service(raw: Any, index: int) -> Service:
"""Parse one service declaration.
Args:
raw: Raw JSON service object.
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.
"""
@@ -814,6 +973,10 @@ def _parse_service(raw: Any, index: int) -> Service:
)
# ============================ Public interface ============================
def parse_topology(raw: Mapping[str, Any]) -> Topology:
"""Parse a raw JSON dictionary into the typed topology model.
@@ -828,8 +991,8 @@ def parse_topology(raw: Mapping[str, Any]) -> Topology:
topology grammar.
Notes:
This function intentionally does not validate cross references such as
whether a link references an existing node. Those checks belong to
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.
"""
@@ -858,6 +1021,9 @@ def parse_topology(raw: Mapping[str, Any]) -> Topology:
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."""
source/topology_plan.py
+75 -10
View File
@@ -32,6 +32,8 @@ from topology_parser import (
)
# ========================== Execution plan model ==========================
class TopologyPlanError(ValueError):
"""Raised when a validated topology cannot be lowered into a plan."""
@@ -182,6 +184,12 @@ class PlannedService:
class ExecutionPlan:
"""Contain the complete, fully resolved execution plan.
Ordered tuples of frozen operation objects capture the result of planning.
They carry concrete names and configuration into the renderer without
exposing the mutable allocation context or the source model objects.
The renderer emits the operations in lifecycle order: construction,
configuration, and service startup.
Args:
nodes: Node creation operations in declaration order.
links: Link creation operations in declaration order.
@@ -201,21 +209,45 @@ class ExecutionPlan:
services: Tuple[PlannedService, ...]
# ======================== Mutable planning context ========================
@dataclass
class _PlanningContext:
"""Hold mutable state used while lowering a topology.
_build_context() initializes node lookup, reservations, free pools, and
counters from the validated topology.
During _plan_links(), allocation helpers consume pool entries, advance
counters, and record interface mappings.
Interface, node-setting, and service planning then read those mappings.
A logical interface is identified by (node_name, logical_name), since tags
need only be unique within their owning node.
interface_names and interface_models acquire the same keys on allocation:
one supplies the backend name, and the other supplies its configuration.
A reservation alone does not create an entry in either mapping.
Args:
nodes: Nodes indexed by logical name.
reserved_interfaces: Explicitly referenced logical interfaces per node.
free_interfaces: Remaining declared interfaces available to AUTO.
dynamic_interface_counters: Synthetic logical-name counters for nodes
whose interface sets were omitted.
physical_interface_counters: Concrete eth-index counters per node.
interface_names: Mapping from resolved logical interfaces to concrete
Linux/Mininet interface names.
interface_models: Mapping from resolved logical interfaces to their
source Interface objects.
nodes: Source nodes indexed by name, safe after uniqueness validation.
reserved_interfaces: All explicit link references per node, used to
construct free pools and retained unchanged during allocation.
free_interfaces: Ordered lists of unreserved declared tags for fixed
pools; AUTO removes the first entry.
An empty list means exhaustion; dynamic nodes have no entry.
dynamic_interface_counters: Next synthetic logical-tag indices, such
as auto0; advanced only for nodes with omitted interface sets.
physical_interface_counters: Next concrete eth-index per node;
advanced for every allocation, including explicit selections.
interface_names: Allocated (node, logical tag) pairs mapped to concrete
Linux/Mininet names, in interface-configuration order.
interface_models: The same allocated keys mapped to source Interface
objects or default models created for dynamic interfaces.
Example:
A fixed pool declared as p1, p2, uplink with an explicit uplink
reference starts with free_interfaces containing [p1, p2].
If the first endpoint for that node uses AUTO, it consumes p1 and
receives <node>-eth0, recording both mappings for (node, p1).
"""
nodes: Dict[str, Node]
@@ -227,9 +259,15 @@ class _PlanningContext:
interface_models: Dict[Tuple[str, str], Interface]
# ========================= Context initialization =========================
def _build_context(topology: Topology) -> _PlanningContext:
"""Create planning state and reserve all explicit link interfaces.
Scan every link before constructing the free-interface pools so an AUTO
endpoint cannot consume an interface explicitly referenced by a later link.
Preserve declaration order in each pool for deterministic allocation.
Args:
topology: Validated topology to lower.
@@ -267,6 +305,8 @@ def _build_context(topology: Topology) -> _PlanningContext:
)
# ========================= Node and link planning =========================
def _plan_nodes(topology: Topology) -> Tuple[PlannedNode, ...]:
"""Lower topology nodes into concrete node creation operations.
@@ -296,6 +336,11 @@ def _allocate_logical_interface(
) -> Tuple[str, Interface]:
"""Resolve one link endpoint to a concrete logical interface.
Explicit tags look up their declared models without consuming a free pool.
AUTO consumes the first available tag for a fixed pool, or creates a
synthetic tag and an unconfigured model for a dynamic pool.
Concrete backend names are assigned separately by physical allocation.
Args:
node: Node that owns the endpoint.
selector: Explicit interface tag or AUTO selector.
@@ -373,6 +418,7 @@ def _allocate_physical_interface(
context.physical_interface_counters[node_name] = index + 1
interface_name = "%s-eth%d" % (node_name, index)
# These mappings let later stages recover both the name and configuration.
context.interface_names[key] = interface_name
context.interface_models[key] = interface
@@ -385,6 +431,11 @@ def _plan_links(
) -> Tuple[PlannedLink, ...]:
"""Resolve all link endpoints and concrete interface names.
Traverse links and their endpoints in source order, resolving each logical
selection before assigning its backend name.
Allocation records mappings in the shared context for subsequent interface,
node-setting, and service planning, so traversal order determines naming.
Args:
topology: Validated topology to lower.
context: Mutable planning context.
@@ -436,6 +487,8 @@ def _plan_links(
return tuple(planned_links)
# ==================== Interface and node configuration ====================
def _interface_requires_materialization(interface: Interface) -> bool:
"""Return whether an unused declared interface carries configuration.
@@ -462,6 +515,12 @@ def _plan_interfaces(
) -> Tuple[PlannedInterface, ...]:
"""Create interface-configuration operations for materialized interfaces.
First reject configured declarations absent from the allocation mappings:
Mininet creates these interfaces through links, so there would be no
concrete interface on which to apply their configuration.
Empty unused declarations need no operation and may remain unallocated.
Then pair each allocated name with its model in allocation order.
Args:
topology: Validated topology to lower.
context: Planning context containing resolved interface mappings.
@@ -581,6 +640,8 @@ def _plan_node_settings(
return tuple(planned)
# ============================ Service planning ============================
def _resolve_service_logical_interface(
service: Service,
node: Node,
@@ -694,6 +755,8 @@ def _plan_services(
return tuple(planned)
# ============================ Public interface ============================
def plan(topology: Topology) -> ExecutionPlan:
"""Lower a validated topology into a fully resolved execution plan.
@@ -730,6 +793,8 @@ def plan(topology: Topology) -> ExecutionPlan:
)
# =============================== Self-test ===============================
def _self_test() -> None:
"""Run a smoke test against the public plan() interface."""
source/topology_renderer.py
+19
View File
@@ -24,6 +24,8 @@ from topology_plan import (
)
# =========================== Rendering helpers ===========================
def _python_string(value: str) -> str:
"""Return a safe Python string literal.
@@ -187,6 +189,12 @@ def _render_gateway_configuration(interface: PlannedInterface) -> List[str]:
def _render_service_start(service: PlannedService, index: int) -> List[str]:
"""Render startup code for one planned service.
Construct dnsmasq arguments from the resolved binding and DHCP settings,
using the service index for lease, PID, and log filenames.
Emit commands that remove previous lease and PID files, launch the daemon
in the background, and record its shell PID for later cleanup.
The emitted startup sequence does not check daemon readiness.
Args:
service: Fully resolved service operation.
index: Stable service index used for temporary runtime files.
@@ -238,9 +246,18 @@ def _render_service_start(service: PlannedService, index: int) -> List[str]:
]
# ============================ Public interface ============================
def render_python(plan: ExecutionPlan) -> str:
"""Render an execution plan as a complete executable Python program.
Assemble source lines for runtime helpers and the network lifecycle.
Emit construction first so interfaces exist before configuration; emit
addresses before default routes and services that depend on them.
Finish with the interactive CLI and a finally block that attempts tracked
service cleanup followed by network shutdown.
Rendering itself only produces text and does not execute these operations.
Args:
plan: Fully resolved execution plan returned by topology_plan.plan().
@@ -377,6 +394,8 @@ def render_python(plan: ExecutionPlan) -> str:
return "\n".join(lines)
# =============================== Self-test ===============================
def _self_test() -> None:
"""Run a smoke test against the public render_python() interface."""
source/topology_validator.py
+268 -80
View File
@@ -3,11 +3,9 @@
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
@@ -18,11 +16,11 @@ validation fails.
Author: Christos Choutouridis <cchoutou@ece.auth.gr>
"""
from dataclasses import dataclass
from dataclasses import dataclass
import ipaddress
import math
import re
from typing import Dict, List, Optional, Sequence, Set, Tuple
from typing import Dict, List, Optional, Sequence, Set, Tuple
from topology_parser import (
AddressingMode,
@@ -43,6 +41,8 @@ from topology_parser import (
)
# ========================== Validation constants ==========================
_RESERVED_INTERFACE_TAGS = {
LinkInterfaceSelector.AUTO.value,
ServiceInterfaceSelector.ALL.value,
@@ -53,6 +53,8 @@ _MAC_ADDRESS_RE = re.compile(
)
# ====================== Diagnostics and lookup state ======================
@dataclass(frozen=True)
class ValidationIssue:
"""Describe one semantic topology validation failure.
@@ -63,7 +65,7 @@ class ValidationIssue:
"""
location: str
message: str
message: str
class TopologyValidationError(ValueError):
@@ -96,7 +98,7 @@ class TopologyValidationError(ValueError):
count = len(self.issues)
header = "%d topology validation error%s" % (
count,
"" if count == 1 else "s",
"" if count == 1 else "s", # ;)
)
lines = [header]
@@ -110,26 +112,57 @@ class TopologyValidationError(ValueError):
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: Unique node declarations indexed by name.
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]
nodes: Dict[str, Node]
duplicate_names: Set[str]
# =========================== Validation helpers ===========================
def _add_issue(
issues: List[ValidationIssue],
issues: List[ValidationIssue],
location: str,
message: str,
message: str,
) -> None:
"""Append one validation issue.
Args:
issues: Mutable list receiving validation failures.
issues: Mutable list receiving validation failures.
location: Logical location associated with the failure.
message: Human-readable description of the failure.
message: Human-readable description of the failure.
"""
issues.append(ValidationIssue(location=location, message=message))
@@ -143,13 +176,21 @@ def _build_node_index(
Args:
topology: Parsed topology being validated.
issues: Mutable list receiving validation failures.
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] = {}
nodes: Dict[str, Node] = {}
first_index: Dict[str, int] = {}
duplicate_names: Set[str] = set()
@@ -179,17 +220,21 @@ def _build_node_index(
)
# ============================ Validation core =============================
def _validate_node_settings(
node: Node,
node: Node,
location: str,
issues: List[ValidationIssue],
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.
issues: Mutable list receiving validation failures.
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 = {
@@ -206,20 +251,46 @@ def _validate_node_settings(
)
def _parse_ipv4_interface(
value: str,
def _validate_ipv4_interface(
value: str,
location: str,
issues: List[ValidationIssue],
issues: List[ValidationIssue],
) -> Optional[ipaddress.IPv4Interface]:
"""Parse and validate one IPv4 CIDR interface address.
"""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.
issues: Mutable list receiving validation failures.
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:
Parsed IPv4Interface on success, otherwise None.
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:
@@ -243,20 +314,47 @@ def _parse_ipv4_interface(
return parsed
def _parse_ipv4_address(
value: str,
def _validate_ipv4_address(
value: str,
location: str,
issues: List[ValidationIssue],
issues: List[ValidationIssue],
) -> Optional[ipaddress.IPv4Address]:
"""Parse and validate one IPv4 address without a prefix.
"""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.
issues: Mutable list receiving validation failures.
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:
Parsed IPv4Address on success, otherwise None.
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:
@@ -273,18 +371,18 @@ def _parse_ipv4_address(
def _validate_interface(
node: Node,
node: Node,
interface: Interface,
location: str,
issues: List[ValidationIssue],
location: str,
issues: List[ValidationIssue],
) -> None:
"""Validate one declared logical interface.
Args:
node: Node that owns the interface.
node: Node that owns the interface.
interface: Interface being validated.
location: Logical location of the interface.
issues: Mutable list receiving validation failures.
location: Logical location of the interface (where the interface data came from).
issues: Mutable list receiving validation failures.
"""
if not interface.name:
@@ -314,16 +412,18 @@ def _validate_interface(
"static addressing requires an address",
)
else:
parsed_address = _parse_ipv4_interface(
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
network.num_addresses > 2 # /30 and up
and parsed_address.ip in {network.network_address, network.broadcast_address}
):
_add_issue(
@@ -333,7 +433,7 @@ def _validate_interface(
)
if interface.gateway is not None:
parsed_gateway = _parse_ipv4_address(
parsed_gateway = _validate_ipv4_address(
interface.gateway,
location + ".gateway",
issues,
@@ -353,6 +453,7 @@ def _validate_interface(
"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 {
@@ -398,7 +499,7 @@ def _validate_interface(
def _validate_nodes(
topology: Topology,
issues: List[ValidationIssue],
issues: List[ValidationIssue],
) -> _NodeIndex:
"""Validate node declarations and their local interface configuration.
@@ -413,6 +514,7 @@ def _validate_nodes(
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)
@@ -423,12 +525,14 @@ def _validate_nodes(
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,
@@ -437,10 +541,7 @@ def _validate_nodes(
)
_validate_interface(
node,
interface,
interface_location,
issues,
node, interface, interface_location, issues
)
if interface.gateway is not None:
@@ -457,24 +558,26 @@ def _validate_nodes(
return node_index
# ============================ Link validation ============================
def _validate_link_settings(
settings: LinkSettings,
location: str,
issues: List[ValidationIssue],
issues: List[ValidationIssue],
) -> None:
"""Validate optional link provisioning settings.
Args:
settings: Link settings being validated.
location: Logical location of the settings object.
issues: Mutable list receiving validation failures.
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,
"delay_ms": settings.delay_ms,
"loss_percent": settings.loss_percent,
"jitter_ms": settings.jitter_ms,
}
for name, value in numeric_settings.items():
@@ -525,7 +628,7 @@ def _validate_link_names(
Args:
topology: Parsed topology being validated.
issues: Mutable list receiving validation failures.
issues: Mutable list receiving validation failures.
"""
first_index: Dict[str, int] = {}
@@ -559,10 +662,28 @@ def _validate_links(
) -> 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.
topology: Parsed topology being validated.
node_index: Node lookup information prepared by node validation.
issues: Mutable list receiving validation failures.
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)
@@ -578,6 +699,7 @@ def _validate_links(
)
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,
@@ -651,9 +773,11 @@ def _validate_links(
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
@@ -670,39 +794,58 @@ def _validate_links(
issues,
"links",
"node %r has %d automatic interface request%s but only %d "
"unused declared interface%s remain%s"
% (
"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"
% (
"; %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: Service,
service_location: str,
node: Node,
issues: List[ValidationIssue],
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.
node: Node on which the service runs.
issues: Mutable list receiving validation failures.
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
@@ -772,18 +915,34 @@ def _resolve_service_interface(
def _validate_dhcp_server(
service: Service,
service_location: str,
node: Node,
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.
node: Node on which the service runs.
interface: Resolved service interface, if available.
issues: Mutable list receiving validation failures.
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:
@@ -801,12 +960,12 @@ def _validate_dhcp_server(
)
return
start = _parse_ipv4_address(
start = _validate_ipv4_address(
service.settings.address_range.start,
service_location + ".settings.range.start",
issues,
)
end = _parse_ipv4_address(
end = _validate_ipv4_address(
service.settings.address_range.end,
service_location + ".settings.range.end",
issues,
@@ -821,7 +980,7 @@ def _validate_dhcp_server(
gateway: Optional[ipaddress.IPv4Address] = None
if service.settings.gateway is not None:
gateway = _parse_ipv4_address(
gateway = _validate_ipv4_address(
service.settings.gateway,
service_location + ".settings.gateway",
issues,
@@ -842,7 +1001,7 @@ def _validate_dhcp_server(
# The interface validator reports this as well.
return
parsed_interface = _parse_ipv4_interface(
parsed_interface = _validate_ipv4_interface(
interface.address,
service_location + ".interface",
issues,
@@ -962,18 +1121,45 @@ def _validate_services(
_validate_dhcp_server(
service,
location,
node,
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.
@@ -994,6 +1180,8 @@ def validate(topology: Topology) -> None:
raise TopologyValidationError(issues)
# =============================== Self-test ===============================
def _self_test() -> None:
"""Run smoke tests against the public validate() interface."""