Finished release 3.3.

Changelog.md

 # Changelog
 
+# [3.3] - 2025-05-07
+- Fix import L3 core services bug
+
 # [3.2] - 2025-05-02
 - Allow running the Edge Port modification workflow on Juniper routers.
 - Update labels of IAS flavors.

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 = {

setup.py

@@ -4,7 +4,7 @@ from setuptools import find_packages, setup
 
 setup(
     name="geant-service-orchestrator",
-    version="3.2",
+    version="3.3",
     author="GÉANT Orchestration and Automation Team",
     author_email="goat@geant.org",
     description="GÉANT Service Orchestrator",

test/cli/test_imports.py

@@ -31,6 +31,7 @@ from gso.products.product_types.router import Router
 from gso.utils.helpers import generate_unique_vc_id, iso_from_ipv4
 from gso.utils.shared_enums import Vendor
 from gso.utils.types.interfaces import PhysicalPortCapacity
+from gso.workflows.l3_core_service.shared import L3_PRODUCT_NAMES
 
 
 ##############
@@ -293,13 +294,10 @@ def edge_port_data(temp_file, faker, router_subscription_factory, partner_factor
 
 @pytest.fixture()
 def l3_core_service_data(temp_file, faker, partner_factory, edge_port_subscription_factory):
-    def _l3_core_service_data(**kwargs):
-        extra_ias_data = {
-            "ias_flavor": IASFlavor.IAS_PS_OPT_OUT,
-        }
+    def _l3_core_service_data(product_name: ProductName, **kwargs):
         l3_core_service_data = {
             "partner": partner_factory()["name"],
-            "product_name": ProductName.IAS.value,
+            "product_name": product_name,
             "service_binding_ports": [
                 {
                     "edge_port": str(edge_port_subscription_factory().subscription_id),
@@ -399,14 +397,15 @@ def l3_core_service_data(temp_file, faker, partner_factory, edge_port_subscripti
                 },
             ],
         }
+        # Only include ias_flavor if product is IAS
+        if product_name == ProductName.IAS.value:
+            l3_core_service_data["ias_flavor"] = IASFlavor.IAS_PS_OPT_OUT
+
         l3_core_service_data.update(**kwargs)
 
         temp_file.write_text(
             json.dumps([
-                [
-                    l3_core_service_data,
-                    extra_ias_data,
-                ]
+                l3_core_service_data,
             ])
         )
         return {"path": str(temp_file)}
@@ -683,17 +682,21 @@ def test_import_edge_port_with_invalid_partner(mock_start_process, mock_sleep, e
     assert mock_start_process.call_count == 0
 
 
+@pytest.mark.parametrize("product_name", L3_PRODUCT_NAMES)
 @patch("gso.cli.imports.time.sleep")
 @patch("gso.cli.imports.start_process")
-def test_import_l3_core_service_success(mock_start_process, mock_sleep, l3_core_service_data, capfd):
-    import_l3_core_service(l3_core_service_data()["path"])
+def test_import_l3_core_service_success(mock_start_process, mock_sleep, l3_core_service_data, capfd, product_name):
+    import_l3_core_service(l3_core_service_data(product_name)["path"])
     assert mock_start_process.call_count == 1
 
 
+@pytest.mark.parametrize("product_name", L3_PRODUCT_NAMES)
 @patch("gso.cli.imports.time.sleep")
 @patch("gso.cli.imports.start_process")
-def test_import_l3_core_service_with_invalid_partner(mock_start_process, mock_sleep, l3_core_service_data, capfd):
-    broken_data = l3_core_service_data(partner="INVALID")
+def test_import_l3_core_service_with_invalid_partner(
+    mock_start_process, mock_sleep, l3_core_service_data, capfd, product_name
+):
+    broken_data = l3_core_service_data(partner="INVALID", product_name=product_name)
     import_l3_core_service(broken_data["path"])
 
     captured_output, _ = capfd.readouterr()
@@ -701,13 +704,15 @@ def test_import_l3_core_service_with_invalid_partner(mock_start_process, mock_sl
     assert mock_start_process.call_count == 0
 
 
+@pytest.mark.parametrize("product_name", L3_PRODUCT_NAMES)
 @patch("gso.cli.imports.time.sleep")
 @patch("gso.cli.imports.start_process")
 def test_import_l3_core_service_with_invalid_edge_port(
-    mock_start_process, mock_sleep, faker, l3_core_service_data, edge_port_subscription_factory, capfd
+    mock_start_process, mock_sleep, faker, l3_core_service_data, edge_port_subscription_factory, capfd, product_name
 ):
     fake_uuid = faker.uuid4()
     broken_data = l3_core_service_data(
+        product_name=product_name,
         service_binding_ports=[
             {
                 "edge_port": fake_uuid,
@@ -790,7 +795,7 @@ def test_import_l3_core_service_with_invalid_edge_port(
                     "bfd_multiplier": faker.pyint(),
                 },
             },
-        ]
+        ],
     )
     import_l3_core_service(broken_data["path"])
 

test/workflows/l3_core_service/test_create_imported_l3_core_service.py

@@ -13,9 +13,6 @@ from test.workflows import assert_complete, extract_state, run_workflow
 
 @pytest.mark.parametrize("product_name", L3_PRODUCT_NAMES)
 def test_create_imported_l3_core_service_success(faker, partner_factory, edge_port_subscription_factory, product_name):
-    extra_ias_data = {
-        "ias_flavor": IASFlavor.IASGWS,
-    }
     creation_form_input_data = {
         "partner": partner_factory()["name"],
         "product_name": product_name,
@@ -73,11 +70,9 @@ def test_create_imported_l3_core_service_success(faker, partner_factory, edge_po
             }
         ],
     }
-
-    input_data = (
-        [creation_form_input_data, extra_ias_data] if product_name == ProductName.IAS else [creation_form_input_data]
-    )
-    result, _, _ = run_workflow(f"{L3_CREAT_IMPORTED_WF_MAP[product_name]}", input_data)
+    if product_name == ProductName.IAS:
+        creation_form_input_data["ias_flavor"] = IASFlavor.IASGWS
+    result, _, _ = run_workflow(f"{L3_CREAT_IMPORTED_WF_MAP[product_name]}", creation_form_input_data)
     state = extract_state(result)
     subscription = SubscriptionModel.from_subscription(state["subscription_id"])
     assert_complete(result)

test/workflows/l3_core_service/test_import_l3_core_service.py

@@ -13,8 +13,9 @@ def test_import_l3_core_service_success(l3_core_service_subscription_factory, pr
     imported_subscription = l3_core_service_subscription_factory(product_name=product_name, is_imported=True)
     assert imported_subscription.product.name == PRODUCT_IMPORTED_MAP[product_name]
     imported_l3_core_service = str(imported_subscription.subscription_id)
-
-    result, _, _ = run_workflow(L3_IMPORT_WF_MAP[product_name], [{"subscription_id": imported_l3_core_service}])
+    result, _, _ = run_workflow(
+        L3_IMPORT_WF_MAP[f"Imported {product_name}"], [{"subscription_id": imported_l3_core_service}]
+    )
     assert_complete(result)
 
     subscription = SubscriptionModel.from_subscription(imported_l3_core_service)