diff --git a/gso/utils/helpers.py b/gso/utils/helpers.py
index c20e8a4e780daff69688fa1b83b7cf57787ce99a..9a73d9c78581666645961ff08b70944446c6b8b7 100644
--- a/gso/utils/helpers.py
+++ b/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()}
diff --git a/gso/workflows/iptrunk/create_iptrunk.py b/gso/workflows/iptrunk/create_iptrunk.py
index 04b0d1459610147a292081868a0aed73c298dbee..e151068a2f931fd45c0c3f7d4a08279b2d9f2d3e 100644
--- a/gso/workflows/iptrunk/create_iptrunk.py
+++ b/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(
diff --git a/gso/workflows/iptrunk/modify_trunk_interface.py b/gso/workflows/iptrunk/modify_trunk_interface.py
index 1f3e9f5431c4e154775f54a52ee2ba71238d1b8b..9bb651f5e4bf587e68cf2214aa462166a8818a4a 100644
--- a/gso/workflows/iptrunk/modify_trunk_interface.py
+++ b/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. "
diff --git a/test/workflows/iptrunk/test_create_iptrunk.py b/test/workflows/iptrunk/test_create_iptrunk.py
index 33c9640aa5c25d6f59ea5a81af53d0150dd7dee4..851c3e3081ebc021457d052af8525c63a58cbee0 100644
--- a/test/workflows/iptrunk/test_create_iptrunk.py
+++ b/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}