Fixed linter errors.

gso/services/subscriptions.py

@@ -20,8 +20,6 @@ from orchestrator.types import SubscriptionLifecycle
 from pydantic_forms.types import UUIDstr
 
 from gso.products import ProductName, ProductType
-from gso.products.product_blocks.router import RouterRole
-from gso.products.product_types.router import Router
 from gso.products.product_types.site import Site
 
 SubscriptionType = dict[str, Any]
@@ -231,4 +229,3 @@ def get_site_by_name(site_name: str) -> Site:
         raise ValueError(msg)
 
     return Site.from_subscription(subscription[0].subscription_id)
-

gso/utils/helpers.py

@@ -55,9 +55,9 @@ def available_interfaces_choices(router_id: UUID, speed: str) -> Choice | None:
 
 
 def available_interfaces_choices_including_current_members(
-        router_id: UUID,
-        speed: str,
-        interfaces: list[IptrunkInterfaceBlock],
+    router_id: UUID,
+    speed: str,
+    interfaces: list[IptrunkInterfaceBlock],
 ) -> Choice | None:
     """Return a list of available interfaces for a given router and speed including the current members.
 
@@ -304,8 +304,8 @@ def generate_fqdn(hostname: str, site_name: str, country_code: str) -> str:
 
 
 def generate_inventory_for_active_routers(
-        router_role: RouterRole,
-        exclude_routers: list[str] | None = None,
+    router_role: RouterRole,
+    exclude_routers: list[str] | None = None,
 ) -> dict:
     """Generate an Ansible-compatible inventory for executing playbooks.
 
@@ -327,15 +327,13 @@ def generate_inventory_for_active_routers(
     return {
         "all": {
             "hosts": {
-                router.router.router_fqdn:
-                    {
-                        "lo4": str(router.router.router_lo_ipv4_address),
-                        "lo6": str(router.router.router_lo_ipv6_address),
-                        "vendor": str(router.router.vendor),
-                    }
+                router.router.router_fqdn: {
+                    "lo4": str(router.router.router_lo_ipv4_address),
+                    "lo6": str(router.router.router_lo_ipv6_address),
+                    "vendor": str(router.router.vendor),
+                }
                 for router in all_routers
-                if router.router.router_role == router_role
-                and router.router.router_fqdn not in exclude_routers
+                if router.router.router_role == router_role and router.router.router_fqdn not in exclude_routers
             }
         }
     }

gso/workflows/router/terminate_router.py

@@ -113,9 +113,7 @@ def remove_device_from_netbox(subscription: Router) -> dict[str, Router]:
 
 
 @step("[DRY RUN] Remove P router from all the PE routers")
-def remove_p_from_all_pe_dry(
-    subscription: Router, callback_route: str, tt_number: str, process_id: UUIDstr
-) -> None:
+def remove_p_from_all_pe_dry(subscription: Router, callback_route: str, tt_number: str, process_id: UUIDstr) -> None:
     """Perform a dry run of removing the terminated router from all the PE routers."""
     extra_vars = {
         "dry_run": True,
@@ -134,9 +132,7 @@ def remove_p_from_all_pe_dry(
 
 
 @step("[REAL RUN] Remove P router from all the PE routers")
-def remove_p_from_all_pe_real(
-    subscription: Router, callback_route: str, tt_number: str, process_id: UUIDstr
-) -> None:
+def remove_p_from_all_pe_real(subscription: Router, callback_route: str, tt_number: str, process_id: UUIDstr) -> None:
     """Perform a real run of removing the terminated router from all the PE routers."""
     extra_vars = {
         "dry_run": False,
@@ -155,9 +151,7 @@ def remove_p_from_all_pe_real(
 
 
 @step("[DRY RUN] Remove PE router from all the PE routers")
-def remove_pe_from_all_pe_dry(
-    subscription: Router, callback_route: str, tt_number: str, process_id: UUIDstr
-) -> None:
+def remove_pe_from_all_pe_dry(subscription: Router, callback_route: str, tt_number: str, process_id: UUIDstr) -> None:
     """Perform a dry run of removing the terminated router from all the PE routers."""
     extra_vars = {
         "dry_run": True,
@@ -171,15 +165,14 @@ def remove_pe_from_all_pe_dry(
         playbook_name="update_ibgp_mesh.yaml",
         callback_route=callback_route,
         inventory=generate_inventory_for_active_routers(
-            RouterRole.PE, exclude_routers=[subscription.router.router_fqdn]),
+            RouterRole.PE, exclude_routers=[subscription.router.router_fqdn]
+        ),
         extra_vars=extra_vars,
     )
 
 
 @step("[REAL RUN] Remove all PE routers from all the PE routers")
-def remove_pe_from_all_pe_real(
-    subscription: Router, callback_route: str, tt_number: str, process_id: UUIDstr
-) -> None:
+def remove_pe_from_all_pe_real(subscription: Router, callback_route: str, tt_number: str, process_id: UUIDstr) -> None:
     """Perform a real run of removing PE router from P router iBGP table."""
     extra_vars = {
         "dry_run": False,
@@ -193,15 +186,14 @@ def remove_pe_from_all_pe_real(
         playbook_name="update_ibgp_mesh.yaml",
         callback_route=callback_route,
         inventory=generate_inventory_for_active_routers(
-            RouterRole.PE, exclude_routers=[subscription.router.router_fqdn]),
+            RouterRole.PE, exclude_routers=[subscription.router.router_fqdn]
+        ),
         extra_vars=extra_vars,
     )
 
 
 @step("[DRY RUN] Remove PE router from all the P routers")
-def remove_pe_from_all_p_dry(
-    subscription: Router, callback_route: str, tt_number: str, process_id: UUIDstr
-) -> None:
+def remove_pe_from_all_p_dry(subscription: Router, callback_route: str, tt_number: str, process_id: UUIDstr) -> None:
     """Perform a dry run of removing PE router from all P routers."""
     extra_vars = {
         "dry_run": True,
@@ -220,9 +212,7 @@ def remove_pe_from_all_p_dry(
 
 
 @step("[REAL RUN] Remove PE router from all P routers")
-def remove_pe_from_all_p_real(
-    subscription: Router, callback_route: str, tt_number: str, process_id: UUIDstr
-) -> None:
+def remove_pe_from_all_p_real(subscription: Router, callback_route: str, tt_number: str, process_id: UUIDstr) -> None:
     """Perform a dry run of removing PE router from all P routers."""
     extra_vars = {
         "dry_run": False,

gso/workflows/router/validate_router.py

@@ -11,11 +11,13 @@ from orchestrator.workflows.steps import resync, store_process_subscription, uns
 from orchestrator.workflows.utils import wrap_modify_initial_input_form
 from pydantic_forms.types import State, UUIDstr
 
+from gso.products.product_blocks.router import RouterRole
 from gso.products.product_types.router import Router
-from gso.services import infoblox, lso_client, subscriptions
+from gso.services import infoblox, lso_client
 from gso.services.librenms_client import LibreNMSClient
 from gso.services.lso_client import anonymous_lso_interaction, execute_playbook
 from gso.services.netbox_client import NetboxClient
+from gso.utils.helpers import generate_inventory_for_active_routers
 from gso.utils.shared_enums import Vendor
 
 
@@ -57,7 +59,7 @@ def verify_p_ibgp(subscription: dict[str, Any], callback_route: str) -> None:
     extra_vars = {
         "dry_run": True,
         "subscription": subscription,
-        "pe_router_list": subscriptions.get_active_pe_router_dict(),
+        "pe_router_list": generate_inventory_for_active_routers(RouterRole.PE)["all"]["hosts"],
         "verb": "verify_p_ibgp",
         "is_verification_workflow": "true",
     }

test/utils/test_helpers.py

@@ -138,4 +138,4 @@ def test_generate_inventory_for_active_routers_with_excluded_router(nokia_router
     router = nokia_router_subscription_factory(router_role=RouterRole.P)
     excluded_routers = [Router.from_subscription(router).router.router_fqdn]
     inventory = generate_inventory_for_active_routers(RouterRole.P, exclude_routers=excluded_routers)
-    assert len(inventory["all"]["hosts"]) == 5   # 6 P routers, the last one is excluded, so 5 P routers are left.
+    assert len(inventory["all"]["hosts"]) == 5  # 6 P routers, the last one is excluded, so 5 P routers are left.