Skip to content
Snippets Groups Projects
Verified Commit 2d9a9d55 authored by Karel van Klink's avatar Karel van Klink :smiley_cat:
Browse files

add docstrings to router workflows

parent a95bfff3
No related branches found
No related tags found
1 merge request!111Feature/ruff everything party hat emoji
"""Workflows related to router subscriptions."""
"""A creation workflow for adding a new router to the network."""
from ipaddress import IPv4Network, IPv6Network
from typing import Any
......@@ -36,6 +38,7 @@ def _site_selector() -> Choice:
def initial_input_form_generator(product_name: str) -> FormGenerator:
"""Gather information about the new router from the operator."""
class CreateRouterForm(FormPage):
class Config:
title = product_name
......@@ -71,6 +74,7 @@ def initial_input_form_generator(product_name: str) -> FormGenerator:
@step("Create subscription")
def create_subscription(product: UUIDstr, customer: UUIDstr) -> State:
"""Create a new subscription object."""
subscription = RouterInactive.from_product_id(product, customer)
return {
......@@ -88,6 +92,7 @@ def initialize_subscription(
router_site: str,
router_role: RouterRole,
) -> State:
"""Initialise the subscription object in the service database."""
subscription.router.router_ts_port = ts_port
subscription.router.router_vendor = router_vendor
subscription.router.router_site = Site.from_subscription(router_site).site
......@@ -108,6 +113,7 @@ def initialize_subscription(
@step("Allocate loopback interfaces in IPAM")
def ipam_allocate_loopback(subscription: RouterProvisioning, is_ias_connected: bool) -> State: # noqa: FBT001
"""Allocate :term:`IPAM` resources for the loopback interface."""
fqdn = subscription.router.router_fqdn
loopback_v4, loopback_v6 = infoblox.allocate_host(f"lo0.{fqdn}", "LO", [fqdn], str(subscription.subscription_id))
......@@ -121,6 +127,7 @@ def ipam_allocate_loopback(subscription: RouterProvisioning, is_ias_connected: b
@step("Allocate IAS connection in IPAM")
def ipam_allocate_ias_networks(subscription: RouterProvisioning) -> State:
"""Allocate required IAS :term:`IPAM` resources."""
fqdn = subscription.router.router_fqdn
subscription.router.router_si_ipv4_network = infoblox.allocate_v4_network(
......@@ -146,6 +153,7 @@ def provision_router_dry(
callback_route: str,
tt_number: str,
) -> State:
"""Perform a dry run of deploying configuration on the router."""
provisioning_proxy.provision_router(subscription, process_id, callback_route, tt_number)
return {"subscription": subscription}
......@@ -158,6 +166,7 @@ def provision_router_real(
callback_route: str,
tt_number: str,
) -> State:
"""Deploy configuration on the router."""
provisioning_proxy.provision_router(subscription, process_id, callback_route, tt_number, dry_run=False)
return {"subscription": subscription}
......@@ -165,6 +174,10 @@ def provision_router_real(
@step("Create NetBox Device")
def create_netbox_device(subscription: RouterProvisioning) -> State:
"""Create a new device in Netbox.
HACK: use a conditional instead for execution of this step
"""
if subscription.router.router_vendor == RouterVendor.NOKIA:
NetboxClient().create_device(
subscription.router.router_fqdn,
......@@ -176,6 +189,7 @@ def create_netbox_device(subscription: RouterProvisioning) -> State:
@step("Verify IPAM resources for loopback interface")
def verify_ipam_loopback(subscription: RouterProvisioning) -> State:
"""Validate the :term:`IPAM` resources for the loopback interface."""
host_record = infoblox.find_host_by_fqdn(f"lo0.{subscription.router.router_fqdn}")
if not host_record or str(subscription.subscription_id) not in host_record.comment:
return {"ipam_warning": "Loopback record is incorrectly configured in IPAM, please investigate this manually!"}
......@@ -185,6 +199,7 @@ def verify_ipam_loopback(subscription: RouterProvisioning) -> State:
@step("Verify IPAM resources for IAS/LT networks")
def verify_ipam_ias(subscription: RouterProvisioning) -> State:
"""Validate the :term:`IPAM` resources related to this new router subscription."""
si_ipv4_network = infoblox.find_network_by_cidr(IPv4Network(subscription.router.router_si_ipv4_network))
ias_lt_ipv4_network = infoblox.find_network_by_cidr(IPv4Network(subscription.router.router_ias_lt_ipv4_network))
ias_lt_ipv6_network = infoblox.find_network_by_cidr(IPv6Network(subscription.router.router_ias_lt_ipv6_network))
......@@ -218,6 +233,14 @@ def verify_ipam_ias(subscription: RouterProvisioning) -> State:
target=Target.CREATE,
)
def create_router() -> StepList:
"""Create a new router in the service database.
* Create and initialise the subscription object in the service database
* Allocate :term:`IPAM` resources for the loopback interface
* Deploy configuration on the new router, first as a dry run
* Validate :term:`IPAM` resources
* Create a new device in Netbox
"""
should_allocate_ias = conditional(lambda state: state["is_ias_connected"])
return (
......
"""A workflow that terminates a router."""
import ipaddress
import logging
......@@ -23,6 +25,7 @@ logger = logging.getLogger(__name__)
def initial_input_form_generator(subscription_id: UUIDstr) -> FormGenerator:
"""Let the operator decide whether to delete configuration on the router, and/or clear up :term:`IPAM` resources."""
Router.from_subscription(subscription_id)
class TerminateForm(FormPage):
......@@ -40,6 +43,7 @@ def initial_input_form_generator(subscription_id: UUIDstr) -> FormGenerator:
@step("Deprovision loopback IPs from IPAM")
def deprovision_loopback_ips(subscription: Router) -> dict:
"""Clear up the loopback addresses from :term:`IPAM`."""
infoblox.delete_host_by_ip(ipaddress.IPv4Address(subscription.router.router_lo_ipv4_address))
return {"subscription": subscription}
......@@ -47,6 +51,7 @@ def deprovision_loopback_ips(subscription: Router) -> dict:
@step("Deprovision SI interface network from IPAM")
def deprovision_si_ips(subscription: Router) -> dict:
"""Clear up SI interface resources from :term:`IPAM`."""
infoblox.delete_network(ipaddress.IPv4Network(subscription.router.router_si_ipv4_network))
return {"subscription": subscription}
......@@ -54,6 +59,7 @@ def deprovision_si_ips(subscription: Router) -> dict:
@step("Deprovision IAS LT interfaces from IPAM")
def deprovision_lt_ips(subscription: Router) -> dict:
"""Clear up IAS LT interfaces from :term:`IPAM`."""
infoblox.delete_network(ipaddress.IPv4Network(subscription.router.router_ias_lt_ipv4_network))
infoblox.delete_network(ipaddress.IPv6Network(subscription.router.router_ias_lt_ipv6_network))
......@@ -62,13 +68,16 @@ def deprovision_lt_ips(subscription: Router) -> dict:
@step("Remove configuration from router")
def remove_config_from_router() -> None:
# FIXME: Add actual content
# TODO: update unit test accordingly
pass
"""Remove configuration from the router, first as a dry run.
FIXME: Add actual content
TODO: update unit test accordingly
"""
@step("Remove Device from NetBox")
def remove_device_from_netbox(subscription: Router) -> dict[str, Router]:
"""Remove the device from Netbox."""
if subscription.router.router_vendor == RouterVendor.NOKIA:
NetboxClient().delete_device(subscription.router.router_fqdn)
return {"subscription": subscription}
......@@ -80,6 +89,13 @@ def remove_device_from_netbox(subscription: Router) -> dict[str, Router]:
target=Target.TERMINATE,
)
def terminate_router() -> StepList:
"""Terminate a router subscription.
* Let the operator decide whether to delete :term:`IPAM` resources, and remove configuration from the router
* Clear up :term:`IPAM` resources, if selected by the operator
* Disable and delete configuration on the router, if selected by the operator
* Mark the subscription as terminated in the service database
"""
run_ipam_steps = conditional(lambda state: state["clean_up_ipam"])
run_config_steps = conditional(lambda state: state["remove_configuration"])
run_ias_removal = conditional(lambda state: state["subscription"]["router"]["router_is_ias_connected"])
......
0% Loading or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment