Create iptrunk

A creation workflow that deploys a new IP trunk service.

This the workflow that brings the subscription from INACTIVE to PROVISIONING. The deployment of a new IP trunk consist in the following steps:

• Fill the form with the necessary fields:
  • SID
  • Type
  • Speed
  • Nodes
  • LAG interfaces with description
  • LAG members with description
• WFO will query IPAM to retrieve the IPv4/IPv6 Networks necessary for the trunk. The container to use is specified in oss-params.json
• The configuration necessary to deploy the LAG is generated and applied to the destination nodes using the Ansible playbook iptrunks.yaml This is done first in a dry mode (without committing) and then in a real mode committing the configuration. The commit message has the subscription_id and the process_id. Included in this, is the configuration necessary to enable LLDP on the physical interfaces.
• Once the LAG interface is deployed, another Ansible playbook is called to verify that IP traffic can actually flow over the trunk ( iptrunk_checks.yaml)
• Once the check is passed, the ISIS configuration will take place using the same iptrunks.yaml. Also in this case first there is a dry run and then a commit.
• After this step the ISIS adjacency gets checked using iptrunks_checks.yaml

The trunk is deployed with an initial ISIS metric of 90.000 to prevent traffic to pass.

initial_input_form_generator(product_name)

Gather input from the user in three steps. General information, and information on both sides of the trunk.

Source code in gso/workflows/iptrunk/create_iptrunk.py

def initial_input_form_generator(product_name: str) -> FormGenerator:
    """Gather input from the user in three steps. General information, and information on both sides of the trunk."""
    #  Add both provisioning and active routers, since trunks are required for promoting a router to active.
    active_and_provisioning_routers = subscriptions.get_active_router_subscriptions(
        includes=["subscription_id", "description"]
    ) + subscriptions.get_provisioning_router_subscriptions(includes=["subscription_id", "description"])
    routers = {str(router["subscription_id"]): router["description"] for router in active_and_provisioning_routers}

    class CreateIptrunkForm(FormPage):
        model_config = ConfigDict(title=product_name)

        tt_number: TTNumber
        partner: ReadOnlyField("GEANT", default_type=str)  # type: ignore[valid-type]
        iptrunk_description: str | None = None
        iptrunk_type: IptrunkType
        iptrunk_speed: PhysicalPortCapacity
        iptrunk_number_of_members: int

    initial_user_input = yield CreateIptrunkForm
    recommended_minimum_links = calculate_recommended_minimum_links(
        initial_user_input.iptrunk_number_of_members, initial_user_input.iptrunk_speed
    )

    class VerifyMinimumLinksForm(FormPage):
        info_label: Label = f"This is the calculated minimum-links for this LAG: {recommended_minimum_links}"
        iptrunk_minimum_links: int = recommended_minimum_links
        info_label2: Label = "Please confirm or modify."

    verify_minimum_links = yield VerifyMinimumLinksForm
    router_enum_a = Choice("Select a router", zip(routers.keys(), routers.items(), strict=True))  # type: ignore[arg-type]

    class SelectRouterSideA(FormPage):
        model_config = ConfigDict(title="Select a router for side A of the trunk.")

        side_a_node_id: NetboxEnabledRouter[router_enum_a]  # type: ignore[valid-type]

    user_input_router_side_a = yield SelectRouterSideA
    router_a = user_input_router_side_a.side_a_node_id.name
    router_a_fqdn = Router.from_subscription(router_a).router.router_fqdn

    juniper_ae_members = Annotated[
        LAGMemberList[JuniperLAGMember],
        Len(
            min_length=initial_user_input.iptrunk_number_of_members,
            max_length=initial_user_input.iptrunk_number_of_members,
        ),
    ]

    if get_router_vendor(router_a) == Vendor.NOKIA:

        class NokiaLAGMemberA(LAGMember):
            interface_name: available_interfaces_choices(  # type: ignore[valid-type]
                router_a,
                initial_user_input.iptrunk_speed,
            )

        ae_members_side_a_type = Annotated[
            LAGMemberList[NokiaLAGMemberA],
            Len(
                min_length=initial_user_input.iptrunk_number_of_members,
                max_length=initial_user_input.iptrunk_number_of_members,
            ),
        ]
    else:
        ae_members_side_a_type = juniper_ae_members  # type: ignore[assignment, misc]

    class CreateIptrunkSideAForm(FormPage):
        model_config = ConfigDict(title=f"Provide subscription details for side A of the trunk. ({router_a_fqdn})")

        side_a_ae_iface: available_lags_choices(router_a) or str  # type: ignore[valid-type]
        side_a_ae_members: ae_members_side_a_type

    user_input_side_a = yield CreateIptrunkSideAForm
    # Remove the selected router for side A, to prevent any loops
    routers.pop(str(router_a))
    router_enum_b = Choice("Select a router", zip(routers.keys(), routers.items(), strict=True))  # type: ignore[arg-type]

    class SelectRouterSideB(FormPage):
        model_config = ConfigDict(title="Select a router for side B of the trunk.")

        side_b_node_id: NetboxEnabledRouter[router_enum_b]  # type: ignore[valid-type]

    user_input_router_side_b = yield SelectRouterSideB
    router_b = user_input_router_side_b.side_b_node_id.name
    router_b_fqdn = Router.from_subscription(router_b).router.router_fqdn

    if get_router_vendor(router_b) == Vendor.NOKIA:

        class NokiaLAGMemberB(LAGMember):
            interface_name: available_interfaces_choices(  # type: ignore[valid-type]
                router_b,
                initial_user_input.iptrunk_speed,
            )

        ae_members_side_b = Annotated[
            LAGMemberList[NokiaLAGMemberB],
            Len(
                min_length=len(user_input_side_a.side_a_ae_members), max_length=len(user_input_side_a.side_a_ae_members)
            ),
        ]
    else:
        ae_members_side_b = juniper_ae_members  # type: ignore[assignment, misc]

    class CreateIptrunkSideBForm(FormPage):
        model_config = ConfigDict(title=f"Provide subscription details for side B of the trunk. ({router_b_fqdn})")

        side_b_ae_iface: available_lags_choices(router_b) or str  # type: ignore[valid-type]
        side_b_ae_members: ae_members_side_b

    user_input_side_b = yield CreateIptrunkSideBForm
    input_forms_data = (
        initial_user_input.model_dump()
        | verify_minimum_links.model_dump()
        | user_input_router_side_a.model_dump()
        | user_input_side_a.model_dump()
        | user_input_router_side_b.model_dump()
        | user_input_side_b.model_dump()
    )
    summary_form_data = input_forms_data | {"side_a_node": router_a_fqdn, "side_b_node": router_b_fqdn}
    summary_fields = [
        "iptrunk_type",
        "iptrunk_speed",
        "iptrunk_description",
        "iptrunk_minimum_links",
        "side_a_node",
        "side_a_ae_iface",
        "side_a_ae_members",
        "side_b_node",
        "side_b_ae_iface",
        "side_b_ae_members",
    ]
    yield from create_summary_form(summary_form_data, product_name, summary_fields)

    return input_forms_data

create_subscription(product, partner)

Create a new subscription object in the database.

Source code in gso/workflows/iptrunk/create_iptrunk.py

@step("Create subscription")
def create_subscription(product: UUIDstr, partner: str) -> State:
    """Create a new subscription object in the database."""
    subscription = IptrunkInactive.from_product_id(product, get_partner_by_name(partner).partner_id)

    return {
        "subscription": subscription,
        "subscription_id": subscription.subscription_id,
    }

get_info_from_ipam(subscription)

Allocate IP resources in IPAM.

Source code in gso/workflows/iptrunk/create_iptrunk.py

@step("Get information from IPAM")
def get_info_from_ipam(subscription: IptrunkInactive) -> State:
    """Allocate IP resources in IPAM."""
    new_ipv4_network = infoblox.allocate_v4_network(
        "TRUNK",
        subscription.iptrunk.iptrunk_description,
    )
    new_ipv6_network = infoblox.allocate_v6_network(
        "TRUNK",
        subscription.iptrunk.iptrunk_description,
    )
    subscription.iptrunk.iptrunk_ipv4_network = new_ipv4_network
    subscription.iptrunk.iptrunk_ipv6_network = new_ipv6_network

    return {
        "subscription": subscription,
        "new_ipv4_network": str(new_ipv4_network),
        "new_ipv6_network": str(new_ipv6_network),
    }

check_existing_trunk_allocations(subscription)

Check if there already is a trunk with the same network resources assigned to it.

Source code in gso/workflows/iptrunk/create_iptrunk.py

@step("Check if assigned networks are already taken by other trunk subscription")
def check_existing_trunk_allocations(subscription: IptrunkInactive) -> None:
    """Check if there already is a trunk with the same network resources assigned to it."""
    if not subscription.iptrunk.iptrunk_ipv4_network or not subscription.iptrunk.iptrunk_ipv6_network:
        msg = "Missing IP resources in subscription object."
        raise ProcessFailureError(
            msg, details=[subscription.iptrunk.iptrunk_ipv4_network, subscription.iptrunk.iptrunk_ipv6_network]
        )

    all_trunks = [
        Iptrunk.from_subscription(trunk["subscription_id"])
        for trunk in get_non_terminated_iptrunk_subscriptions()
        if trunk["subscription_id"] != subscription.subscription_id
    ]
    overlapping_ipv4_networks = [
        (trunk.description, trunk.iptrunk.iptrunk_ipv4_network)
        for trunk in all_trunks
        if trunk.iptrunk.iptrunk_ipv4_network.overlaps(subscription.iptrunk.iptrunk_ipv4_network)
    ]
    overlapping_ipv6_networks = [
        (trunk.description, trunk.iptrunk.iptrunk_ipv6_network)
        for trunk in all_trunks
        if trunk.iptrunk.iptrunk_ipv6_network.overlaps(subscription.iptrunk.iptrunk_ipv6_network)
    ]

    if overlapping_ipv4_networks or overlapping_ipv6_networks:
        msg = "Newly assigned IP networks overlap with existing IP trunk subscriptions, please investigate."
        raise ProcessFailureError(msg, details=[overlapping_ipv4_networks, overlapping_ipv6_networks])

dig_all_hosts_v4(new_ipv4_network)

Check if any hosts have already been assigned inside the IPv4 network in Netbox.

Source code in gso/workflows/iptrunk/create_iptrunk.py

@step("Check for existing DNS records in the assigned IPv4 network")
def dig_all_hosts_v4(new_ipv4_network: str) -> None:
    """Check if any hosts have already been assigned inside the IPv4 network in Netbox."""
    registered_hosts = [host for host in IPv4Network(new_ipv4_network) if infoblox.find_host_by_ip(IPv4Address(host))]

    if registered_hosts:
        msg = "One or more hosts in the assigned IPv4 network are already registered, please investigate."
        raise ProcessFailureError(msg, details=registered_hosts)

dig_all_hosts_v6(new_ipv6_network)

Check if any hosts have already been assigned inside the IPv6 network in Netbox.

Source code in gso/workflows/iptrunk/create_iptrunk.py

@step("Check for existing DNS records in the assigned IPv6 network")
def dig_all_hosts_v6(new_ipv6_network: str) -> None:
    """Check if any hosts have already been assigned inside the IPv6 network in Netbox."""
    registered_hosts = [host for host in IPv6Network(new_ipv6_network) if infoblox.find_host_by_ip(IPv6Address(host))]

    if registered_hosts:
        msg = "One or more hosts in the assigned IPv6 network are already registered, please investigate."
        raise ProcessFailureError(msg, details=registered_hosts)

ping_all_hosts_v4(new_ipv4_network)

Ping all hosts in the IPv4 network to verify they are not in use.

Source code in gso/workflows/iptrunk/create_iptrunk.py

@step("Ping all hosts in the assigned IPv4 network")
def ping_all_hosts_v4(new_ipv4_network: str) -> None:
    """Ping all hosts in the IPv4 network to verify they are not in use."""
    unavailable_hosts = [host for host in IPv4Network(new_ipv4_network) if ping(str(host), timeout=1)]

    if unavailable_hosts:
        msg = "One or more hosts in the assigned IPv4 network are responding to ping, please investigate."
        raise ProcessFailureError(msg, details=unavailable_hosts)

ping_all_hosts_v6(new_ipv6_network)

Ping all hosts in the IPv6 network to verify they are not in use.

Source code in gso/workflows/iptrunk/create_iptrunk.py

@step("Ping all hosts in the assigned IPv6 network")
def ping_all_hosts_v6(new_ipv6_network: str) -> State:
    """Ping all hosts in the IPv6 network to verify they are not in use."""
    unavailable_hosts = [host for host in IPv6Network(new_ipv6_network) if ping(str(host), timeout=1)]

    if unavailable_hosts:
        msg = "One or more hosts in the assigned IPv6 network are responding to ping, please investigate."
        raise ProcessFailureError(msg, details=unavailable_hosts)

    return {"__remove_keys": ["new_ipv4_network", "new_ipv6_network"]}

