Modify ISIS metric working and tested

gso/migrations/versions/2023-06-21_7694b98571f8_add_iptrunk_isis_modify_workflow.py

+"""add Iptrunk ISIS modify workflow.
+
+Revision ID: 7694b98571f8
+Revises: 84d0ad6e7294
+Create Date: 2023-06-21 08:36:42.824952
+
+"""
+import sqlalchemy as sa
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision = '7694b98571f8'
+down_revision = '84d0ad6e7294'
+branch_labels = None
+depends_on = None
+
+
+from orchestrator.migrations.helpers import create_workflow, delete_workflow
+
+new_workflows = [
+    {
+        "name": "modify_isis_metric",
+        "target": "MODIFY",
+        "description": "Modify IP trunk",
+        "product_type": "Iptrunk"
+    }
+]
+
+
+def upgrade() -> None:
+    conn = op.get_bind()
+    for workflow in new_workflows:
+        create_workflow(conn, workflow)
+
+
+def downgrade() -> None:
+    conn = op.get_bind()
+    for workflow in new_workflows:
+        delete_workflow(conn, workflow["name"])

gso/workflows/__init__.py

@@ -10,8 +10,6 @@ LazyWorkflowInstance("gso.workflows.device.get_facts", "get_facts")
 LazyWorkflowInstance("gso.workflows.iptrunk.create_iptrunk", "create_iptrunk")
 LazyWorkflowInstance("gso.workflows.iptrunk.terminate_iptrunk",
                      "terminate_iptrunk")
-LazyWorkflowInstance("gso.workflows.iptrunk.modify_iptrunk_interface",
-                     "modify_iptrunk_interface")
-LazyWorkflowInstance("gso.workflows.iptrunk.modify_iptrunk_isis_metric",
-                     "modify_iptrunk_isis_metric")
+LazyWorkflowInstance("gso.workflows.iptrunk.modify_isis_metric",
+                     "modify_isis_metric")
 LazyWorkflowInstance("gso.workflows.site.create_site", "create_site")

gso/workflows/iptrunk/modify_isis_metric.py

+from orchestrator.forms import FormPage
+from orchestrator.targets import Target
+from orchestrator.types import FormGenerator, State
+from orchestrator.types import UUIDstr
+from orchestrator.workflow import done, init, step, workflow
+from orchestrator.workflows.steps import resync
+from orchestrator.workflows.steps import store_process_subscription, unsync
+from orchestrator.workflows.utils import wrap_modify_initial_input_form
+
+from gso.products.product_types.iptrunk import (
+    Iptrunk,
+)
+from gso.services import provisioning_proxy
+from gso.services.provisioning_proxy import (
+    confirm_pp_results,
+    await_pp_results,
+)
+
+
+def initial_input_form_generator(subscription_id: UUIDstr) -> FormGenerator:
+    subscription = Iptrunk.from_subscription(subscription_id)
+
+    class ModifyIptrunkForm(FormPage):
+        isis_metric: int = subscription.iptrunk.iptrunk_isis_metric
+
+    user_input = yield ModifyIptrunkForm
+
+    return user_input.dict()
+
+
+@step("Update subscription")
+def modify_iptrunk_subscription(
+    subscription: Iptrunk, isis_metric: int
+) -> State:
+    subscription.iptrunk.iptrunk_isis_metric = isis_metric
+    subscription.description = f"Iptrunk {subscription.description} updated)"
+
+    return {"subscription": subscription}
+
+
+@step("Provision IP trunk ISIS interface [DRY RUN]")
+def provision_ip_trunk_isis_iface_dry(
+    subscription: Iptrunk, process_id: UUIDstr
+) -> State:
+    provisioning_proxy.provision_ip_trunk(
+        subscription, process_id, "isis_interface"
+    )
+
+    return {
+        "subscription": subscription,
+        "label_text": (
+            "This is a dry run for the deployment of a new IP "
+            "trunk ISIS interface. "
+            "Deployment is being taken care of by the "
+            "provisioning proxy, please wait for the results to "
+            "come back before continuing."
+        ),
+    }
+
+
+@step("Provision IP trunk ISIS interface [FOR REAL]")
+def provision_ip_trunk_isis_iface_real(
+    subscription: Iptrunk, process_id: UUIDstr
+) -> State:
+    provisioning_proxy.provision_ip_trunk(
+        subscription, process_id, "isis_interface", False
+    )
+
+    return {
+        "subscription": subscription,
+        "label_text": (
+            "This is a live deployment of a new IP trunk "
+            "ISIS interface. "
+            "Deployment is being taken care of by the "
+            "provisioning proxy, please wait for the results to "
+            "come back before continuing."
+        ),
+    }
+
+
+@workflow(
+    "Modify IP trunk",
+    initial_input_form=wrap_modify_initial_input_form(
+        initial_input_form_generator
+    ),
+    target=Target.MODIFY,
+)
+def modify_isis_metric():
+    return (
+        init
+        >> store_process_subscription(Target.MODIFY)
+        >> unsync
+        >> modify_iptrunk_subscription
+        >> provision_ip_trunk_isis_iface_dry
+        >> await_pp_results
+        >> confirm_pp_results
+        >> provision_ip_trunk_isis_iface_real
+        >> await_pp_results
+        >> confirm_pp_results
+        >> resync
+        >> done
+    )