Abstract IP trunk version selector into utils

gso/utils/helpers.py

@@ -10,7 +10,6 @@ from orchestrator.types import SubscriptionLifecycle
 from pydantic_forms.types import UUIDstr
 from pydantic_forms.validators import Choice
 
-from gso import settings
 from gso.products.product_blocks.layer_2_circuit import Layer2CircuitType
 from gso.products.product_blocks.router import RouterRole
 from gso.products.product_types.router import Router
@@ -25,6 +24,7 @@ from gso.services.subscriptions import (
     get_router_subscriptions,
     is_virtual_circuit_id_available,
 )
+from gso.settings import load_oss_params
 from gso.utils.shared_enums import Vendor
 from gso.utils.types.interfaces import PhysicalPortCapacity
 from gso.utils.types.ip_address import IPv4AddressType, IPv4NetworkType, IPv6NetworkType
@@ -131,13 +131,13 @@ def iso_from_ipv4(ipv4_address: IPv4AddressType) -> str:
 
 def generate_fqdn(hostname: str, site_name: str, country_code: str) -> str:
     """Generate an FQDN from a hostname, site name, and a country code."""
-    oss = settings.load_oss_params()
+    oss = load_oss_params()
     return f"{hostname}.{site_name.lower()}.{country_code.lower()}{oss.IPAM.LO.domain_name}"
 
 
 def generate_lan_switch_interconnect_subnet_v4(site_internal_id: int) -> IPv4NetworkType:
     """Generate an IPv4 network in which a LAN Switch Interconnect resides, given a Site internal ID."""
-    ipam_oss = settings.load_oss_params().IPAM.LAN_SWITCH_INTERCONNECT
+    ipam_oss = load_oss_params().IPAM.LAN_SWITCH_INTERCONNECT
 
     result = str(ipam_oss.V4.containers[0]).split(".")[:2]  # Take the first two octets from the IPv4 network.
     result.append(str(site_internal_id))  # Append the side ID as the third octet.
@@ -148,7 +148,7 @@ def generate_lan_switch_interconnect_subnet_v4(site_internal_id: int) -> IPv4Net
 
 def generate_lan_switch_interconnect_subnet_v6(site_internal_id: int) -> IPv6NetworkType:
     """Generate an IPv6 network in which a LAN Switch Interconnect resides, given a Site internal ID."""
-    ipam_oss = settings.load_oss_params().IPAM.LAN_SWITCH_INTERCONNECT
+    ipam_oss = load_oss_params().IPAM.LAN_SWITCH_INTERCONNECT
 
     result = IPv6Network(ipam_oss.V6.containers[0]).exploded[:17]  # Take the first 56 bits of the network
     result += str(hex(site_internal_id)[2:])  # Append the site internal id for bytes 57 to 64 as hexadecimal number
@@ -283,6 +283,19 @@ def active_edge_port_selector(*, partner_id: UUIDstr | None = None) -> TypeAlias
     )
 
 
+def ip_trunk_service_version_selector() -> TypeAlias:
+    """Generate a dropdown selector for choosing a service version."""
+    iptrunk_versions = load_oss_params().SERVICE_VERSIONS.IP_TRUNK.version
+
+    return cast(
+        type[Choice],
+        Choice.__call__(
+            "Select an IP trunk service version.",
+            [(k, f"Version {k} - {iptrunk_versions[k]}") for k in iptrunk_versions],
+        ),
+    )
+
+
 def partner_choice() -> Choice:
     """Return a Choice object containing a list of available partners."""
     partners = {partner.partner_id: partner.name for partner in get_all_partners()}

gso/workflows/iptrunk/create_iptrunk.py

@@ -68,6 +68,7 @@ from gso.utils.helpers import (
     available_lags_choices,
     calculate_recommended_minimum_links,
     get_router_vendor,
+    ip_trunk_service_version_selector,
 )
 from gso.utils.shared_enums import Vendor
 from gso.utils.types.interfaces import JuniperLAGMember, LAGMember, LAGMemberList, PhysicalPortCapacity
@@ -85,10 +86,6 @@ def initial_input_form_generator(product_name: str) -> FormGenerator:
     ) + subscriptions.get_provisioning_router_subscriptions(includes=["subscription_id", "description"])
     routers = {str(router["subscription_id"]): router["description"] for router in active_and_provisioning_routers}
 