initialize_subscription(subscription, iptrunk_type, iptrunk_description, iptrunk_speed, iptrunk_minimum_links, side_a_node_id, side_a_ae_iface, side_a_ae_members, side_b_node_id, side_b_ae_iface, side_b_ae_members)

Take all input from the user, and store it in the database.

Source code in gso/workflows/iptrunk/create_iptrunk.py

@step("Initialize subscription")
def initialize_subscription(
    subscription: IptrunkInactive,
    iptrunk_type: IptrunkType,
    iptrunk_description: str | None,
    iptrunk_speed: PhysicalPortCapacity,
    iptrunk_minimum_links: int,
    side_a_node_id: str,
    side_a_ae_iface: str,
    side_a_ae_members: list[dict],
    side_b_node_id: str,
    side_b_ae_iface: str,
    side_b_ae_members: list[dict],
) -> State:
    """Take all input from the user, and store it in the database."""
    oss_params = load_oss_params()
    side_a = Router.from_subscription(side_a_node_id).router
    side_b = Router.from_subscription(side_b_node_id).router
    gs_id = generate_unique_id(prefix="GS")
    subscription.iptrunk.gs_id = gs_id
    subscription.iptrunk.iptrunk_description = iptrunk_description
    subscription.iptrunk.iptrunk_type = iptrunk_type
    subscription.iptrunk.iptrunk_speed = iptrunk_speed
    subscription.iptrunk.iptrunk_isis_metric = oss_params.GENERAL.isis_high_metric
    subscription.iptrunk.iptrunk_minimum_links = iptrunk_minimum_links

    subscription.iptrunk.iptrunk_sides[0].iptrunk_side_node = side_a
    subscription.iptrunk.iptrunk_sides[0].iptrunk_side_ae_iface = side_a_ae_iface
    side_a_ga_id = generate_unique_id(prefix="GA")
    subscription.iptrunk.iptrunk_sides[0].ga_id = side_a_ga_id
    for member in side_a_ae_members:
        subscription.iptrunk.iptrunk_sides[0].iptrunk_side_ae_members.append(
            IptrunkInterfaceBlockInactive.new(subscription_id=uuid4(), **member),
        )

    subscription.iptrunk.iptrunk_sides[1].iptrunk_side_node = side_b
    subscription.iptrunk.iptrunk_sides[1].iptrunk_side_ae_iface = side_b_ae_iface
    side_b_ga_id = generate_unique_id(prefix="GA")
    subscription.iptrunk.iptrunk_sides[1].ga_id = side_b_ga_id
    for member in side_b_ae_members:
        subscription.iptrunk.iptrunk_sides[1].iptrunk_side_ae_members.append(
            IptrunkInterfaceBlockInactive.new(subscription_id=uuid4(), **member),
        )
    side_names = sorted([side_a.router_site.site_name, side_b.router_site.site_name])
    subscription.description = f"IP trunk {side_names[0]} {side_names[1]}, {gs_id}"

    return {"subscription": subscription}

provision_ip_trunk_iface_dry(subscription, process_id, tt_number)

Perform a dry run of deploying configuration on both sides of the trunk.

Source code in gso/workflows/iptrunk/create_iptrunk.py

@step("[DRY RUN] Provision IP trunk interface")
def provision_ip_trunk_iface_dry(subscription: IptrunkInactive, process_id: UUIDstr, tt_number: str) -> LSOState:
    """Perform a dry run of deploying configuration on both sides of the trunk."""
    extra_vars = {
        "wfo_trunk_json": json.loads(json_dumps(subscription)),
        "dry_run": True,
        "verb": "deploy",
        "config_object": "trunk_interface",
        "commit_comment": f"GSO_PROCESS_ID: {process_id} - TT_NUMBER: {tt_number} - Deploy config for "
        f"{subscription.iptrunk.gs_id}",
    }

    return {
        "playbook_name": "gap_ansible/playbooks/iptrunks.yaml",
        "inventory": {
            "all": {
                "hosts": {
                    subscription.iptrunk.iptrunk_sides[0].iptrunk_side_node.router_fqdn: None,
                    subscription.iptrunk.iptrunk_sides[1].iptrunk_side_node.router_fqdn: None,
                }
            }
        },
        "extra_vars": extra_vars,
    }

