Various updates

reformat some code
add PyYAML type hints
update the use of the LibreNMS client in the iBGP mesh workflow
update the conftest sample configuration fixture

gso/services/librenms_client.py

 """The LibreNMS module interacts with the inventory management system of :term:`GAP`."""
+
 import logging
 from http import HTTPStatus
 from importlib import metadata

gso/services/subscriptions.py

@@ -88,7 +88,7 @@ def get_active_router_subscriptions(
 
 
 def get_active_iptrunk_subscriptions(
-        includes: list[str] | None = None,
+    includes: list[str] | None = None,
 ) -> list[SubscriptionType]:
     """Retrieve active subscriptions specifically for IP trunks.
 
@@ -113,10 +113,15 @@ def get_active_trunks_that_terminate_on_router(subscription_id: UUIDstr) -> list
     :return: A list of IP trunk subscriptions
     :rtype: list[SubscriptionTable]
     """
-    return query_in_use_by_subscriptions(UUID(subscription_id)).join(ProductTable).filter(
-        ProductTable.product_type == "Iptrunk",
-        SubscriptionTable.status == "active",
-    ).all()
+    return (
+        query_in_use_by_subscriptions(UUID(subscription_id))
+        .join(ProductTable)
+        .filter(
+            ProductTable.product_type == "Iptrunk",
+            SubscriptionTable.status == "active",
+        )
+        .all()
+    )
 
 
 def get_product_id_by_name(product_name: ProductType) -> UUID:

gso/workflows/router/update_ibgp_mesh.py

+"""Update iBGP mesh workflow. Adds a new P router to the mesh of PE routers in the network."""
+
 from typing import Any
 
 from orchestrator.forms import FormPage
@@ -13,9 +15,15 @@ from gso.products.product_types.router import Router
 from gso.services import librenms_client, provisioning_proxy, subscriptions
 from gso.services.provisioning_proxy import indifferent_pp_interaction, pp_interaction
 from gso.services.subscriptions import get_active_trunks_that_terminate_on_router
+from gso.utils.helpers import SNMPVersion
 
 
 def initial_input_form_generator(subscription_id: UUIDstr) -> FormGenerator:
+    """Show a confirmation window before running the workflow.
+
+    Does not allow for user input, but does run a validation check. The workflow is only allowed to run, if the router
+    already is connected by at least one IP trunk.
+    """
     subscription = Router.from_subscription(subscription_id)
 
     class AddBGPSessionForm(FormPage):
@@ -37,14 +45,17 @@ def initial_input_form_generator(subscription_id: UUIDstr) -> FormGenerator:
 
 @step("Calculate list of all active PE routers")
 def calculate_pe_router_list() -> State:
-    all_routers = [Router.from_subscription(r["subscription_id"]) for r in
-                   subscriptions.get_active_router_subscriptions()]
+    """Calculate a list of all active PE routers in the network."""
+    all_routers = [
+        Router.from_subscription(r["subscription_id"]) for r in subscriptions.get_active_router_subscriptions()
+    ]
     all_pe_routers = [router for router in all_routers if router.router.router_role == RouterRole.PE]
 
     return {"pe_router_list": all_pe_routers}
 
 
 def _generate_pe_inventory(pe_router_list: list[Router]) -> dict[str, Any]:
+    """Generate an Ansible-compatible inventory for executing playbooks. Contains all active PE routers."""
     return {
         "_meta": {
             "vars": {
@@ -52,19 +63,19 @@ def _generate_pe_inventory(pe_router_list: list[Router]) -> dict[str, Any]:
                     "lo4": router.router.router_lo_ipv4_address,
                     "lo6": router.router.router_lo_ipv6_address,
                     "vendor": router.router.vendor,
-                },
-            } for router in pe_router_list
+                }
+                for router in pe_router_list
+            },
         },
         "all": {
-            "hosts": {
-                router.router.router_fqdn: None for router in pe_router_list
-            },
+            "hosts": {router.router.router_fqdn: None for router in pe_router_list},
         },
     }
 
 
 @step("[DRY RUN] Add P router to iBGP mesh")
 def add_p_to_mesh_dry(subscription: Router, callback_route: str, pe_router_list: list[Router]) -> State:
