rebase into develop

gso/migrations/versions/2024-04-02_1ec810b289c0_add_orchestrator_2_1_2_migrations.py → gso/migrations/versions/2024-04-20_1ec810b289c0_add_orchestrator_2_1_2_migrations.py

@@ -10,7 +10,7 @@ from orchestrator.migrations.helpers import create_workflow, delete_workflow
 
 # revision identifiers, used by Alembic.
 revision = '1ec810b289c0'
-down_revision = '4ec89ab289c0'
+down_revision = '393acfa175c0'
 branch_labels = None
 # TODO: check it carefuly
 depends_on = '048219045729'  # in this revision, SURF has added a new columns to the workflow table like delted_at, so we need to add a dependency on the revision that added the columns to the workflow table.

gso/products/product_blocks/pop_vlan.py

 """Pop VLAN product block that has all parameters of a subscription throughout its lifecycle."""
 
 from ipaddress import IPv4Network, IPv6Network
-from typing import TypeVar
+from typing import TypeVar, Annotated
 
 from orchestrator.domain.base import ProductBlockModel
-from orchestrator.forms.validators import UniqueConstrainedList
 from orchestrator.types import SubscriptionLifecycle
+from pydantic import AfterValidator
 from pydantic_forms.types import strEnum
+from pydantic_forms.validators import validate_unique_list
 
 from gso.products.product_blocks.lan_switch_interconnect import (
     LanSwitchInterconnectBlock,
@@ -14,7 +15,7 @@ from gso.products.product_blocks.lan_switch_interconnect import (
     LanSwitchInterconnectBlockProvisioning,
 )
 
-T_co = TypeVar("T_co", covariant=True)
+T = TypeVar("T")
 
 
 class LayerPreference(strEnum):
@@ -24,8 +25,7 @@ class LayerPreference(strEnum):
     L3 = "L3"
 
 
-class PortList(UniqueConstrainedList[T_co]):  # type: ignore[type-var]
-    """A list of ports."""
+PortList = Annotated[list[T], AfterValidator(validate_unique_list)]
 
 
 class PopVlanPortBlockInactive(

gso/workflows/iptrunk/modify_trunk_interface.py

@@ -62,7 +62,7 @@ def initialize_ae_members(subscription: Iptrunk, initial_user_input: dict, side_
                 )
             )
 
-        ae_members = Annotated[list[NokiaLAGMember], AfterValidator(validate_unique_list), Len(min_length=iptrunk_number_of_members, max_lenght=iptrunk_number_of_members)]
+        ae_members = Annotated[list[NokiaLAGMember], AfterValidator(validate_unique_list), Len(min_length=iptrunk_number_of_members, max_length=iptrunk_number_of_members)]
     else:
 
         ae_members = Annotated[list[LAGMember], AfterValidator(validate_unique_list), Len(min_length=iptrunk_number_of_members, max_length=iptrunk_number_of_members)]

gso/workflows/tasks/import_iptrunk.py

 """A creation workflow for adding an existing IP trunk to the service database."""
 
 import ipaddress
+from typing import Annotated
 from uuid import uuid4
 
 from orchestrator import workflow
 from orchestrator.forms import FormPage
-from orchestrator.forms.validators import Choice, UniqueConstrainedList
+from orchestrator.forms.validators import Choice
 from orchestrator.targets import Target
 from orchestrator.types import FormGenerator, State, SubscriptionLifecycle
 from orchestrator.workflow import StepList, done, init, step
 from orchestrator.workflows.steps import resync, set_status, store_process_subscription
+from pydantic import AfterValidator, ConfigDict
+from pydantic_forms.validators import validate_unique_list
 
 from gso.products import ProductName
 from gso.products.product_blocks.iptrunk import IptrunkInterfaceBlockInactive, IptrunkType, PhysicalPortCapacity