provision_ip_trunk_iface_real(subscription, process_id, tt_number)

Deploy IP trunk configuration on both sides.

Source code in gso/workflows/iptrunk/create_iptrunk.py

@step("[FOR REAL] Provision IP trunk interface")
def provision_ip_trunk_iface_real(subscription: IptrunkInactive, process_id: UUIDstr, tt_number: str) -> LSOState:
    """Deploy IP trunk configuration on both sides."""
    extra_vars = {
        "wfo_trunk_json": json.loads(json_dumps(subscription)),
        "dry_run": False,
        "verb": "deploy",
        "config_object": "trunk_interface",
        "commit_comment": f"GSO_PROCESS_ID: {process_id} - TT_NUMBER: {tt_number} - Deploy config for "
        f"{subscription.iptrunk.gs_id}",
    }

    return {
        "playbook_name": "gap_ansible/playbooks/iptrunks.yaml",
        "inventory": {
            "all": {
                "hosts": {
                    subscription.iptrunk.iptrunk_sides[0].iptrunk_side_node.router_fqdn: None,
                    subscription.iptrunk.iptrunk_sides[1].iptrunk_side_node.router_fqdn: None,
                }
            }
        },
        "extra_vars": extra_vars,
    }

check_ip_trunk_connectivity(subscription)

Check successful connectivity across the new trunk.

Source code in gso/workflows/iptrunk/create_iptrunk.py

@step("Check IP connectivity of the trunk")
def check_ip_trunk_connectivity(subscription: IptrunkInactive) -> LSOState:
    """Check successful connectivity across the new trunk."""
    extra_vars = {"wfo_ip_trunk_json": json.loads(json_dumps(subscription)), "check": "ping"}

    return {
        "playbook_name": "gap_ansible/playbooks/iptrunks_checks.yaml",
        "inventory": {"all": {"hosts": {subscription.iptrunk.iptrunk_sides[0].iptrunk_side_node.router_fqdn: None}}},
        "extra_vars": extra_vars,
    }

provision_ip_trunk_isis_iface_dry(subscription, process_id, tt_number)

Perform a dry run of deploying ISIS configuration.

Source code in gso/workflows/iptrunk/create_iptrunk.py

@step("[DRY RUN] Provision IP trunk ISIS interface")
def provision_ip_trunk_isis_iface_dry(subscription: IptrunkInactive, process_id: UUIDstr, tt_number: str) -> LSOState:
    """Perform a dry run of deploying ISIS configuration."""
    extra_vars = {
        "wfo_trunk_json": json.loads(json_dumps(subscription)),
        "dry_run": True,
        "verb": "deploy",
        "config_object": "isis_interface",
        "commit_comment": f"GSO_PROCESS_ID: {process_id} - TT_NUMBER: {tt_number} - Deploy config for "
        f"{subscription.iptrunk.gs_id}",
    }

    return {
        "playbook_name": "gap_ansible/playbooks/iptrunks.yaml",
        "inventory": {
            "all": {
                "hosts": {
                    subscription.iptrunk.iptrunk_sides[0].iptrunk_side_node.router_fqdn: None,
                    subscription.iptrunk.iptrunk_sides[1].iptrunk_side_node.router_fqdn: None,
                }
            }
        },
        "extra_vars": extra_vars,
    }

provision_ip_trunk_isis_iface_real(subscription, process_id, tt_number)

Deploy ISIS configuration on both sides.

Source code in gso/workflows/iptrunk/create_iptrunk.py

@step("[FOR REAL] Provision IP trunk ISIS interface")
def provision_ip_trunk_isis_iface_real(subscription: IptrunkInactive, process_id: UUIDstr, tt_number: str) -> LSOState:
    """Deploy ISIS configuration on both sides."""
    extra_vars = {
        "wfo_trunk_json": json.loads(json_dumps(subscription)),
        "dry_run": False,
        "verb": "deploy",
        "config_object": "isis_interface",
        "commit_comment": f"GSO_PROCESS_ID: {process_id} - TT_NUMBER: {tt_number} - Deploy config for "
        f"{subscription.iptrunk.gs_id}",
    }

    return {
        "playbook_name": "gap_ansible/playbooks/iptrunks.yaml",
        "inventory": {
            "all": {
                "hosts": {
                    subscription.iptrunk.iptrunk_sides[0].iptrunk_side_node.router_fqdn: None,
                    subscription.iptrunk.iptrunk_sides[1].iptrunk_side_node.router_fqdn: None,
                }
            }
        },
        "extra_vars": extra_vars,
    }

