Update router creation workflow to create a Kentik device

gso/services/kentik_client.py

 """The Kentik service is used for external interactions with Kentik."""
 
 import logging
-from typing import Any
+from typing import Any, Literal
 
 import requests
 from pydantic import BaseModel
@@ -39,7 +39,9 @@ class KentikClient:
             "Content-Type": "application/json",
         }
 
-    def _send_request(self, method, endpoint: str, data: dict[str, Any] | None = None) -> Response:
+    def _send_request(
+        self, method: Literal["GET", "POST", "PUT", "DELETE"], endpoint: str, data: dict[str, Any] | None = None
+    ) -> Response:
         url = self.config.api_base + endpoint
         logger.debug("Kentik - Sending request", extra={"method": method, "endpoint": url, "form_data": data})
         result = requests.request(method, url, json=data, headers=self.headers)

gso/workflows/router/create_router.py

@@ -18,6 +18,7 @@ from gso.products.product_blocks.router import RouterRole
 from gso.products.product_types.router import RouterInactive, RouterProvisioning
 from gso.products.product_types.site import Site
 from gso.services import infoblox, subscriptions
+from gso.services.kentik_client import KentikClient, NewKentikDevice
 from gso.services.lso_client import lso_interaction
 from gso.services.netbox_client import NetboxClient
 from gso.services.partners import get_partner_by_name
@@ -143,6 +144,46 @@ def create_netbox_device(subscription: RouterInactive) -> State:
     return {"subscription": subscription}
 
 
+@step("Create Kentik device")
+def create_kentik_device(subscription: RouterInactive) -> State:
+    """Create a new device in Kentik."""
+    if not (
+        subscription.router.router_site
+        and subscription.router.router_site.site_name
+        and subscription.router.router_site.site_tier
+        and subscription.router.router_fqdn
+    ):
+        msg = "Router object is missing required properties."
+        raise ProcessFailureError(msg)
+
+    kentik_client = KentikClient()
+    kentik_site = kentik_client.get_site_by_name(subscription.router.router_site.site_name)
+
+    if not kentik_site:
+        msg = f"Site could not be found in Kentik: {subscription.router.router_site.site_name}"
+        raise ProcessFailureError(msg)
+
+    site_tier = subscription.router.router_site.site_tier
+    new_device = NewKentikDevice(
+        device_name=subscription.router.router_fqdn,
+        device_description=str(subscription.subscription_id),
+        sending_ips=[str(subscription.router.router_lo_ipv4_address)],
+        site_tier=site_tier,
+        site_id=kentik_site["id"],
+        device_snmp_ip=str(subscription.router.router_lo_ipv4_address),
+        device_bgp_flowspec=False,
+        device_bgp_neighbor_ip=str(subscription.router.router_lo_ipv4_address),
+        device_bgp_neighbor_ip6=str(subscription.router.router_lo_ipv6_address),
+    )
+    kentik_device = kentik_client.create_device(new_device)
+
+    if "error" in kentik_device or "kentik_error" in kentik_device:
+        raise ProcessFailureError(str(kentik_device))
+
+    kentik_device.pop("custom_column_data", None)
+    return {"kentik_device": kentik_device}
+
+
 @step("Verify IPAM resources for loopback interface")
 def verify_ipam_loopback(subscription: RouterInactive) -> None:
     """Validate the :term:`IPAM` resources for the loopback interface.
@@ -272,6 +313,7 @@ def create_router() -> StepList:
         >> prompt_insert_in_ims
         >> prompt_insert_in_radius
         >> router_is_nokia(create_netbox_device)
+        >> create_kentik_device
         >> lso_interaction(run_checks_after_base_config)
         >> set_status(SubscriptionLifecycle.PROVISIONING)
         >> create_new_sharepoint_checklist

test/workflows/router/test_create_router.py

@@ -42,7 +42,9 @@ def router_creation_input_form_data(site_subscription_factory, faker):
 @patch("gso.workflows.router.create_router.infoblox.find_host_by_fqdn")
 @patch("gso.workflows.router.create_router.infoblox.allocate_host")
 @patch("gso.workflows.router.create_router.SharePointClient")
+@patch("gso.workflows.router.create_router.KentikClient")
 def test_create_nokia_router_success(
+    mock_kentik_client,
     mock_sharepoint_client,
     mock_allocate_host,
     mock_find_host_by_fqdn,
@@ -113,6 +115,7 @@ def test_create_nokia_router_success(
     assert mock_netbox_create_device.call_count == 1
     assert mock_find_host_by_fqdn.call_count == 1
     assert mock_sharepoint_client.call_count == 1
+    assert mock_kentik_client.call_count == 1
     assert "ipam_warning" not in state