update input form, migrate IPtrunk model, update workflows accordingly

gso/products/product_blocks/iptrunk.py

 """IP trunk product block that has all parameters of a subscription throughout its lifecycle."""
 
 import ipaddress
-from typing import Optional
+from typing import Optional, TypeVar
 
 from orchestrator.domain.base import ProductBlockModel
+from orchestrator.forms.validators import UniqueConstrainedList
 from orchestrator.types import SubscriptionLifecycle, strEnum
 from pydantic import Field
 
@@ -15,6 +16,40 @@ class IptrunkType(strEnum):
     LEASED = "Leased"
 
 
+T = TypeVar("T", covariant=True)
+
+
+class IptrunkSides(UniqueConstrainedList[T]):
+    min_items = 2
+    max_items = 2
+
+
+class IptrunkSideBlockInactive(
+    ProductBlockModel, lifecycle=[SubscriptionLifecycle.INITIAL], product_block_name="IptrunkSideBlock"
+):
+    iptrunk_side_node: RouterBlockInactive
+    iptrunk_side_ae_iface: Optional[str] = None
+    iptrunk_side_ae_geant_a_sid: Optional[str] = None
+    iptrunk_side_ae_members: list[str] = Field(default_factory=list)
+    iptrunk_side_ae_members_description: list[str] = Field(default_factory=list)
+
+
+class IptrunkSideBlockProvisioning(IptrunkSideBlockInactive, lifecycle=[SubscriptionLifecycle.PROVISIONING]):
+    iptrunk_side_node: RouterBlockProvisioning
+    iptrunk_side_ae_iface: Optional[str] = None
+    iptrunk_side_ae_geant_a_sid: Optional[str] = None
+    iptrunk_side_ae_members: list[str] = Field(default_factory=list)
+    iptrunk_side_ae_members_description: list[str] = Field(default_factory=list)
+
+
+class IptrunkSideBlock(IptrunkSideBlockProvisioning, lifecycle=[SubscriptionLifecycle.ACTIVE]):
+    iptrunk_side_node: RouterBlock
+    iptrunk_side_ae_iface: Optional[str] = None
+    iptrunk_side_ae_geant_a_sid: Optional[str] = None
+    iptrunk_side_ae_members: list[str] = Field(default_factory=list)
+    iptrunk_side_ae_members_description: list[str] = Field(default_factory=list)
+
+
 class IptrunkBlockInactive(
     ProductBlockModel, lifecycle=[SubscriptionLifecycle.INITIAL], product_block_name="IptrunkBlock"
 ):
@@ -29,17 +64,7 @@ class IptrunkBlockInactive(
     iptrunk_ipv4_network: Optional[ipaddress.IPv4Network] = None
     iptrunk_ipv6_network: Optional[ipaddress.IPv6Network] = None
     #
-    iptrunk_sideA_node: RouterBlockInactive
-    iptrunk_sideA_ae_iface: Optional[str] = None
-    iptrunk_sideA_ae_geant_a_sid: Optional[str] = None
-    iptrunk_sideA_ae_members: list[str] = Field(default_factory=list)
-    iptrunk_sideA_ae_members_description: list[str] = Field(default_factory=list)
-    #
-    iptrunk_sideB_node: RouterBlockInactive
-    iptrunk_sideB_ae_iface: Optional[str] = None
-    iptrunk_sideB_ae_geant_a_sid: Optional[str] = None
-    iptrunk_sideB_ae_members: list[str] = Field(default_factory=list)
-    iptrunk_sideB_ae_members_description: list[str] = Field(default_factory=list)
+    iptrunk_sides: IptrunkSides[IptrunkSideBlockInactive]
 
 
 class IptrunkBlockProvisioning(IptrunkBlockInactive, lifecycle=[SubscriptionLifecycle.PROVISIONING]):
@@ -54,17 +79,7 @@ class IptrunkBlockProvisioning(IptrunkBlockInactive, lifecycle=[SubscriptionLife
     iptrunk_ipv4_network: Optional[ipaddress.IPv4Network] = None
     iptrunk_ipv6_network: Optional[ipaddress.IPv6Network] = None
     #
-    iptrunk_sideA_node: RouterBlockProvisioning
-    iptrunk_sideA_ae_iface: Optional[str] = None
-    iptrunk_sideA_ae_geant_a_sid: Optional[str] = None
-    iptrunk_sideA_ae_members: list[str] = Field(default_factory=list)
-    iptrunk_sideA_ae_members_description: list[str] = Field(default_factory=list)
-    #
-    iptrunk_sideB_node: RouterBlockProvisioning
-    iptrunk_sideB_ae_iface: Optional[str] = None
-    iptrunk_sideB_ae_geant_a_sid: Optional[str] = None
-    iptrunk_sideB_ae_members: list[str] = Field(default_factory=list)
-    iptrunk_sideB_ae_members_description: list[str] = Field(default_factory=list)
+    iptrunk_sides: IptrunkSides[IptrunkSideBlockProvisioning]
 
 
 class IptrunkBlock(IptrunkBlockProvisioning, lifecycle=[SubscriptionLifecycle.ACTIVE]):
@@ -86,22 +101,4 @@ class IptrunkBlock(IptrunkBlockProvisioning, lifecycle=[SubscriptionLifecycle.AC
     """The IPv4 network used for this trunk."""
     iptrunk_ipv6_network: ipaddress.IPv6Network
     """The IPv6 network used for this trunk."""
-    #
-    iptrunk_sideA_node: RouterBlock
-    """The router that hosts the A side of the trunk."""
-    iptrunk_sideA_ae_iface: str
-    """The name of the interface on which the trunk connects."""
-    iptrunk_sideA_ae_geant_a_sid: str
-    """The service ID of the interface."""
-    iptrunk_sideA_ae_members: list[str] = Field(default_factory=list)
-    """A list of interface members that make up the aggregated Ethernet interface."""
-    iptrunk_sideA_ae_members_description: list[str] = Field(default_factory=list)
-    """The list of descriptions that describe the list of interface members."""
-    #
-    iptrunk_sideB_node: RouterBlock
-    """The router that hosts the B side of the trunk. It possesses the same attributes as the A-side, including the
-    interfaces and its descriptions."""
-    iptrunk_sideB_ae_iface: str
-    iptrunk_sideB_ae_geant_a_sid: str
-    iptrunk_sideB_ae_members: list[str] = Field(default_factory=list)
-    iptrunk_sideB_ae_members_description: list[str] = Field(default_factory=list)
+    iptrunk_sides: IptrunkSides[IptrunkSideBlock]

gso/workflows/iptrunk/create_iptrunk.py

+from uuid import UUID
+
+from orchestrator.db.models import ProductTable, SubscriptionTable
 from orchestrator.forms import FormPage
 from orchestrator.forms.validators import Choice, UniqueConstrainedList
 from orchestrator.targets import Target
@@ -7,7 +10,7 @@ from orchestrator.workflows.steps import resync, set_status, store_process_subsc
 from orchestrator.workflows.utils import wrap_create_initial_input_form
 
 from gso.products.product_blocks import PhyPortCapacity
-from gso.products.product_blocks.iptrunk import IptrunkType
+from gso.products.product_blocks.iptrunk import IptrunkType, IptrunkSideBlock
 from gso.products.product_types.iptrunk import IptrunkInactive, IptrunkProvisioning
 from gso.products.product_types.router import Router
 from gso.services import ipam, provisioning_proxy, subscriptions
@@ -128,17 +131,17 @@ def initialize_subscription(
     subscription.iptrunk.iptrunk_isis_metric = 9000
     subscription.iptrunk.iptrunk_minimum_links = iptrunk_minimum_links
 
-    subscription.iptrunk.iptrunk_sideA_node = Router.from_subscription(iptrunk_sideA_node_id).router
-    subscription.iptrunk.iptrunk_sideA_ae_iface = iptrunk_sideA_ae_iface
-    subscription.iptrunk.iptrunk_sideA_ae_geant_a_sid = iptrunk_sideA_ae_geant_a_sid
-    subscription.iptrunk.iptrunk_sideA_ae_members = iptrunk_sideA_ae_members
-    subscription.iptrunk.iptrunk_sideA_ae_members_description = iptrunk_sideA_ae_members_descriptions
+    subscription.iptrunk.iptrunk_sides[0].iptrunk_side_node = Router.from_subscription(iptrunk_sideA_node_id).router
+    subscription.iptrunk.iptrunk_sides[0].iptrunk_side_ae_iface = iptrunk_sideA_ae_iface
+    subscription.iptrunk.iptrunk_sides[0].iptrunk_side_ae_geant_a_sid = iptrunk_sideA_ae_geant_a_sid
+    subscription.iptrunk.iptrunk_sides[0].iptrunk_side_ae_members = iptrunk_sideA_ae_members
+    subscription.iptrunk.iptrunk_sides[0].iptrunk_side_ae_members_description = iptrunk_sideA_ae_members_descriptions
 
-    subscription.iptrunk.iptrunk_sideB_node = Router.from_subscription(iptrunk_sideB_node_id).router
-    subscription.iptrunk.iptrunk_sideB_ae_iface = iptrunk_sideB_ae_iface
-    subscription.iptrunk.iptrunk_sideB_ae_geant_a_sid = iptrunk_sideB_ae_geant_a_sid
-    subscription.iptrunk.iptrunk_sideB_ae_members = iptrunk_sideB_ae_members
-    subscription.iptrunk.iptrunk_sideB_ae_members_description = iptrunk_sideB_ae_members_descriptions
+    subscription.iptrunk.iptrunk_sides[1].iptrunk_side_node = Router.from_subscription(iptrunk_sideB_node_id).router
+    subscription.iptrunk.iptrunk_sides[1].iptrunk_side_ae_iface = iptrunk_sideB_ae_iface
+    subscription.iptrunk.iptrunk_sides[1].iptrunk_side_ae_geant_a_sid = iptrunk_sideB_ae_geant_a_sid
+    subscription.iptrunk.iptrunk_sides[1].iptrunk_side_ae_members = iptrunk_sideB_ae_members
+    subscription.iptrunk.iptrunk_sides[1].iptrunk_side_ae_members_description = iptrunk_sideB_ae_members_descriptions
 
     subscription.description = f"IP trunk, geant_s_sid:{geant_s_sid}"
     subscription = IptrunkProvisioning.from_other_lifecycle(subscription, SubscriptionLifecycle.PROVISIONING)
@@ -202,7 +205,7 @@ def check_ip_trunk_isis(subscription: IptrunkProvisioning, process_id: UUIDstr)
 
     return {
         "subscription": subscription,
-        "label_text": "Checking ISIS adjacencies, please refresh to get the results of the playbook.",
+        "label_text": "Checking ISIS adjacency, please refresh to get the results of the playbook.",
     }
 
 

gso/workflows/iptrunk/migrate_iptrunk.py

@@ -10,9 +10,10 @@ from orchestrator.types import FormGenerator, State, UUIDstr
 from orchestrator.workflow import StepList, done, init
 from orchestrator.workflows.steps import resync, store_process_subscription, unsync
 from orchestrator.workflows.utils import wrap_modify_initial_input_form
-from products import Iptrunk
 from pydantic import validator
 
+from products import Iptrunk
+
 
 def initial_input_form_generator(subscription_id: UUIDstr) -> FormGenerator:
     subscription = Iptrunk.from_subscription(subscription_id)
@@ -28,35 +29,38 @@ def initial_input_form_generator(subscription_id: UUIDstr) -> FormGenerator:
         .all()
     ):
         if router_id not in [
-            subscription.iptrunk.iptrunk_sideA_node.subscription.subscription_id,
-            subscription.iptrunk.iptrunk_sideB_node.subscription.subscription_id,
+            subscription.iptrunk.iptrunk_sides[0].iptrunk_side_node.subscription.subscription_id,
+            subscription.iptrunk.iptrunk_sides[1].iptrunk_side_node.subscription.subscription_id,
         ]:
             routers[str(router_id)] = router_description
 
     NewRouterEnum = Choice("Select a new router", zip(routers.keys(), routers.items()))  # type: ignore
+    ReplacedSide = Choice(
+            "Select the side of the IP trunk to be replaced",
+            [  # type: ignore
+                (str(subscription.iptrunk.iptrunk_sides[0].iptrunk_side_node.subscription.subscription_id),
+                 subscription.iptrunk.iptrunk_sides[0].iptrunk_side_node.subscription.description),
+                (str(subscription.iptrunk.iptrunk_sides[1].iptrunk_side_node.subscription.subscription_id),
+                 subscription.iptrunk.iptrunk_sides[1].iptrunk_side_node.subscription.description),
+            ],
+        )
 
     class LagMemberList(UniqueConstrainedList[str]):
-        min_items = len(subscription.iptrunk.iptrunk_sideA_ae_members)
-        max_items = len(subscription.iptrunk.iptrunk_sideA_ae_members)
+        min_items = len(subscription.iptrunk.iptrunk_sides[0].iptrunk_side_ae_members)
+        max_items = len(subscription.iptrunk.iptrunk_sides[1].iptrunk_side_ae_members)
 
     class ModifyIptrunkForm(FormPage):
         class Config:
             title = (
                 f"Subscription {subscription.iptrunk.geant_s_sid} from "
-                f"{subscription.iptrunk.iptrunk_sideA_node.router_fqdn} to "
-                f"{subscription.iptrunk.iptrunk_sideB_node.router_fqdn}"
+                f"{subscription.iptrunk.iptrunk_sides[0].iptrunk_side_node.router_fqdn} to "
+                f"{subscription.iptrunk.iptrunk_sides[1].iptrunk_side_node.router_fqdn}"
             )
 
-        replace_side = Choice(
-            "Select the side of the IP trunk to be replaced",
-            [  # type: ignore
-                (subscription.iptrunk.iptrunk_sideA_node.router_fqdn, subscription.iptrunk.iptrunk_sideA_node),
-                (subscription.iptrunk.iptrunk_sideB_node.router_fqdn, subscription.iptrunk.iptrunk_sideB_node),
-            ],
-        )
+        replace_side: ReplacedSide  # type: ignore
         new_node: NewRouterEnum  # type: ignore
         new_lag_interface: str
-        new_lag_member_interfaces = LagMemberList
+        new_lag_member_interfaces: LagMemberList
 
         @validator("new_lag_interface", allow_reuse=True, pre=True, always=True)
         def lag_interface_proper_name(cls, new_lag_name: str) -> str | NoReturn:
@@ -84,4 +88,15 @@ def temp_test_step(subscription: Iptrunk) -> State:
     target=Target.MODIFY,
 )
 def migrate_iptrunk() -> StepList:
-    return init >> store_process_subscription(Target.MODIFY) >> unsync >> temp_test_step >> resync >> done
+    return (
+            init
+            >> store_process_subscription(Target.MODIFY)
+            >> unsync
+            >> temp_test_step
+            # >> set ISIS to 9000
+            # >> wait confirm
+            # >> disable config
+            # >>
+            >> resync
+            >> done
+    )