Prevent change of Kentik license on non-PE routers

gso/translations/en-GB.json

@@ -48,6 +48,7 @@
         "modify_site": "Modify Site",
         "modify_trunk_interface": "Modify IP Trunk interface",
         "modify_connection_strategy": "Modify connection strategy",
+        "modify_router_kentik_license": "Modify device license in Kentik",
         "terminate_iptrunk": "Terminate IP Trunk",
         "terminate_router": "Terminate Router",
         "terminate_site": "Terminate Site",

gso/workflows/router/modify_kentik_license.py

 """A workflow that modifies the Kentik license of a router."""
 
+import logging
+from typing import Any
+
 from orchestrator.forms import FormPage
 from orchestrator.targets import Target
 from orchestrator.types import FormGenerator, State, UUIDstr
 from orchestrator.utils.errors import ProcessFailureError
 from orchestrator.workflow import StepList, begin, done, step, workflow
-from orchestrator.workflows.steps import resync, store_process_subscription, unsync
+from orchestrator.workflows.steps import store_process_subscription
 from orchestrator.workflows.utils import wrap_modify_initial_input_form
+from pydantic import model_validator
 from pydantic_forms.validators import Choice
 
+from gso.products.product_blocks.router import RouterRole
 from gso.products.product_types.router import Router
 from gso.services.kentik_client import KentikClient
 
+logger = logging.getLogger()
+
 
 def _initial_input_form(subscription_id: UUIDstr) -> FormGenerator:
     router = Router.from_subscription(subscription_id)
+    active_kentik_plans = {str(plan["id"]): plan["name"] for plan in KentikClient().get_plans() if plan["active"]}
+    available_kentik_plans = Choice(
+        "Select a Kentik license",
+        zip(active_kentik_plans.keys(), active_kentik_plans.items(), strict=True),  # type: ignore[arg-type]
+    )
 
     class ModifyKentikLicenseForm(FormPage):
-        @staticmethod
-        def _get_available_plans() -> list[dict[str, str]]:
-            kentik_plans = KentikClient().get_plans()
-            return [{plan["name"]: plan["id"]} for plan in kentik_plans]
+        new_plan_id: available_kentik_plans  # type: ignore[valid-type]
+
+        @model_validator(mode="before")
+        def router_must_be_nokia_p(cls, data: Any) -> Any:
+            if router.router.router_role != RouterRole.PE:
+                msg = "Can only update Kentik license for PE routers."
+                raise ValueError(msg)
 
-        new_plan_id = Choice("Select a Kentik license", _get_available_plans())
+            return data
 
     user_input = yield ModifyKentikLicenseForm
 
@@ -36,10 +51,9 @@ def update_kentik_license(subscription: Router, new_plan_id: int) -> State:
     kentik_device = kentik_client.get_device_by_name(subscription.router.router_fqdn)
     if "id" not in kentik_device:
         msg = "Failed to find Kentik device by name"
-        raise ProcessFailureError(msg, details=kentik_device)
+        raise ProcessFailureError(msg, details=subscription.router.router_fqdn)
 
-    updated_kentik_device = kentik_client.update_device(kentik_device["id"], {"plan_id": new_plan_id})
-    updated_kentik_device.pop("custom_column_data", None)
+    updated_kentik_device = kentik_client.update_device(kentik_device["id"], {"device": {"plan_id": new_plan_id}})
 
     return {"kentik_device": updated_kentik_device}
 
@@ -51,4 +65,4 @@ def update_kentik_license(subscription: Router, new_plan_id: int) -> State:
 )
 def modify_router_kentik_license() -> StepList:
     """Apply a selected Kentik license on an existing PE router."""
-    return begin >> store_process_subscription(Target.MODIFY) >> unsync >> update_kentik_license >> resync >> done
+    return begin >> store_process_subscription(Target.MODIFY) >> update_kentik_license >> done