Add topology inner representation and python assembly-like file generation(planer)

source/topology_loader.py

@@ -10,6 +10,8 @@ The loading stage has two public entry points:
         model by calling topology_parser.parse_topology().
 
 No Mininet code is imported or executed in this module.
+
+Author: Christos Choutouridis <cchoutou@ece.auth.gr>
 """
 
 import json

source/topology_plan.py

@@ -0,0 +1,812 @@
+"""Lower a validated topology into a fully resolved execution plan.
+
+This module is the planning stage of the topology pipeline.
+
+The parser produces a typed description of the user's intent and the validator
+proves that the description is semantically coherent.  The planner then makes
+all implementation choices that must be resolved before executable Mininet
+Python can be generated.
+
+In particular, the planner resolves logical and automatic interface references
+into concrete Linux/Mininet interface names.  It does not import Mininet,
+execute commands, modify the network, or render Python source.
+
+Author: Christos Choutouridis <cchoutou@ece.auth.gr>
+"""
+
+from dataclasses import dataclass
+from typing import Dict, List, Optional, Set, Tuple, Union
+
+from topology_parser import (
+    AddressingMode,
+    DHCPServerSettings,
+    Interface,
+    LinkInterfaceSelector,
+    Node,
+    NodeType,
+    RouterSettings,
+    Service,
+    ServiceInterfaceSelector,
+    ServiceType,
+    Topology,
+)
+
+
+class TopologyPlanError(ValueError):
+    """Raised when a validated topology cannot be lowered into a plan."""
+
+
+@dataclass(frozen=True)
+class PlannedNode:
+    """Describe one concrete node creation operation.
+
+    Args:
+        name: Logical node name used by Mininet.
+        type: Concrete semantic node type to create.
+    """
+
+    name: str
+    type: NodeType
+
+
+@dataclass(frozen=True)
+class PlannedLinkEndpoint:
+    """Describe one fully resolved endpoint of a planned link.
+
+    Args:
+        node: Logical node name that owns the endpoint.
+        logical_interface: Resolved logical interface identifier.  For a node
+            with a dynamic interface set, the planner creates a deterministic
+            synthetic identifier for an automatically allocated interface.
+        interface_name: Concrete Linux/Mininet interface name.
+    """
+
+    node: str
+    logical_interface: str
+    interface_name: str
+
+
+@dataclass(frozen=True)
+class PlannedLinkSettings:
+    """Contain resolved provisioning settings for one link.
+
+    Args:
+        bandwidth_mbps: Optional bandwidth limit in megabits per second.
+        delay_ms: Optional one-way delay in milliseconds.
+        loss_percent: Optional packet-loss percentage.
+        jitter_ms: Optional delay variation in milliseconds.
+    """
+
+    bandwidth_mbps: Optional[float] = None
+    delay_ms: Optional[float] = None
+    loss_percent: Optional[float] = None
+    jitter_ms: Optional[float] = None
+
+
+@dataclass(frozen=True)
+class PlannedLink:
+    """Describe one fully resolved Mininet link creation operation.
+
+    Args:
+        endpoints: Ordered pair of fully resolved link endpoints.
+        name: Optional logical link name preserved for diagnostics.
+        settings: Concrete link provisioning settings.
+    """
+
+    endpoints: Tuple[PlannedLinkEndpoint, PlannedLinkEndpoint]
+    name: Optional[str]
+    settings: PlannedLinkSettings
+
+
+@dataclass(frozen=True)
+class PlannedInterface:
+    """Describe post-link configuration for one concrete interface.
+
+    Args:
+        node: Logical node name that owns the interface.
+        logical_interface: Resolved logical interface identifier.
+        interface_name: Concrete Linux/Mininet interface name.
+        addressing: Layer-3 addressing strategy.
+        address: Optional static CIDR address.
+        gateway: Optional static default gateway.
+        mac: Optional explicit MAC address.
+        mtu: Optional MTU value.
+    """
+
+    node: str
+    logical_interface: str
+    interface_name: str
+    addressing: AddressingMode
+    address: Optional[str]
+    gateway: Optional[str]
+    mac: Optional[str]
+    mtu: Optional[int]
+
+
+@dataclass(frozen=True)
+class PlannedSysctl:
+    """Describe one namespace-scoped sysctl operation.
+
+    Args:
+        node: Logical node on which the sysctl is applied.
+        key: Fully resolved sysctl key.
+        value: Integer value written to the key.
+    """
+
+    node: str
+    key: str
+    value: int
+
+
+@dataclass(frozen=True)
+class PlannedDHCPServerSettings:
+    """Contain fully resolved DHCP server settings.
+
+    Args:
+        range_start: First address offered to DHCP clients.
+        range_end: Last address offered to DHCP clients.
+        gateway: Optional default gateway advertised to DHCP clients.
+    """
+
+    range_start: str
+    range_end: str
+    gateway: Optional[str]
+
+
+PlannedServiceSettings = PlannedDHCPServerSettings
+
+
+@dataclass(frozen=True)
+class PlannedService:
+    """Describe one fully resolved service startup operation.
+
+    Args:
+        type: Service implementation type.
+        node: Logical node on which the service runs.
+        logical_interface: Resolved logical interface identifier.
+        interface_name: Concrete Linux/Mininet interface name to bind.
+        settings: Service-type-specific resolved settings.
+    """
+
+    type: ServiceType
+    node: str
+    logical_interface: str
+    interface_name: str
+    settings: PlannedServiceSettings
+
+
+@dataclass(frozen=True)
+class ExecutionPlan:
+    """Contain the complete, fully resolved execution plan.
+
+    Args:
+        nodes: Node creation operations in declaration order.
+        links: Link creation operations in declaration order.
+        interfaces: Concrete interface configuration operations.
+        node_settings: Concrete namespace-level configuration operations.
+        services: Service startup operations in declaration order.
+
+    Notes:
+        The plan contains no AUTO or ALL selectors.  All such decisions are
+        resolved before an ExecutionPlan is returned.
+    """
+
+    nodes: Tuple[PlannedNode, ...]
+    links: Tuple[PlannedLink, ...]
+    interfaces: Tuple[PlannedInterface, ...]
+    node_settings: Tuple[PlannedSysctl, ...]
+    services: Tuple[PlannedService, ...]
+
+
+@dataclass
+class _PlanningContext:
+    """Hold mutable state used while lowering a topology.
+
+    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: Dict[str, Node]
+    reserved_interfaces: Dict[str, Set[str]]
+    free_interfaces: Dict[str, List[str]]
+    dynamic_interface_counters: Dict[str, int]
+    physical_interface_counters: Dict[str, int]
+    interface_names: Dict[Tuple[str, str], str]
+    interface_models: Dict[Tuple[str, str], Interface]
+
+
+def _build_context(topology: Topology) -> _PlanningContext:
+    """Create planning state and reserve all explicit link interfaces.
+
+    Args:
+        topology: Validated topology to lower.
+
+    Returns:
+        Initialized mutable planning context.
+    """
+
+    nodes = {node.name: node for node in topology.nodes}
+    reserved: Dict[str, Set[str]] = {node.name: set() for node in topology.nodes}
+
+    for link in topology.links:
+        for endpoint in link.endpoints:
+            if isinstance(endpoint.interface, str):
+                reserved[endpoint.node].add(endpoint.interface)
+
+    free_interfaces: Dict[str, List[str]] = {}
+    for node in topology.nodes:
+        if node.interfaces is None:
+            continue
+
+        free_interfaces[node.name] = [
+            name
+            for name in node.interfaces.keys()
+            if name not in reserved[node.name]
+        ]
+
+    return _PlanningContext(
+        nodes=nodes,
+        reserved_interfaces=reserved,
+        free_interfaces=free_interfaces,
+        dynamic_interface_counters={node.name: 0 for node in topology.nodes},
+        physical_interface_counters={node.name: 0 for node in topology.nodes},
+        interface_names={},
+        interface_models={},
+    )
+
+
+def _plan_nodes(topology: Topology) -> Tuple[PlannedNode, ...]:
+    """Lower topology nodes into concrete node creation operations.
+
+    Args:
+        topology: Validated topology to lower.
+
+    Returns:
+        Planned nodes in source declaration order.
+    """
+
+    return tuple(
+        PlannedNode(name=node.name, type=node.type)
+        for node in topology.nodes
+    )
+
+
+def _allocate_logical_interface(
+    node: Node,
+    selector: Union[str, LinkInterfaceSelector],
+    context: _PlanningContext,
+) -> Tuple[str, Interface]:
+    """Resolve one link endpoint to a concrete logical interface.
+
+    Args:
+        node: Node that owns the endpoint.
+        selector: Explicit interface tag or AUTO selector.
+        context: Mutable planning context.
+
+    Returns:
+        Pair containing the resolved logical interface name and its interface
+        configuration model.
+
+    Raises:
+        TopologyPlanError: If the topology violates a planner precondition.
+    """
+
+    if isinstance(selector, str):
+        if node.interfaces is None or selector not in node.interfaces:
+            raise TopologyPlanError(
+                "explicit interface %r cannot be resolved on node %r"
+                % (selector, node.name)
+            )
+
+        return selector, node.interfaces[selector]
+
+    if selector is not LinkInterfaceSelector.AUTO:
+        raise TopologyPlanError(
+            "unsupported link interface selector %r" % selector
+        )
+
+    if node.interfaces is not None:
+        free = context.free_interfaces[node.name]
+        if not free:
+            raise TopologyPlanError(
+                "node %r has no declared interface available for AUTO"
+                % node.name
+            )
+
+        logical_name = free.pop(0)
+        return logical_name, node.interfaces[logical_name]
+
+    counter = context.dynamic_interface_counters[node.name]
+    context.dynamic_interface_counters[node.name] = counter + 1
+
+    logical_name = "auto%d" % counter
+    return logical_name, Interface(name=logical_name)
+
+
+def _allocate_physical_interface(
+    node_name: str,
+    logical_name: str,
+    interface: Interface,
+    context: _PlanningContext,
+) -> str:
+    """Assign one deterministic concrete interface name.
+
+    Args:
+        node_name: Logical node that owns the interface.
+        logical_name: Resolved logical interface identifier.
+        interface: Interface configuration model.
+        context: Mutable planning context.
+
+    Returns:
+        Concrete Linux/Mininet interface name.
+
+    Raises:
+        TopologyPlanError: If the same logical interface is allocated twice.
+    """
+
+    key = (node_name, logical_name)
+    if key in context.interface_names:
+        raise TopologyPlanError(
+            "interface %s:%s was allocated more than once"
+            % (node_name, logical_name)
+        )
+
+    index = context.physical_interface_counters[node_name]
+    context.physical_interface_counters[node_name] = index + 1
+
+    interface_name = "%s-eth%d" % (node_name, index)
+    context.interface_names[key] = interface_name
+    context.interface_models[key] = interface
+
+    return interface_name
+
+
+def _plan_links(
+    topology: Topology,
+    context: _PlanningContext,
+) -> Tuple[PlannedLink, ...]:
+    """Resolve all link endpoints and concrete interface names.
+
+    Args:
+        topology: Validated topology to lower.
+        context: Mutable planning context.
+
+    Returns:
+        Planned links in source declaration order.
+    """
+
+    planned_links: List[PlannedLink] = []
+
+    for link in topology.links:
+        planned_endpoints: List[PlannedLinkEndpoint] = []
+
+        for endpoint in link.endpoints:
+            node = context.nodes[endpoint.node]
+            logical_name, interface = _allocate_logical_interface(
+                node,
+                endpoint.interface,
+                context,
+            )
+            interface_name = _allocate_physical_interface(
+                node.name,
+                logical_name,
+                interface,
+                context,
+            )
+
+            planned_endpoints.append(
+                PlannedLinkEndpoint(
+                    node=node.name,
+                    logical_interface=logical_name,
+                    interface_name=interface_name,
+                )
+            )
+
+        planned_links.append(
+            PlannedLink(
+                endpoints=(planned_endpoints[0], planned_endpoints[1]),
+                name=link.name,
+                settings=PlannedLinkSettings(
+                    bandwidth_mbps=link.settings.bandwidth_mbps,
+                    delay_ms=link.settings.delay_ms,
+                    loss_percent=link.settings.loss_percent,
+                    jitter_ms=link.settings.jitter_ms,
+                ),
+            )
+        )
+
+    return tuple(planned_links)
+
+
+def _interface_requires_materialization(interface: Interface) -> bool:
+    """Return whether an unused declared interface carries configuration.
+
+    Args:
+        interface: Declared interface to inspect.
+
+    Returns:
+        True when silently leaving the interface unmaterialized would discard
+        requested configuration.
+    """
+
+    return (
+        interface.addressing is not AddressingMode.NONE
+        or interface.address is not None
+        or interface.gateway is not None
+        or interface.mac is not None
+        or interface.mtu is not None
+    )
+
+
+def _plan_interfaces(
+    topology: Topology,
+    context: _PlanningContext,
+) -> Tuple[PlannedInterface, ...]:
+    """Create interface-configuration operations for materialized interfaces.
+
+    Args:
+        topology: Validated topology to lower.
+        context: Planning context containing resolved interface mappings.
+
+    Returns:
+        Planned interface configuration operations.
+
+    Raises:
+        TopologyPlanError: If a configured declared interface is not attached to
+            any link and therefore has no concrete Mininet interface to
+            configure.
+    """
+
+    for node in topology.nodes:
+        if node.interfaces is None:
+            continue
+
+        for logical_name, interface in node.interfaces.items():
+            key = (node.name, logical_name)
+            if (
+                key not in context.interface_names
+                and _interface_requires_materialization(interface)
+            ):
+                raise TopologyPlanError(
+                    "configured interface %s:%s is not attached to any link"
+                    % (node.name, logical_name)
+                )
+
+    planned: List[PlannedInterface] = []
+
+    # The mapping insertion order is the exact allocation order from
+    # _plan_links(), which is deterministic on supported Python versions.
+    for key, interface_name in context.interface_names.items():
+        node_name, logical_name = key
+        interface = context.interface_models[key]
+
+        planned.append(
+            PlannedInterface(
+                node=node_name,
+                logical_interface=logical_name,
+                interface_name=interface_name,
+                addressing=interface.addressing,
+                address=interface.address,
+                gateway=interface.gateway,
+                mac=interface.mac,
+                mtu=interface.mtu,
+            )
+        )
+
+    return tuple(planned)
+
+
+def _plan_node_settings(
+    topology: Topology,
+    context: _PlanningContext,
+) -> Tuple[PlannedSysctl, ...]:
+    """Lower node settings into concrete namespace sysctl operations.
+
+    Args:
+        topology: Validated topology to lower.
+        context: Planning context containing concrete interface names.
+
+    Returns:
+        Planned sysctl operations in deterministic order.
+    """
+
+    planned: List[PlannedSysctl] = []
+
+    for node in topology.nodes:
+        if node.type is not NodeType.ROUTER:
+            continue
+
+        if not isinstance(node.settings, RouterSettings):
+            raise TopologyPlanError(
+                "router %r does not carry RouterSettings" % node.name
+            )
+
+        if node.settings.ip_forward is not None:
+            planned.append(
+                PlannedSysctl(
+                    node=node.name,
+                    key="net.ipv4.ip_forward",
+                    value=1 if node.settings.ip_forward else 0,
+                )
+            )
+
+        if node.settings.rp_filter is not None:
+            value = 1 if node.settings.rp_filter else 0
+
+            planned.append(
+                PlannedSysctl(
+                    node=node.name,
+                    key="net.ipv4.conf.all.rp_filter",
+                    value=value,
+                )
+            )
+            planned.append(
+                PlannedSysctl(
+                    node=node.name,
+                    key="net.ipv4.conf.default.rp_filter",
+                    value=value,
+                )
+            )
+
+            for (owner, _), interface_name in context.interface_names.items():
+                if owner != node.name:
+                    continue
+
+                planned.append(
+                    PlannedSysctl(
+                        node=node.name,
+                        key="net.ipv4.conf.%s.rp_filter" % interface_name,
+                        value=value,
+                    )
+                )
+
+    return tuple(planned)
+
+
+def _resolve_service_logical_interface(
+    service: Service,
+    node: Node,
+) -> str:
+    """Resolve one validated service interface selector.
+
+    Args:
+        service: Service whose interface binding is resolved.
+        node: Node on which the service runs.
+
+    Returns:
+        Concrete logical interface tag.
+
+    Raises:
+        TopologyPlanError: If the selector cannot be lowered uniquely.
+    """
+
+    if isinstance(service.interface, str):
+        return service.interface
+
+    if service.interface is ServiceInterfaceSelector.AUTO:
+        if node.interfaces is None:
+            raise TopologyPlanError(
+                "automatic service binding on node %r requires declared "
+                "interfaces" % node.name
+            )
+
+        candidates = [
+            name
+            for name, interface in node.interfaces.items()
+            if interface.addressing is AddressingMode.STATIC
+        ]
+
+        if len(candidates) != 1:
+            raise TopologyPlanError(
+                "automatic service binding on node %r is not uniquely "
+                "resolvable" % node.name
+            )
+
+        return candidates[0]
+
+    if service.interface is ServiceInterfaceSelector.ALL:
+        raise TopologyPlanError(
+            "service type %r cannot be lowered from the ALL selector"
+            % service.type.value
+        )
+
+    raise TopologyPlanError(
+        "unsupported service interface selector %r" % service.interface
+    )
+
+
+def _plan_services(
+    topology: Topology,
+    context: _PlanningContext,
+) -> Tuple[PlannedService, ...]:
+    """Resolve service bindings and lower service-specific settings.
+
+    Args:
+        topology: Validated topology to lower.
+        context: Planning context containing concrete interface names.
+
+    Returns:
+        Planned service startup operations in declaration order.
+
+    Raises:
+        TopologyPlanError: If a service cannot be mapped to a materialized
+            interface or has unsupported settings.
+    """
+
+    planned: List[PlannedService] = []
+
+    for service in topology.services:
+        node = context.nodes[service.node]
+        logical_name = _resolve_service_logical_interface(service, node)
+        key = (node.name, logical_name)
+
+        if key not in context.interface_names:
+            raise TopologyPlanError(
+                "service %r binds to interface %s:%s, but that interface is "
+                "not attached to any link"
+                % (service.type.value, node.name, logical_name)
+            )
+
+        if service.type is ServiceType.DHCP_SERVER:
+            if not isinstance(service.settings, DHCPServerSettings):
+                raise TopologyPlanError(
+                    "DHCP service on %r has incompatible settings" % node.name
+                )
+
+            settings = PlannedDHCPServerSettings(
+                range_start=service.settings.address_range.start,
+                range_end=service.settings.address_range.end,
+                gateway=service.settings.gateway,
+            )
+        else:
+            raise TopologyPlanError(
+                "unsupported service type %r" % service.type.value
+            )
+
+        planned.append(
+            PlannedService(
+                type=service.type,
+                node=node.name,
+                logical_interface=logical_name,
+                interface_name=context.interface_names[key],
+                settings=settings,
+            )
+        )
+
+    return tuple(planned)
+
+
+def plan(topology: Topology) -> ExecutionPlan:
+    """Lower a validated topology into a fully resolved execution plan.
+
+    Args:
+        topology: Topology that has already passed semantic validation.
+
+    Returns:
+        Fully resolved execution plan suitable for source-code rendering.
+
+    Raises:
+        TopologyPlanError: If a planner precondition is violated or if a
+            validated construct cannot be lowered by the current backend.
+
+    Notes:
+        This function deliberately does not call validate().  Validation and
+        planning remain separate pipeline stages, and callers are expected to
+        validate the topology before invoking the planner.
+    """
+
+    context = _build_context(topology)
+
+    nodes = _plan_nodes(topology)
+    links = _plan_links(topology, context)
+    interfaces = _plan_interfaces(topology, context)
+    node_settings = _plan_node_settings(topology, context)
+    services = _plan_services(topology, context)
+
+    return ExecutionPlan(
+        nodes=nodes,
+        links=links,
+        interfaces=interfaces,
+        node_settings=node_settings,
+        services=services,
+    )
+
+
+def _self_test() -> None:
+    """Run a smoke test against the public plan() interface."""
+
+    from topology_parser import parse_topology
+    from topology_validator import validate
+
+    raw = {
+        "nodes": [
+            {
+                "name": "h1",
+                "type": "host",
+                "interfaces": {
+                    "net0": {
+                        "addressing": "dhcp",
+                    }
+                },
+            },
+            {
+                "name": "s1",
+                "type": "switch",
+                "interfaces": {
+                    "client": {},
+                    "uplink": {},
+                },
+            },
+            {
+                "name": "r0",
+                "type": "router",
+                "interfaces": {
+                    "lan": {
+                        "addressing": "static",
+                        "address": "192.168.1.1/24",
+                    }
+                },
+                "settings": {
+                    "ip_forward": True,
+                    "rp_filter": False,
+                },
+            },
+        ],
+        "links": [
+            {
+                "endpoints": [
+                    {"node": "h1", "interface": "net0"},
+                    {"node": "s1", "interface": "auto"},
+                ]
+            },
+            {
+                "endpoints": [
+                    {"node": "r0", "interface": "lan"},
+                    {"node": "s1", "interface": "uplink"},
+                ]
+            },
+        ],
+        "services": [
+            {
+                "type": "dhcp-server",
+                "node": "r0",
+                "interface": "lan",
+                "settings": {
+                    "range": {
+                        "start": "192.168.1.100",
+                        "end": "192.168.1.200",
+                    },
+                    "gateway": "192.168.1.1",
+                },
+            }
+        ],
+    }
+
+    topology = parse_topology(raw)
+    validate(topology)
+    execution_plan = plan(topology)
+
+    assert len(execution_plan.nodes) == 3
+    assert len(execution_plan.links) == 2
+    assert execution_plan.links[0].endpoints[1].logical_interface == "client"
+    assert execution_plan.links[0].endpoints[0].interface_name == "h1-eth0"
+    assert execution_plan.links[1].endpoints[0].interface_name == "r0-eth0"
+    assert execution_plan.services[0].interface_name == "r0-eth0"
+    assert any(
+        setting.key == "net.ipv4.ip_forward" and setting.value == 1
+        for setting in execution_plan.node_settings
+    )
+
+
+if __name__ == "__main__":
+    _self_test()
+    print("topology_plan: self-test passed")

