diff --git a/gso/workflows/router/terminate_router.py b/gso/workflows/router/terminate_router.py index 4599c6f099555b79a9a5a6a5f1229f4878dd8013..75a78281d7cb8914059e69b44bedeb90148a9f97 100644 --- a/gso/workflows/router/terminate_router.py +++ b/gso/workflows/router/terminate_router.py @@ -15,6 +15,7 @@ The workflow consists of the following steps: - Set the subscription status to `TERMINATED`. """ +import datetime import ipaddress import json import logging @@ -34,6 +35,7 @@ from orchestrator.workflows.steps import ( unsync, ) from orchestrator.workflows.utils import wrap_modify_initial_input_form +from pydantic import Field from pydantic_forms.types import FormGenerator, State, UUIDstr from requests import HTTPError @@ -63,19 +65,28 @@ def initial_input_form_generator(subscription_id: UUIDstr) -> FormGenerator: class TerminateForm(SubmitFormPage): if router.status == SubscriptionLifecycle.INITIAL: - info_label_2: Label = ( + info_label_2: Label = Field( "This will immediately mark the subscription as terminated, preventing any other workflows from " - "interacting with this product subscription." + "interacting with this product subscription.", + exclude=True, + ) + info_label_3: Label = Field( + "ONLY EXECUTE THIS WORKFLOW WHEN YOU ARE ABSOLUTELY SURE WHAT YOU ARE DOING.", exclude=True ) - info_label_3: Label = "ONLY EXECUTE THIS WORKFLOW WHEN YOU ARE ABSOLUTELY SURE WHAT YOU ARE DOING." tt_number: TTNumber - termination_label: Label = "Please confirm whether configuration should get removed from the router." + termination_label: Label = Field( + "Please confirm whether configuration should get removed from the router.", exclude=True + ) remove_configuration: bool = False - update_ibgp_mesh_label: Label = "Please confirm whether the iBGP mesh should get updated." + update_ibgp_mesh_label: Label = Field("Please confirm whether the iBGP mesh should get updated.", exclude=True) update_ibgp_mesh: bool = True - update_sdp_mesh_label: Label = "Please confirm whether the SDP mesh should get updated." + update_sdp_mesh_label: Label = Field("Please confirm whether the SDP mesh should get updated.", exclude=True) update_sdp_mesh: bool = True + remove_loopback_from_ipam_label: Label = Field( + "Please confirm whether the loopback address should be released in IPAM.", exclude=True + ) + remove_loopback_from_ipam: bool = False user_input = yield TerminateForm return user_input.model_dump() | { @@ -85,11 +96,20 @@ def initial_input_form_generator(subscription_id: UUIDstr) -> FormGenerator: @step("Deprovision loopback IPs from IPAM") -def deprovision_loopback_ips(subscription: Router) -> dict: +def deprovision_loopback_ips(subscription: Router, remove_loopback_from_ipam: bool, process_id: UUIDstr) -> None: # noqa: FBT001 """Clear up the loopback addresses from IPAM.""" - infoblox.delete_host_by_ip(ipaddress.IPv4Address(subscription.router.router_lo_ipv4_address)) - - return {"subscription": subscription} + if remove_loopback_from_ipam: + infoblox.delete_host_by_ip(ipaddress.IPv4Address(subscription.router.router_lo_ipv4_address)) + else: + record = infoblox.find_host_by_fqdn(subscription.router.router_fqdn) + if record: + # We keep the record in IPAM but add a comment stating that this router is terminated. + # This is done to prevent an address from being re-used. + record.comment = ( + f"This router was terminated by GAP process {process_id} on " + f"{datetime.datetime.now(tz=datetime.UTC).strftime("%d/%m/%Y")}." + ) + record.update() @step("[DRY RUN] Remove configuration from router") diff --git a/test/workflows/router/test_terminate_router.py b/test/workflows/router/test_terminate_router.py index a7303181ecbf3ab50106ed9627f539e96b1add6f..2aeaca372b0a97042dc8e9bd9d858bcba5095790 100644 --- a/test/workflows/router/test_terminate_router.py +++ b/test/workflows/router/test_terminate_router.py @@ -37,6 +37,7 @@ def test_terminate_pe_router_full_success( "remove_configuration": remove_configuration, "update_ibgp_mesh": update_ibgp_mesh, "update_sdp_mesh": update_sdp_mesh, + "remove_loopback_from_ipam": True, } lso_interaction_count = 0 if remove_configuration: @@ -89,6 +90,7 @@ def test_terminate_p_router_full_success( "tt_number": faker.tt_number(), "remove_configuration": remove_configuration, "update_ibgp_mesh": update_ibgp_mesh, + "remove_loopback_from_ipam": True, } lso_interaction_count = 0 if remove_configuration: