Fixed import l3 core services bug

gso/cli/imports.py

@@ -12,6 +12,7 @@ import typer
 import yaml
 from orchestrator.db import db
 from orchestrator.services.processes import start_process
+from orchestrator.services.products import get_product_by_id
 from orchestrator.types import SubscriptionLifecycle
 from pydantic import BaseModel, NonNegativeInt, ValidationError, field_validator, model_validator
 from pydantic_forms.types import UUIDstr
@@ -53,7 +54,7 @@ from gso.utils.types.ip_address import (
     PortNumber,
 )
 from gso.utils.types.virtual_identifiers import VC_ID, VLAN_ID
-from gso.workflows.l3_core_service.shared import L3_CREAT_IMPORTED_WF_MAP, L3ProductNameType
+from gso.workflows.l3_core_service.shared import L3_CREAT_IMPORTED_WF_MAP, L3_IMPORT_WF_MAP, L3ProductNameType
 
 app: typer.Typer = typer.Typer()
 IMPORT_WAIT_MESSAGE = "Waiting for the dust to settle before importing new products..."
@@ -672,18 +673,18 @@ def import_l3_core_service(filepath: str = common_filepath_option) -> None:
     l3_core_service_list = _read_data(Path(filepath))
 
     for l3_core_service in l3_core_service_list:
-        partner = l3_core_service[0]["partner"]
-        product_name = l3_core_service[0]["product_name"]
+        partner = l3_core_service["partner"]
+        product_name = l3_core_service["product_name"]
         typer.echo(f"Creating imported {product_name} for {partner}")
 
         try:
             if product_name == ProductName.IAS.value:
-                initial_data = IASImportModel(**l3_core_service[0], **l3_core_service[1]).model_dump()
+                initial_data = IASImportModel(**l3_core_service).model_dump()
             else:
-                initial_data = L3CoreServiceImportModel(**l3_core_service[0], **l3_core_service[1]).model_dump()
+                initial_data = L3CoreServiceImportModel(**l3_core_service).model_dump()
 
             start_process(L3_CREAT_IMPORTED_WF_MAP[product_name], [initial_data])
-            edge_ports = [sbp["edge_port"] for sbp in l3_core_service[0]["service_binding_ports"]]
+            edge_ports = [sbp["edge_port"] for sbp in l3_core_service["service_binding_ports"]]
             successfully_imported_data.append(edge_ports)
             typer.echo(f"Successfully created imported {product_name} for {partner}")
         except ValidationError as e:
@@ -701,12 +702,13 @@ def import_l3_core_service(filepath: str = common_filepath_option) -> None:
             ProductType.IMPORTED_COPERNICUS,
         ],
         lifecycles=[SubscriptionLifecycle.ACTIVE],
-        includes=["subscription_id"],
+        includes=["subscription_id", "product_id"],
     )
 
-    for subscription_id in imported_products:
-        typer.echo(f"Importing {subscription_id}")
-        start_process("import_l3_core_service", [subscription_id])
+    for subscription in imported_products:
+        product_name = get_product_by_id(subscription["product_id"]).name  # type: ignore[union-attr]
+        typer.echo(f"Importing {product_name} with {subscription["product_id"]}")
+        start_process(L3_IMPORT_WF_MAP[product_name], [{"subscription_id": subscription["subscription_id"]}])
 
     if successfully_imported_data:
         typer.echo("Successfully created imported L3 Core Services:")

gso/workflows/l3_core_service/base_create_imported_l3_core_service.py

@@ -19,48 +19,57 @@ from gso.utils.types.virtual_identifiers import VLAN_ID
 from gso.workflows.l3_core_service.shared import L3ProductNameType
 
 
+class BFDSettingsModel(BaseModel):
+    """BFD settings model for import L3 core services workflows."""
+
+    bfd_enabled: bool = False
+    bfd_interval_rx: int | None = None
+    bfd_interval_tx: int | None = None
+    bfd_multiplier: int | None = None
+
+
+class BaseBGPPeer(BaseModel):
+    """BGP peer model for import L3 core services workflows."""
+
+    bfd_enabled: bool = False
+    has_custom_policies: bool = False
+    authentication_key: str | None
+    multipath_enabled: bool = False
+    send_default_route: bool = False
+    is_passive: bool = False
+    peer_address: IPAddress
+    families: list[IPFamily]
+    is_multi_hop: bool
+    rtbh_enabled: bool
+    prefix_limit: NonNegativeInt | None = None
+    ttl_security: NonNegativeInt | None = None
+
+
+class ServiceBindingPort(BaseModel):
+    """Service binding port model for import L3 core services workflows."""
+
+    edge_port: UUIDstr
+    ap_type: str
+    custom_service_name: str | None = None
+    gs_id: IMPORTED_GS_ID
+    sbp_type: SBPType = SBPType.L3
+    is_tagged: bool = False
+    vlan_id: VLAN_ID
+    custom_firewall_filters: bool = False
+    ipv4_address: IPv4AddressType
+    ipv4_mask: IPv4Netmask
+    ipv6_address: IPv6AddressType
+    ipv6_mask: IPv6Netmask
+    rtbh_enabled: bool = True
+    is_multi_hop: bool = True
+    bgp_peers: list[BaseBGPPeer]
+    v4_bfd_settings: BFDSettingsModel
+    v6_bfd_settings: BFDSettingsModel
+
+
 def initial_input_form_generator() -> FormGenerator:
     """Take all information passed to this workflow by the API endpoint that was called."""
 
-    class BFDSettingsModel(BaseModel):
-        bfd_enabled: bool = False
-        bfd_interval_rx: int | None = None
-        bfd_interval_tx: int | None = None
-        bfd_multiplier: int | None = None
-
-    class BaseBGPPeer(BaseModel):
-        bfd_enabled: bool = False
-        has_custom_policies: bool = False
-        authentication_key: str | None
-        multipath_enabled: bool = False
-        send_default_route: bool = False
-        is_passive: bool = False
-        peer_address: IPAddress
-        families: list[IPFamily]
-        is_multi_hop: bool
-        rtbh_enabled: bool
-        prefix_limit: NonNegativeInt | None = None
-        ttl_security: NonNegativeInt | None = None
-
-    class ServiceBindingPort(BaseModel):
-        edge_port: UUIDstr
-        ap_type: str
-        custom_service_name: str | None = None
-        gs_id: IMPORTED_GS_ID
-        sbp_type: SBPType = SBPType.L3
-        is_tagged: bool = False
-        vlan_id: VLAN_ID
-        custom_firewall_filters: bool = False
-        ipv4_address: IPv4AddressType
-        ipv4_mask: IPv4Netmask
-        ipv6_address: IPv6AddressType
-        ipv6_mask: IPv6Netmask
-        rtbh_enabled: bool = True
-        is_multi_hop: bool = True
-        bgp_peers: list[BaseBGPPeer]
-        v4_bfd_settings: BFDSettingsModel
-        v6_bfd_settings: BFDSettingsModel
-
     class ImportL3CoreServiceForm(SubmitFormPage):
         partner: str
         service_binding_ports: list[ServiceBindingPort]

gso/workflows/l3_core_service/ias/create_imported_ias.py

 """A creation workflow for adding an existing Imported IAS to the service database."""
 
 from orchestrator import workflow
-from orchestrator.forms import FormPage
+from orchestrator.forms import SubmitFormPage
 from orchestrator.targets import Target
 from orchestrator.types import SubscriptionLifecycle
 from orchestrator.workflow import StepList, begin, done, step
@@ -14,24 +14,24 @@ from gso.products.product_types.ias import ImportedIASInactive
 from gso.services.partners import get_partner_by_name
 from gso.services.subscriptions import get_product_id_by_name
 from gso.workflows.l3_core_service.base_create_imported_l3_core_service import (
-    initial_input_form_generator as base_initial_input_form_generator,
-)
-from gso.workflows.l3_core_service.base_create_imported_l3_core_service import (
+    ServiceBindingPort,
     initialize_subscription,
 )
+from gso.workflows.l3_core_service.shared import L3ProductNameType
 
 
 def initial_input_form_generator() -> FormGenerator:
     """Initial input form generator for creating a new imported IAS subscription."""
-    initial_generator = base_initial_input_form_generator()
-    initial_user_input = yield from initial_generator
 
-    # Additional IAS step
-    class IASExtraForm(FormPage):
+    class ImportL3CoreServiceForm(SubmitFormPage):
+        partner: str
+        service_binding_ports: list[ServiceBindingPort]
+        product_name: L3ProductNameType
         ias_flavor: IASFlavor = IASFlavor.IAS_PS_OPT_OUT
 
-    ias_extra = yield IASExtraForm
-    return initial_user_input | ias_extra.model_dump()
+    user_input = yield ImportL3CoreServiceForm
+
+    return user_input.model_dump()
 
 
 @step("Create subscription")

gso/workflows/l3_core_service/shared.py

@@ -52,10 +52,10 @@ L3_MODIFICATION_WF_MAP = {
 
 
 L3_IMPORT_WF_MAP = {
-    ProductName.COPERNICUS: "import_copernicus",
-    ProductName.GEANT_IP: "import_geant_ip",
-    ProductName.IAS: "import_ias",
-    ProductName.LHCONE: "import_lhcone",
+    ProductName.IMPORTED_COPERNICUS: "import_copernicus",
+    ProductName.IMPORTED_GEANT_IP: "import_geant_ip",
+    ProductName.IMPORTED_IAS: "import_ias",
+    ProductName.IMPORTED_LHCONE: "import_lhcone",
 }
 
 L3_VALIDATION_WF_MAP = {