Add import WF for L2Circuit- WIP

gso/migrations/versions/2024-10-29_72a4f7aa499d_add_l2circuit_workflows.py

@@ -41,7 +41,13 @@ new_workflows = [
         "target": "CREATE",
         "description": "Create imported Layer 2 Circuit",
         "product_type": "ImportedLayer2Circuit"
-    }
+    },
+    {
+        "name": "import_layer_2_circuit",
+        "target": "MODIFY",
+        "description": "Import Layer 2 Circuit",
+        "product_type": "ImportedLayer2Circuit"
+    },
 ]
 
 

gso/workflows/__init__.py

@@ -130,3 +130,4 @@ LazyWorkflowInstance("gso.workflows.l2_circuit.create_layer_2_circuit", "create_
 LazyWorkflowInstance("gso.workflows.l2_circuit.modify_layer_2_circuit", "modify_layer_2_circuit")
 LazyWorkflowInstance("gso.workflows.l2_circuit.terminate_layer_2_circuit", "terminate_layer_2_circuit")
 LazyWorkflowInstance("gso.workflows.l2_circuit.create_imported_layer_2_circuit", "create_imported_layer_2_circuit")
+LazyWorkflowInstance("gso.workflows.l2_circuit.import_layer_2_circuit", "import_layer_2_circuit")

gso/workflows/l2_circuit/import_layer_2_circuit.py

-"""Import a Layer 2 Circuit."""
+"""A modification workflow for migrating an ImportedLayer2Circuit to an Layer2Circuit subscription."""
+
+from orchestrator.targets import Target
+from orchestrator.types import State, UUIDstr
+from orchestrator.utils.errors import ProcessFailureError
+from orchestrator.workflow import StepList, done, init, step, workflow
+from orchestrator.workflows.steps import resync, store_process_subscription, unsync
+from orchestrator.workflows.utils import wrap_modify_initial_input_form
+
+from gso.products import ProductName
+from gso.products.product_types.layer_2_circuit import ImportedLayer2Circuit, Layer2Circuit, Layer2CircuitServiceType
+from gso.services.subscriptions import get_product_id_by_name
+
+
+@step("Create imported subscription")
+def import_layer_2_circuit_subscription(subscription_id: UUIDstr) -> State:
+    """Take an imported subscription, and turn it into a layer 2 circuit subscription."""
+    old_layer_2_circuit_subscription = ImportedLayer2Circuit.from_subscription(subscription_id)
+    if old_layer_2_circuit_subscription.layer_2_circuit_type == Layer2CircuitServiceType.IMPORTED_GEANT_PLUS:
+        new_subscription_id = get_product_id_by_name(ProductName.GEANT_PLUS)
+    elif old_layer_2_circuit_subscription.layer_2_circuit_type == Layer2CircuitServiceType.IMPORTED_EXPRESSROUTE:
+        new_subscription_id = get_product_id_by_name(ProductName.EXPRESSROUTE)
+    else:
+        msg = f"This {old_layer_2_circuit_subscription.layer_2_circuit_service_type} is already imported."
+        raise ProcessFailureError(message=msg, details=old_layer_2_circuit_subscription)
+    new_subscription = Layer2Circuit.from_other_product(old_layer_2_circuit_subscription, new_subscription_id)  # type: ignore[arg-type]
+
+    return {"subscription": new_subscription}
+
+
+@workflow(
+    "Import Layer 2 Circuit", target=Target.MODIFY, initial_input_form=wrap_modify_initial_input_form(None)
+)
+def import_layer_2_circuit() -> StepList:
+    """Modify an imported subscription into a layer 2 circuit subscription to complete the import."""
+    return (
+        init
+        >> store_process_subscription(Target.MODIFY)
+        >> unsync
+        >> import_layer_2_circuit_subscription
+        >> resync
+        >> done
+    )

test/workflows/l2_circuit/test_import_layer_2_circuit.py

+import pytest
+from orchestrator.types import SubscriptionLifecycle
+
+from gso.products import ProductName
+from gso.products.product_types.layer_2_circuit import Layer2Circuit
+from test.workflows import assert_complete, run_workflow
+
+
+@pytest.mark.workflow()
+@pytest.mark.parametrize(
+    "layer_2_circuit_service_type", [ProductName.IMPORTED_EXPRESSROUTE, ProductName.IMPORTED_GEANT_PLUS]
+)
+def test_import_layer_2_circuit_success(layer_2_circuit_service_type, layer_2_circuit_subscription_factory):
+    imported_layer_2_circuit = layer_2_circuit_subscription_factory(
+        layer_2_circuit_service_type=layer_2_circuit_service_type
+    )
+    result, _, _ = run_workflow(
+        "import_layer_2_circuit", [{"subscription_id": imported_layer_2_circuit}]
+    )
+    subscription = Layer2Circuit.from_subscription(imported_layer_2_circuit)
+
+    assert_complete(result)
+    assert subscription.product.name == layer_2_circuit_service_type
+    assert subscription.status == SubscriptionLifecycle.ACTIVE
+    assert subscription.insync is True