source/topology_renderer.py

@@ -0,0 +1,482 @@
+"""Render a resolved execution plan as executable Mininet Python source.
+
+The renderer is intentionally a serialization stage.  It does not perform
+semantic validation, allocate AUTO interfaces, import Mininet itself, or make
+networking decisions.  All topology decisions must already be represented in
+the ExecutionPlan produced by topology_plan.plan().
+
+The returned Python source is both an inspectable intermediate artifact and the
+canonical executable form of the planned topology.
+
+Author: Christos Choutouridis <cchoutou@ece.auth.gr>
+"""
+
+import ast
+from typing import List
+
+from topology_parser import AddressingMode, NodeType, ServiceType
+from topology_plan import (
+    ExecutionPlan,
+    PlannedDHCPServerSettings,
+    PlannedInterface,
+    PlannedLink,
+    PlannedService,
+)
+
+
+def _python_string(value: str) -> str:
+    """Return a safe Python string literal.
+
+    Args:
+        value: String to serialize into generated Python source.
+
+    Returns:
+        Python source literal representing the string exactly.
+    """
+
+    return repr(value)
+
+
+def _render_link_call(link: PlannedLink) -> List[str]:
+    """Render one Mininet addLink() call.
+
+    Args:
+        link: Fully resolved link to render.
+
+    Returns:
+        Source lines implementing the link creation operation.
+    """
+
+    first, second = link.endpoints
+    settings = link.settings
+
+    arguments = [
+        "nodes[%s]" % _python_string(first.node),
+        "nodes[%s]" % _python_string(second.node),
+        "intfName1=%s" % _python_string(first.interface_name),
+        "intfName2=%s" % _python_string(second.interface_name),
+    ]
+
+    has_tc_settings = any(
+        value is not None
+        for value in (
+            settings.bandwidth_mbps,
+            settings.delay_ms,
+            settings.loss_percent,
+            settings.jitter_ms,
+        )
+    )
+
+    if has_tc_settings:
+        arguments.append("cls=TCLink")
+
+    if settings.bandwidth_mbps is not None:
+        arguments.append("bw=%r" % settings.bandwidth_mbps)
+
+    if settings.delay_ms is not None:
+        arguments.append("delay=%s" % _python_string("%gms" % settings.delay_ms))
+    elif settings.jitter_ms is not None:
+        # Linux netem expresses jitter as variation around a base delay.  A
+        # zero base delay preserves the declarative meaning when only jitter
+        # was provided in the topology.
+        arguments.append("delay=%s" % _python_string("0ms"))
+
+    if settings.loss_percent is not None:
+        arguments.append("loss=%r" % settings.loss_percent)
+
+    if settings.jitter_ms is not None:
+        arguments.append("jitter=%s" % _python_string("%gms" % settings.jitter_ms))
+
+    lines = ["        net.addLink("]
+    for argument in arguments:
+        lines.append("            %s," % argument)
+    lines.append("        )")
+
+    return lines
+
+
+def _render_interface_configuration(interface: PlannedInterface) -> List[str]:
+    """Render configuration commands for one concrete interface.
+
+    Args:
+        interface: Fully resolved interface configuration operation.
+
+    Returns:
+        Generated Python source lines for the interface.
+    """
+
+    node = "nodes[%s]" % _python_string(interface.node)
+    name = _python_string(interface.interface_name)
+    lines: List[str] = []
+
+    if interface.mac is not None:
+        lines.append(
+            "        _node_cmd(%s, 'ip', 'link', 'set', 'dev', %s, "
+            "'address', %s)"
+            % (node, name, _python_string(interface.mac))
+        )
+
+    if interface.mtu is not None:
+        lines.append(
+            "        _node_cmd(%s, 'ip', 'link', 'set', 'dev', %s, "
+            "'mtu', %s)"
+            % (node, name, _python_string(str(interface.mtu)))
+        )
+
+    if interface.addressing in {
+        AddressingMode.STATIC,
+        AddressingMode.DHCP,
+    }:
+        # addHost(ip=None) prevents Mininet from assigning its default address,
+        # and this flush makes the generated artifact explicit about the
+        # desired IPv4 state before static or manual-DHCP configuration.
+        lines.append(
+            "        _node_cmd(%s, 'ip', '-4', 'addr', 'flush', 'dev', %s)"
+            % (node, name)
+        )
+
+    if interface.addressing is AddressingMode.STATIC:
+        if interface.address is None:
+            raise ValueError(
+                "planned static interface %s:%s has no address"
+                % (interface.node, interface.logical_interface)
+            )
+
+        lines.append(
+            "        _node_cmd(%s, 'ip', 'addr', 'add', %s, 'dev', %s)"
+            % (
+                node,
+                _python_string(interface.address),
+                name,
+            )
+        )
+
+    if (
+        interface.addressing in {AddressingMode.STATIC, AddressingMode.DHCP}
+        or interface.mac is not None
+        or interface.mtu is not None
+    ):
+        lines.append(
+            "        _node_cmd(%s, 'ip', 'link', 'set', 'dev', %s, 'up')"
+            % (node, name)
+        )
+
+    return lines
+
+
+def _render_gateway_configuration(interface: PlannedInterface) -> List[str]:
+    """Render an optional static default route for one interface.
+
+    Args:
+        interface: Fully resolved interface configuration operation.
+
+    Returns:
+        Generated Python source lines, possibly empty.
+    """
+
+    if interface.gateway is None:
+        return []
+
+    node = "nodes[%s]" % _python_string(interface.node)
+
+    return [
+        "        _node_cmd(%s, 'ip', 'route', 'replace', 'default', 'via', "
+        "%s, 'dev', %s)"
+        % (
+            node,
+            _python_string(interface.gateway),
+            _python_string(interface.interface_name),
+        )
+    ]
+
+
+def _render_service_start(service: PlannedService, index: int) -> List[str]:
+    """Render startup code for one planned service.
+
+    Args:
+        service: Fully resolved service operation.
+        index: Stable service index used for temporary runtime files.
+
+    Returns:
+        Generated Python source lines that start and track the service.
+
+    Raises:
+        ValueError: If the plan contains an unsupported service type or
+            incompatible settings object.
+    """
+
+    if service.type is not ServiceType.DHCP_SERVER:
+        raise ValueError(
+            "unsupported planned service type %r" % service.type.value
+        )
+
+    if not isinstance(service.settings, PlannedDHCPServerSettings):
+        raise ValueError("DHCP service has incompatible planned settings")
+
+    node = "nodes[%s]" % _python_string(service.node)
+    lease_file = "/tmp/mininet-dnsmasq-%d.leases" % index
+    pid_file = "/tmp/mininet-dnsmasq-%d.pid" % index
+    log_file = "/tmp/mininet-dnsmasq-%d.log" % index
+
+    args = [
+        "dnsmasq",
+        "--no-daemon",
+        "--conf-file=",
+        "--port=0",
+        "--bind-interfaces",
+        "--dhcp-authoritative",
+        "--interface=%s" % service.interface_name,
+        "--dhcp-range=%s,%s,12h"
+        % (service.settings.range_start, service.settings.range_end),
+        "--dhcp-leasefile=%s" % lease_file,
+        "--pid-file=%s" % pid_file,
+    ]
+
+    if service.settings.gateway is not None:
+        args.append(
+            "--dhcp-option=3,%s" % service.settings.gateway
+        )
+
+    rendered_args = ", ".join(_python_string(arg) for arg in args)
+
+    return [
+        "        pid = _start_background(",
+        "            %s," % node,
+        "            [%s]," % rendered_args,
+        "            %s," % _python_string(log_file),
+        "        )",
+        "        service_processes.append((%s, pid))" % node,
+    ]
+
+
+def render_python(plan: ExecutionPlan) -> str:
+    """Render an execution plan as a complete executable Python program.
+
+    Args:
+        plan: Fully resolved execution plan returned by topology_plan.plan().
+
+    Returns:
+        Python source code that creates the network, applies configuration,
+        starts services, opens the Mininet CLI, and performs structured cleanup.
+
+    Raises:
+        ValueError: If the supplied plan contains an operation that this
+            renderer cannot serialize.
+    """
+
+    lines: List[str] = [
+        "#!/usr/bin/env python3",
+        '"""Generated Mininet topology.  Do not edit by hand."""',
+        "",
+        "import shlex",
+        "",
+        "from mininet.cli import CLI",
+        "from mininet.link import TCLink",
+        "from mininet.net import Mininet",
+        "",
+        "",
+        "def _shell_join(args):",
+        '    """Quote command arguments for execution inside a Mininet node."""',
+        "",
+        "    return ' '.join(shlex.quote(str(arg)) for arg in args)",
+        "",
+        "",
+        "def _node_cmd(node, *args):",
+        '    """Execute one safely quoted command inside a Mininet node."""',
+        "",
+        "    return node.cmd(_shell_join(args))",
+        "",
+        "",
+        "def _start_background(node, args, log_path):",
+        '    """Start one background process and return its shell PID."""',
+        "",
+        "    command = _shell_join(args)",
+        "    shell_command = (",
+        "        command",
+        "        + ' >'",
+        "        + shlex.quote(log_path)",
+        "        + ' 2>&1 & echo $!'",
+        "    )",
+        "    return node.cmd('sh -c ' + shlex.quote(shell_command)).strip()",
+        "",
+        "",
+        "def main():",
+        '    """Build, configure, expose, and clean up the planned network."""',
+        "",
+        "    net = Mininet(controller=None, build=False)",
+        "    nodes = {}",
+        "    service_processes = []",
+        "",
+        "    try:",
+        "        # Create nodes.",
+    ]
+
+    for node in plan.nodes:
+        if node.type in {NodeType.HOST, NodeType.ROUTER}:
+            lines.append(
+                "        nodes[%s] = net.addHost(%s, ip=None)"
+                % (
+                    _python_string(node.name),
+                    _python_string(node.name),
+                )
+            )
+        elif node.type is NodeType.SWITCH:
+            lines.append(
+                "        nodes[%s] = net.addSwitch(%s)"
+                % (
+                    _python_string(node.name),
+                    _python_string(node.name),
+                )
+            )
+        else:
+            raise ValueError("unsupported planned node type %r" % node.type)
+
+    lines.extend(["", "        # Create links."])
+    for link in plan.links:
+        lines.extend(_render_link_call(link))
+
+    lines.extend(
+        [
+            "",
+            "        # Materialize the Mininet topology.",
+            "        net.build()",
+            "        net.start()",
+            "",
+            "        # Configure concrete interfaces.",
+        ]
+    )
+
+    for interface in plan.interfaces:
+        lines.extend(_render_interface_configuration(interface))
+
+    lines.extend(["", "        # Configure static default routes."])
+    for interface in plan.interfaces:
+        lines.extend(_render_gateway_configuration(interface))
+
+    lines.extend(["", "        # Apply node-level networking settings."])
+    for setting in plan.node_settings:
+        lines.append(
+            "        _node_cmd(nodes[%s], 'sysctl', '-w', %s)"
+            % (
+                _python_string(setting.node),
+                _python_string("%s=%d" % (setting.key, setting.value)),
+            )
+        )
+
+    lines.extend(["", "        # Start planned services."])
+    for index, service in enumerate(plan.services):
+        lines.extend(_render_service_start(service, index))
+
+    lines.extend(
+        [
+            "",
+            "        # DHCP clients are intentionally not started here.",
+            "        # The assignment captures DHCP before dhclient is run manually.",
+            "        CLI(net)",
+            "",
+            "    finally:",
+            "        # Stop only the service processes started by this script.",
+            "        for node, pid in reversed(service_processes):",
+            "            if pid:",
+            "                _node_cmd(node, 'kill', pid)",
+            "",
+            "        net.stop()",
+            "",
+            "",
+            "if __name__ == '__main__':",
+            "    main()",
+            "",
+        ]
+    )
+
+    return "\n".join(lines)
+
+
+def _self_test() -> None:
+    """Run a smoke test against the public render_python() interface."""
+
+    from topology_parser import parse_topology
+    from topology_plan import plan
+    from topology_validator import validate
+
+    raw = {
+        "nodes": [
+            {
+                "name": "h1",
+                "type": "host",
+                "interfaces": {
+                    "net0": {
+                        "addressing": "dhcp",
+                    }
+                },
+            },
+            {
+                "name": "s1",
+                "type": "switch",
+            },
+            {
+                "name": "r0",
+                "type": "router",
+                "interfaces": {
+                    "lan": {
+                        "addressing": "static",
+                        "address": "192.168.1.1/24",
+                    }
+                },
+                "settings": {
+                    "ip_forward": True,
+                    "rp_filter": False,
+                },
+            },
+        ],
+        "links": [
+            {
+                "endpoints": [
+                    {"node": "h1", "interface": "net0"},
+                    {"node": "s1", "interface": "auto"},
+                ]
+            },
+            {
+                "endpoints": [
+                    {"node": "r0", "interface": "lan"},
+                    {"node": "s1", "interface": "auto"},
+                ],
+                "settings": {
+                    "delay_ms": 1.5,
+                },
+            },
+        ],
+        "services": [
+            {
+                "type": "dhcp-server",
+                "node": "r0",
+                "interface": "lan",
+                "settings": {
+                    "range": {
+                        "start": "192.168.1.100",
+                        "end": "192.168.1.200",
+                    },
+                    "gateway": "192.168.1.1",
+                },
+            }
+        ],
+    }
+
+    topology = parse_topology(raw)
+    validate(topology)
+    source = render_python(plan(topology))
+
+    ast.parse(source)
+
+    assert "net.addHost('h1', ip=None)" in source
+    assert "net.addSwitch('s1')" in source
+    assert "net.addLink(" in source
+    assert "cls=TCLink" in source
+    assert "dnsmasq" in source
+    assert "CLI(net)" in source
+    assert "dhclient" in source
+
+
+if __name__ == "__main__":
+    _self_test()
+    print("topology_renderer: self-test passed")