check_ip_trunk_isis(subscription)

Run an Ansible playbook to confirm ISIS adjacency.

Source code in gso/workflows/iptrunk/create_iptrunk.py

@step("Check ISIS adjacency")
def check_ip_trunk_isis(subscription: IptrunkInactive) -> LSOState:
    """Run an Ansible playbook to confirm ISIS adjacency."""
    extra_vars = {"wfo_ip_trunk_json": json.loads(json_dumps(subscription)), "check": "isis"}

    return {
        "playbook_name": "gap_ansible/playbooks/iptrunks_checks.yaml",
        "inventory": {"all": {"hosts": {subscription.iptrunk.iptrunk_sides[0].iptrunk_side_node.router_fqdn: None}}},
        "extra_vars": extra_vars,
    }

register_dns_records(subscription)

Register DNS records for both sides of the newly created IP trunk.

Source code in gso/workflows/iptrunk/create_iptrunk.py

@step("Register DNS records for both sides of the trunk")
def register_dns_records(subscription: IptrunkInactive) -> State:
    """Register DNS records for both sides of the newly created IP trunk."""
    for index, side in enumerate(subscription.iptrunk.iptrunk_sides):
        fqdn = f"{side.iptrunk_side_ae_iface}-0.{side.iptrunk_side_node.router_fqdn}"
        if not (subscription.iptrunk.iptrunk_ipv4_network and subscription.iptrunk.iptrunk_ipv6_network):
            msg = f"Missing IP resources in trunk, cannot allocate DNS record for side {fqdn}!"
            raise ValueError(msg)
        ipv4_addr = subscription.iptrunk.iptrunk_ipv4_network[index]
        ipv6_addr = subscription.iptrunk.iptrunk_ipv6_network[index + 1]

        infoblox.create_host_by_ip(
            fqdn, "TRUNK", str(subscription.subscription_id), ipv4_address=ipv4_addr, ipv6_address=ipv6_addr
        )

    return {"subscription": subscription}

reserve_interfaces_in_netbox(subscription)

Create the LAG interfaces in NetBox and attach the LAG interfaces to the physical interfaces.

Source code in gso/workflows/iptrunk/create_iptrunk.py

@step("NextBox integration")
def reserve_interfaces_in_netbox(subscription: IptrunkInactive) -> State:
    """Create the LAG interfaces in NetBox and attach the LAG interfaces to the physical interfaces."""
    nbclient = NetboxClient()
    for trunk_side in subscription.iptrunk.iptrunk_sides:
        if get_router_vendor(trunk_side.iptrunk_side_node.owner_subscription_id) == Vendor.NOKIA:
            # Create LAG interfaces
            lag_interface: Interfaces = nbclient.create_interface(
                iface_name=trunk_side.iptrunk_side_ae_iface,  # type: ignore[arg-type]
                interface_type="lag",
                device_name=trunk_side.iptrunk_side_node.router_fqdn,  # type: ignore[arg-type]
                description=str(subscription.subscription_id),
                enabled=True,
            )
            # Attach physical interfaces to LAG
            # Update interface description to subscription ID
            # Reserve interfaces
            for interface in trunk_side.iptrunk_side_ae_members:
                nbclient.attach_interface_to_lag(
                    device_name=trunk_side.iptrunk_side_node.router_fqdn,  # type: ignore[arg-type]
                    lag_name=lag_interface.name,
                    iface_name=interface.interface_name,  # type: ignore[arg-type]
                    description=str(subscription.subscription_id),
                )
                nbclient.reserve_interface(
                    device_name=trunk_side.iptrunk_side_node.router_fqdn,  # type: ignore[arg-type]
                    iface_name=interface.interface_name,  # type: ignore[arg-type]
                )
    return {
        "subscription": subscription,
    }

netbox_allocate_side_a_interfaces(subscription)

Allocate the LAG interfaces for the Nokia router on side A.

Source code in gso/workflows/iptrunk/create_iptrunk.py

@step("Allocate interfaces in Netbox for side A")
def netbox_allocate_side_a_interfaces(subscription: IptrunkInactive) -> None:
    """Allocate the LAG interfaces for the Nokia router on side A."""
    _allocate_interfaces_in_netbox(subscription.iptrunk.iptrunk_sides[0])

netbox_allocate_side_b_interfaces(subscription)

Allocate the LAG interfaces for the Nokia router on side B.

Source code in gso/workflows/iptrunk/create_iptrunk.py

@step("Allocate interfaces in Netbox for side B")
def netbox_allocate_side_b_interfaces(subscription: IptrunkInactive) -> None:
    """Allocate the LAG interfaces for the Nokia router on side B."""
    _allocate_interfaces_in_netbox(subscription.iptrunk.iptrunk_sides[1])

create_new_sharepoint_checklist(subscription, tt_number, process_id)

Create a new checklist item in SharePoint for approving this IP trunk.

Source code in gso/workflows/iptrunk/create_iptrunk.py

@step("Create a new SharePoint checklist item")
def create_new_sharepoint_checklist(subscription: IptrunkProvisioning, tt_number: str, process_id: UUIDstr) -> State:
    """Create a new checklist item in SharePoint for approving this IP trunk."""
    new_list_item_url = SharePointClient().add_list_item(
        list_name="ip_trunk",
        fields={
            "Title": f"{subscription.description} - {subscription.iptrunk.gs_id}",
            "TT_NUMBER": tt_number,
            "GAP_PROCESS_URL": f"{load_oss_params().GENERAL.public_hostname}/workflows/{process_id}",
            "ACTIVITY_TYPE": "Creation",
        },
    )

    return {"checklist_url": new_list_item_url}

create_iptrunk()

Create a new IP trunk.

• Create the subscription object in the database
• Gather relevant information from Infoblox
• Reserve interfaces in Netbox
• Deploy configuration on the two sides of the trunk, first as a dry run
• Check connectivity on the new trunk
• Deploy the new ISIS metric on the trunk, first as a dry run
• Verify ISIS adjacency
• Allocate the interfaces in Netbox
• Set the subscription to active in the database

Source code in gso/workflows/iptrunk/create_iptrunk.py

@workflow(
    "Create IP trunk",
    initial_input_form=wrap_create_initial_input_form(initial_input_form_generator),
    target=Target.CREATE,
)
def create_iptrunk() -> StepList:
    """Create a new IP trunk.

    * Create the subscription object in the database
    * Gather relevant information from Infoblox
    * Reserve interfaces in Netbox
    * Deploy configuration on the two sides of the trunk, first as a dry run
    * Check connectivity on the new trunk
    * Deploy the new ISIS metric on the trunk, first as a dry run
    * Verify ISIS adjacency
    * Allocate the interfaces in Netbox
    * Set the subscription to active in the database
    """
    side_a_is_nokia = conditional(lambda state: get_router_vendor(state["side_a_node_id"]) == Vendor.NOKIA)
    side_b_is_nokia = conditional(lambda state: get_router_vendor(state["side_b_node_id"]) == Vendor.NOKIA)

    assign_ip_networks = step_group(
        name="Assign IP networks",
        steps=(
            begin
            >> get_info_from_ipam
            >> check_existing_trunk_allocations
            >> dig_all_hosts_v4
            >> dig_all_hosts_v6
            >> ping_all_hosts_v4
            >> ping_all_hosts_v6
        ),
    )

    return (
        begin
        >> create_subscription
        >> store_process_subscription(Target.CREATE)
        >> initialize_subscription
        >> assign_ip_networks
        >> reserve_interfaces_in_netbox
        >> lso_interaction(provision_ip_trunk_iface_dry)
        >> lso_interaction(provision_ip_trunk_iface_real)
        >> lso_interaction(check_ip_trunk_connectivity)
        >> lso_interaction(provision_ip_trunk_isis_iface_dry)
        >> lso_interaction(provision_ip_trunk_isis_iface_real)
        >> lso_interaction(check_ip_trunk_isis)
        >> register_dns_records
        >> side_a_is_nokia(netbox_allocate_side_a_interfaces)
        >> side_b_is_nokia(netbox_allocate_side_b_interfaces)
        >> set_status(SubscriptionLifecycle.PROVISIONING)
        >> create_new_sharepoint_checklist
        >> prompt_sharepoint_checklist_url
        >> resync
        >> done
    )