+    """Perform a dry run of adding the new P router to the PE router mesh."""
     extra_vars = {
         "dry_run": True,
         "subscription": subscription,
@@ -82,6 +93,7 @@ def add_p_to_mesh_dry(subscription: Router, callback_route: str, pe_router_list:
 
 @step("[FOR REAL] Add P router to iBGP mesh")
 def add_p_to_mesh_real(subscription: Router, callback_route: str, pe_router_list: list[Router]) -> State:
+    """Add the P router to the mesh of PE routers."""
     extra_vars = {
         "dry_run": False,
         "subscription": subscription,
@@ -99,6 +111,7 @@ def add_p_to_mesh_real(subscription: Router, callback_route: str, pe_router_list
 
 @step("[DRY RUN] Add all PE routers to P router iBGP table")
 def add_all_pe_to_p_dry(subscription: Router, pe_router_list: list[Router], callback_route: str) -> State:
+    """Perform a dry run of adding the list of all PE routers to the new P router."""
     extra_vars = {
         "dry_run": True,
         "pe_router_list": {
@@ -106,15 +119,14 @@ def add_all_pe_to_p_dry(subscription: Router, pe_router_list: list[Router], call
                 "lo4": router.router.router_lo_ipv4_address,
                 "lo6": router.router.router_lo_ipv6_address,
                 "vendor": router.router.vendor,
-            } for router in pe_router_list
+            }
+            for router in pe_router_list
         },
     }
 
     inventory = {
         "all": {
-            "hosts": {
-                router.router.router_fqdn: None,
-            } for router in pe_router_list
+            "hosts": {router.router.router_fqdn: None for router in pe_router_list},
         },
     }
 
@@ -130,6 +142,7 @@ def add_all_pe_to_p_dry(subscription: Router, pe_router_list: list[Router], call
 
 @step("[FOR REAL] Add all PE routers to P router iBGP table")
 def add_all_pe_to_p_real(subscription: Router, pe_router_list: list[Router], callback_route: str) -> State:
+    """Add the list of all PE routers to the new P router."""
     extra_vars = {
         "dry_run": False,
         "pe_router_list": {
@@ -137,15 +150,14 @@ def add_all_pe_to_p_real(subscription: Router, pe_router_list: list[Router], cal
                 "lo4": router.router.router_lo_ipv4_address,
                 "lo6": router.router.router_lo_ipv6_address,
                 "vendor": router.router.vendor,
-            } for router in pe_router_list
+            }
+            for router in pe_router_list
         },
     }
 
     inventory = {
         "all": {
-            "hosts": {
-                router.router.router_fqdn: None,
-            } for router in pe_router_list
+            "hosts": {router.router.router_fqdn: None for router in pe_router_list},
         },
     }
 
@@ -161,6 +173,7 @@ def add_all_pe_to_p_real(subscription: Router, pe_router_list: list[Router], cal
 
 @step("Verify iBGP session health")
 def check_ibgp_session(subscription: Router, callback_route: str) -> State:
+    """Run a playbook using the provisioning proxy, to check the health of the new iBGP session."""
     inventory = {
         "all": {
             "hosts": {
@@ -181,13 +194,16 @@ def check_ibgp_session(subscription: Router, callback_route: str) -> State:
 
 @step("Add the router to LibreNMS")
 def add_device_to_librenms(subscription: Router) -> State:
-    librenms_client.add_device(subscription)
+    """Add the router as a device to LibreNMS."""
+    client = librenms_client.LibreNMSClient()
+    client.add_device(subscription.router.router_fqdn, SNMPVersion.V2C)
 
     return {"subscription": subscription}
 
 
 @step("Update subscription model")
 def update_subscription_model(subscription: Router) -> State:
+    """Update the database model, such that it should not be reached via :term:`OOB` access anymore."""
     subscription.router.router_access_via_ts = False
 
     return {"subscription": subscription}
@@ -208,17 +224,17 @@ def update_ibgp_mesh() -> StepList:
     * Update the subscription model.
     """
     return (
-            init
-            >> store_process_subscription(Target.MODIFY)
-            >> unsync
-            >> calculate_pe_router_list
-            >> pp_interaction(add_p_to_mesh_dry)
-            >> pp_interaction(add_p_to_mesh_real)
-            >> pp_interaction(add_all_pe_to_p_dry)
-            >> pp_interaction(add_all_pe_to_p_real)
-            >> indifferent_pp_interaction(check_ibgp_session)
-            >> add_device_to_librenms
-            >> update_subscription_model
-            >> resync
-            >> done
+        init
+        >> store_process_subscription(Target.MODIFY)
+        >> unsync
+        >> calculate_pe_router_list
+        >> pp_interaction(add_p_to_mesh_dry)
+        >> pp_interaction(add_p_to_mesh_real)
+        >> pp_interaction(add_all_pe_to_p_dry)
+        >> pp_interaction(add_all_pe_to_p_real)
+        >> indifferent_pp_interaction(check_ibgp_session)
+        >> add_device_to_librenms
+        >> update_subscription_model
+        >> resync
+        >> done
     )

test/conftest.py

@@ -189,6 +189,25 @@ def configuration_data() -> dict:
                     "dns_view": "default",
                 },
             },
+            "MONITORING": {
+                "LIBRENMS": {
+                    "base_url": "http://fake.url.local",
+                    "token": "secret-token",
+                },
+                "SNMP": {
+                    "v2c": {
+                        "community": "fake-community",
+                    },
+                    "v3": {
+                        "authlevel": "AuthPriv",
+                        "authname": "librenms",
+                        "authpass": "<password1>",
+                        "authalgo": "sha",
+                        "cryptopass": "<password2>",
+                        "cryptoalgo": "aes",
+                    },
+                },
+            },
             "PROVISIONING_PROXY": {
                 "scheme": "https",
                 "api_base": "localhost:44444",