Add create_vrf WF.

gso/workflows/vrf/__init__.py

+""":term:`VRF` Virtual Routing and Forwarding."""

gso/workflows/vrf/create_vrf.py

+"""A creation workflow for adding a new virtual routing and forwarding (:term:`VRF`) service."""
+
+from orchestrator.forms import FormPage
+from orchestrator.targets import Target
+from orchestrator.types import FormGenerator, State, SubscriptionLifecycle, UUIDstr
+from orchestrator.workflow import StepList, begin, done, step, workflow
+from orchestrator.workflows.steps import resync, set_status, store_process_subscription
+from orchestrator.workflows.utils import wrap_create_initial_input_form
+from pydantic import ConfigDict
+from pydantic_forms.validators import ReadOnlyField
+
+from gso.products.product_types.vrf import VRFInactive
+from gso.services.partners import get_partner_by_name
+from gso.utils.types.tt_number import TTNumber
+from gso.workflows.shared import create_summary_form
+
+
+def initial_input_form_generator(product_name: str) -> FormGenerator:
+    """Gather information about the new router from the operator."""
+
+    class CreateVRFForm(FormPage):
+        model_config = ConfigDict(title=product_name)
+
+        tt_number: TTNumber
+        partner: ReadOnlyField("GEANT", default_type=str)  # type: ignore[valid-type]
+        vrf_name: str
+        route_distinguisher: str
+        route_target: str
+
+    user_input = yield CreateVRFForm
+    user_input = user_input.model_dump()
+    summary_fields = ["vrf_name", "route_distinguisher", "route_target"]
+    yield from create_summary_form(user_input, product_name, summary_fields)
+
+    return user_input
+
+
+@step("Create subscription")
+def create_subscription(product: UUIDstr, partner: str) -> State:
+    """Create a new :term:`VRF` subscription."""
+    subscription = VRFInactive.from_product_id(product, get_partner_by_name(partner)["partner_id"])
+
+    return {
+        "subscription": subscription,
+        "subscription_id": subscription.subscription_id,
+    }
+
+
+@step("Initialize subscription")
+def initialize_subscription(
+    subscription: VRFInactive,
+    vrf_name: str,
+    route_distinguisher: str,
+    route_target: str,
+) -> State:
+    """Initialise the subscription object in the service database."""
+    subscription.vrf.vrf_name = vrf_name
+    subscription.vrf.route_distinguisher = route_distinguisher
+    subscription.vrf.route_target = route_target
+    subscription.description = f"VRF {vrf_name}"
+
+    return {"subscription": subscription}
+
+
+@workflow(
+    "Create VRF",
+    initial_input_form=wrap_create_initial_input_form(initial_input_form_generator),
+    target=Target.CREATE,
+)
+def create_vrf() -> StepList:
+    """Create a virtual routing and forwarding (:term:`VRF`) service."""
+    return (
+        begin
+        >> create_subscription
+        >> store_process_subscription(Target.CREATE)
+        >> initialize_subscription
+        >> set_status(SubscriptionLifecycle.ACTIVE)
+        >> resync
+        >> done
+    )