-    # Get version choices from config
-    iptrunk_versions = list(load_oss_params().SERVICE_VERSIONS.IP_TRUNK.version.keys())
-    iptrunk_version_choices = Choice("Select version", [(v, v) for v in iptrunk_versions])  # type: ignore[arg-type]
-
     class CreateIptrunkForm(FormPage):
         model_config = ConfigDict(title=product_name)
 
@@ -99,7 +96,7 @@ def initial_input_form_generator(product_name: str) -> FormGenerator:
         iptrunk_speed: PhysicalPortCapacity
         iptrunk_number_of_members: int
         iptrunk_description_suffix: str | None = None
-        iptrunk_config_version: iptrunk_version_choices  # type: ignore[valid-type]
+        iptrunk_config_version: ip_trunk_service_version_selector()  # type: ignore[valid-type]
 
     initial_user_input = yield CreateIptrunkForm
     recommended_minimum_links = calculate_recommended_minimum_links(

gso/workflows/iptrunk/modify_trunk_interface.py

@@ -20,7 +20,7 @@ from orchestrator.workflows.steps import resync, store_process_subscription, uns
 from orchestrator.workflows.utils import wrap_modify_initial_input_form
 from pydantic import AfterValidator, ConfigDict, Field
 from pydantic_forms.types import FormGenerator, State, UUIDstr
-from pydantic_forms.validators import Choice, Label, ReadOnlyField
+from pydantic_forms.validators import Label, ReadOnlyField
 
 from gso.products.product_blocks.iptrunk import (
     IptrunkInterfaceBlock,
@@ -30,12 +30,12 @@ from gso.products.product_blocks.iptrunk import (
 from gso.products.product_types.iptrunk import Iptrunk
 from gso.services.lso_client import LSOState, lso_interaction
 from gso.services.netbox_client import NetboxClient
-from gso.settings import load_oss_params
 from gso.utils.helpers import (
     available_interfaces_choices,
     available_interfaces_choices_including_current_members,
     calculate_recommended_minimum_links,
     get_router_vendor,
+    ip_trunk_service_version_selector,
 )
 from gso.utils.shared_enums import Vendor
 from gso.utils.types.interfaces import JuniperLAGMember, LAGMember, LAGMemberList, PhysicalPortCapacity
@@ -87,10 +87,6 @@ def initial_input_form_generator(subscription_id: UUIDstr) -> FormGenerator:
     """Gather input from the operator on the interfaces that should be modified."""
     subscription = Iptrunk.from_subscription(subscription_id)
 
-    # Get version choices from config (single-select dropdown)
-    iptrunk_versions = list(load_oss_params().SERVICE_VERSIONS.IP_TRUNK.version.keys())
-    iptrunk_version_choices = Choice("Select version", [(v, v) for v in iptrunk_versions])  # type: ignore[arg-type]
-
     class ModifyIptrunkForm(FormPage):
         tt_number: TTNumber
         gs_id: (
@@ -100,7 +96,7 @@ def initial_input_form_generator(subscription_id: UUIDstr) -> FormGenerator:
             | None
         ) = subscription.iptrunk.gs_id
         iptrunk_description: str | None = subscription.iptrunk.iptrunk_description
-        iptrunk_config_version: iptrunk_version_choices = subscription.iptrunk.iptrunk_config_version  # type: ignore[valid-type]
+        iptrunk_config_version: ip_trunk_service_version_selector() | str = subscription.iptrunk.iptrunk_config_version  # type: ignore[valid-type]
         iptrunk_type: IptrunkType | str = subscription.iptrunk.iptrunk_type  # FIXME: remove str workaround
         warning_label: Label = (
             "Changing the PhyPortCapacity will result in the deletion of all AE members. "

test/workflows/iptrunk/test_create_iptrunk.py

@@ -74,7 +74,7 @@ def input_form_wizard_data(request, router_subscription_factory, faker):
         "iptrunk_speed": PhysicalPortCapacity.HUNDRED_GIGABIT_PER_SECOND,
         "iptrunk_number_of_members": 2,
         "iptrunk_description_suffix": faker.word(),
-        "iptrunk_config_version": "1.0",
+        "iptrunk_config_version": "Version 1.0 - Base Version",
     }
     create_ip_trunk_confirm_step = {"iptrunk_minimum_links": 1}
     create_ip_trunk_side_a_router_name = {"side_a_node_id": router_side_a}