SDP mesh functions

• Externalized update_sdp_mesh_dry and _real into workflow_steps
• Added _update_sdp_single_pe into workflow_steps
• Added necessary imports

gso/utils/workflow_steps.py

@@ -3,7 +3,6 @@
 import json
 from typing import Any
 
-from gso.products.product_blocks import router
 from orchestrator import inputstep, step
 from orchestrator.config.assignee import Assignee
 from orchestrator.types import State, UUIDstr
@@ -13,10 +12,13 @@ from pydantic_forms.core import FormPage
 from pydantic_forms.types import FormGenerator
 from pydantic_forms.validators import Label
 
+from gso.products.product_blocks import router
+from gso.products.product_blocks.router import RouterRole
 from gso.products.product_types.iptrunk import Iptrunk
 from gso.services import lso_client
 from gso.settings import load_oss_params
 from gso.utils.helpers import generate_inventory_for_active_routers
+from gso.utils.shared_enums import Vendor
 
 
 def _deploy_base_config(
@@ -49,12 +51,11 @@ def _update_sdp_mesh(
     callback_route: str,
     tt_number: str,
     process_id: UUIDstr,
-    dry_run: bool
-    ) -> None:
-    inventory=generate_inventory_for_active_routers(
-        router_role=RouterRole.PE,
-        router_vendor=Vendor.NOKIA,
-        exclude_routers=[subscription["router"]["router_fqdn"]]
+    *,
+    dry_run: bool,
+) -> None:
+    inventory = generate_inventory_for_active_routers(
+        router_role=RouterRole.PE, router_vendor=Vendor.NOKIA, exclude_routers=[subscription["router"]["router_fqdn"]]
     )
     extra_vars = {
         "dry_run": dry_run,
@@ -83,9 +84,10 @@ def _update_sdp_single_pe(
     callback_route: str,
     tt_number: str,
     process_id: UUIDstr,
-    dry_run: bool
-    ) -> None:
-    inventory=subscription["router"]["router_fqdn"]
+    *,
+    dry_run: bool,
+) -> None:
+    inventory = subscription["router"]["router_fqdn"]
     extra_vars = {
         "dry_run": dry_run,
         "subscription": subscription,
@@ -94,7 +96,7 @@ def _update_sdp_single_pe(
         "verb": "update_sdp_mesh",
         "pe_router_list": generate_inventory_for_active_routers(
             router.RouterRole.PE, exclude_routers=[subscription["router"]["router_fqdn"]]
-        )
+        ),
     }
 
     lso_client.execute_playbook(
@@ -103,7 +105,7 @@ def _update_sdp_single_pe(
         inventory=inventory,
         extra_vars=extra_vars,
     )
- 
+
 
 @step("[DRY RUN] Deploy base config")
 def deploy_base_config_dry(
@@ -133,22 +135,17 @@ def deploy_base_config_real(
 
 @step("[DRY RUN] Include new PE into SDP mesh on other Nokia PEs")
 def update_sdp_mesh_dry(
-    subscription: dict[str, Any],
-    callback_route: str,
-    tt_number: str,
-    process_id: UUIDstr
+    subscription: dict[str, Any], callback_route: str, tt_number: str, process_id: UUIDstr
 ) -> State:
-    """Perform a dry run of including new PE router in SDP mesh on other NOKIA PE routers"""
+    """Perform a dry run of including new PE router in SDP mesh on other NOKIA PE routers."""
     _update_sdp_mesh(subscription, tt_number, callback_route, process_id, dry_run=True)
 
     return {"subscription": subscription}
 
+
 @step("[FOR REAL] Include new PE into SDP mesh on other Nokia PEs")
 def update_sdp_mesh_real(
-    subscription: dict[str, Any],
-    callback_route: str,
-    tt_number: str,
-    process_id: UUIDstr
+    subscription: dict[str, Any], callback_route: str, tt_number: str, process_id: UUIDstr
 ) -> State:
     """Include new PE router in SDP mesh on other NOKIA PE routers."""
     _update_sdp_mesh(subscription, tt_number, callback_route, process_id, dry_run=False)
@@ -158,12 +155,9 @@ def update_sdp_mesh_real(
 
 @step("[DRY RUN] Configure SDP on a new PE to all other Nokia PEs")
 def update_sdp_single_pe_dry(
-    subscription: dict[str, Any],
-    callback_route: str,
-    tt_number: str,
-    process_id: UUIDstr
+    subscription: dict[str, Any], callback_route: str, tt_number: str, process_id: UUIDstr
 ) -> State:
-    """Perform a dry run of configuring SDP on a new PE router to all other NOKIA PE routers"""
+    """Perform a dry run of configuring SDP on a new PE router to all other NOKIA PE routers."""
     _update_sdp_single_pe(subscription, tt_number, callback_route, process_id, dry_run=True)
 
     return {"subscription": subscription}
@@ -171,10 +165,7 @@ def update_sdp_single_pe_dry(
 
 @step("[FOR REAL] Configure SDP on a new PE to all other Nokia PEs")
 def update_sdp_single_pe_real(
-    subscription: dict[str, Any],
-    callback_route: str,
-    tt_number: str,
-    process_id: UUIDstr
+    subscription: dict[str, Any], callback_route: str, tt_number: str, process_id: UUIDstr
 ) -> State:
     """Configure SDP on a new PE router to all other NOKIA PE routers."""
     _update_sdp_single_pe(subscription, tt_number, callback_route, process_id, dry_run=False)