update all workflows and associated tests with the router remodel accordingly

gso/api/v1/imports.py

@@ -78,13 +78,9 @@ class RouterImportModel(BaseModel):
     ts_port: int
     router_vendor: RouterVendor
     router_role: RouterRole
-    is_ias_connected: bool
     router_lo_ipv4_address: ipaddress.IPv4Address
     router_lo_ipv6_address: ipaddress.IPv6Address
     router_lo_iso_address: str
-    router_si_ipv4_network: ipaddress.IPv4Network | None = None
-    router_ias_lt_ipv4_network: ipaddress.IPv4Network | None = None
-    router_ias_lt_ipv6_network: ipaddress.IPv6Network | None = None
 
 
 class IptrunkImportModel(BaseModel):

gso/api/v1/subscriptions.py

@@ -14,7 +14,7 @@ router = APIRouter(prefix="/subscriptions", tags=["Subscriptions"], dependencies
 
 @router.get("/routers", status_code=status.HTTP_200_OK, response_model=list[SubscriptionDomainModelSchema])
 def subscription_routers() -> list[dict[str, Any]]:
-    """Retrieve all active routers subscriptions."""
+    """Retrieve all active router subscriptions."""
     subscriptions = []
     for r in get_active_router_subscriptions():
         subscription = SubscriptionModel.from_subscription(r["subscription_id"])

gso/products/__init__.py

@@ -14,11 +14,12 @@ from gso.products.product_types.site import Site
 
 
 class ProductType(strEnum):
-    SITE = "Site"
-    NOKIA_ROUTER = "Nokia router"
-    JUNIPER_ROUTER = "Juniper router"
-    IP_TRUNK = "IP trunk"
     IAS_GATEWAY = "IAS gateway"
+    IP_TRUNK = "IP trunk"
+    JUNIPER_ROUTER = "Juniper router"
+    NOKIA_ROUTER = "Nokia router"
+    ROUTER = "Router"
+    SITE = "Site"
 
 
 SUBSCRIPTION_MODEL_REGISTRY.update(

gso/services/subscriptions.py

@@ -72,9 +72,7 @@ def get_active_router_subscriptions(includes: list[str] | None = None) -> list[S
     :return: A list of Subscription objects for routers.
     :rtype: list[Subscription]
     """
-    active_nokia_routers = get_active_subscriptions(product_type=ProductType.NOKIA_ROUTER, includes=includes)
-    active_juniper_routers = get_active_subscriptions(product_type=ProductType.JUNIPER_ROUTER, includes=includes)
-    return active_nokia_routers + active_juniper_routers
+    return get_active_subscriptions(product_type="Router", includes=includes)
 
 
 def get_product_id_by_name(product_name: ProductType) -> UUID:

gso/workflows/iptrunk/create_iptrunk.py

@@ -279,7 +279,7 @@ def reserve_interfaces_in_netbox(subscription: IptrunkProvisioning) -> State:
 
     nbclient = NetboxClient()
     for trunk_side in subscription.iptrunk.iptrunk_sides:
-        if trunk_side.iptrunk_side_node.router_vendor == RouterVendor.NOKIA:
+        if get_router_vendor(trunk_side.iptrunk_side_node.owner_subscription_id) == RouterVendor.NOKIA:
             # Create LAG interfaces
             lag_interface: Interfaces = nbclient.create_interface(
                 iface_name=trunk_side.iptrunk_side_ae_iface,
@@ -311,7 +311,7 @@ def reserve_interfaces_in_netbox(subscription: IptrunkProvisioning) -> State:
 def allocate_interfaces_in_netbox(subscription: IptrunkProvisioning) -> State:
     """Allocate the LAG interfaces in NetBox and attach the lag interfaces to the physical interfaces."""
     for trunk_side in subscription.iptrunk.iptrunk_sides:
-        if trunk_side.iptrunk_side_node.router_vendor == RouterVendor.NOKIA:
+        if get_router_vendor(trunk_side.iptrunk_side_node.owner_subscription_id) == RouterVendor.NOKIA:
             for interface in trunk_side.iptrunk_side_ae_members:
                 NetboxClient().allocate_interface(
                     device_name=trunk_side.iptrunk_side_node.router_fqdn,

gso/workflows/iptrunk/migrate_iptrunk.py

@@ -481,13 +481,13 @@ def update_netbox(
 ) -> State:
     new_side = subscription.iptrunk.iptrunk_sides[replace_index]
     nbclient = NetboxClient()
-    if new_side.iptrunk_side_node.router_vendor == RouterVendor.NOKIA:
+    if get_router_vendor(new_side.iptrunk_side_node.owner_subscription_id) == RouterVendor.NOKIA:
         for interface in new_side.iptrunk_side_ae_members:
             nbclient.allocate_interface(
                 device_name=new_side.iptrunk_side_node.router_fqdn,
                 iface_name=interface.interface_name,
             )
-    if old_side_data["iptrunk_side_node"]["router_vendor"] == RouterVendor.NOKIA:
+    if get_router_vendor(old_side_data["iptrunk_side_node"]["owner_subscription_id"]) == RouterVendor.NOKIA:
         # Set interfaces to free
         for iface in old_side_data["iptrunk_side_ae_members"]:
             nbclient.free_interface(old_side_data["iptrunk_side_node"]["router_fqdn"], iface["interface_name"])

gso/workflows/iptrunk/modify_trunk_interface.py

@@ -203,30 +203,30 @@ def provision_ip_trunk_iface_real(
 @step("Update interfaces in Netbox. Reserving interfaces.")
 def update_interfaces_in_netbox(subscription: Iptrunk, removed_ae_members: dict, previous_ae_members: dict) -> State:
     nbclient = NetboxClient()
-    for side in range(0, 2):
-        if subscription.iptrunk.iptrunk_sides[side].iptrunk_side_node.router_vendor == RouterVendor.NOKIA:
-            lag_interface = subscription.iptrunk.iptrunk_sides[side].iptrunk_side_ae_iface
-            router_name = subscription.iptrunk.iptrunk_sides[side].iptrunk_side_node.router_fqdn
+    for index, side in enumerate(subscription.iptrunk.iptrunk_sides):
+        if get_router_vendor(side.iptrunk_side_node.owner_subscription_id) == RouterVendor.NOKIA:
+            lag_interface = side.iptrunk_side_ae_iface
+            router_name = side.iptrunk_side_node.router_fqdn
             # Free removed interfaces
-            for member in removed_ae_members[str(side)]:
+            for member in removed_ae_members[str(index)]:
                 nbclient.free_interface(router_name, member["interface_name"])
             # Attach physical interfaces to LAG
             # Update interface description to subscription ID
             # Reserve interfaces
-            for interface in subscription.iptrunk.iptrunk_sides[side].iptrunk_side_ae_members:
+            for interface in side.iptrunk_side_ae_members:
                 if any(
                     ae_member.get("interface_name") == interface.interface_name
-                    for ae_member in previous_ae_members[str(side)]
+                    for ae_member in previous_ae_members[str(index)]
                 ):
                     continue
                 nbclient.attach_interface_to_lag(
-                    device_name=subscription.iptrunk.iptrunk_sides[side].iptrunk_side_node.router_fqdn,
+                    device_name=side.iptrunk_side_node.router_fqdn,
                     lag_name=lag_interface,
                     iface_name=interface.interface_name,
                     description=str(subscription.subscription_id),
                 )
                 nbclient.reserve_interface(
-                    device_name=subscription.iptrunk.iptrunk_sides[side].iptrunk_side_node.router_fqdn,
+                    device_name=side.iptrunk_side_node.router_fqdn,
                     iface_name=interface.interface_name,
                 )
     return {
@@ -241,23 +241,23 @@ def allocate_interfaces_in_netbox(subscription: Iptrunk, previous_ae_members: di
     attach the lag interfaces to the physical interfaces detach old ones from the LAG.
     """
 
-    for side in range(0, 2):
+    for index, side in enumerate(subscription.iptrunk.iptrunk_sides):
         nbclient = NetboxClient()
-        if subscription.iptrunk.iptrunk_sides[side].iptrunk_side_node.router_vendor == RouterVendor.NOKIA:
-            for interface in subscription.iptrunk.iptrunk_sides[side].iptrunk_side_ae_members:
+        if get_router_vendor(side.iptrunk_side_node.owner_subscription_id) == RouterVendor.NOKIA:
+            for interface in side.iptrunk_side_ae_members:
                 if any(
                     ae_member.get("interface_name") == interface.interface_name
-                    for ae_member in previous_ae_members[str(side)]
+                    for ae_member in previous_ae_members[str(index)]
                 ):
                     continue
                 nbclient.allocate_interface(
-                    device_name=subscription.iptrunk.iptrunk_sides[side].iptrunk_side_node.router_fqdn,
+                    device_name=side.iptrunk_side_node.router_fqdn,
                     iface_name=interface.interface_name,
                 )
             # detach the old interfaces from lag
             nbclient.detach_interfaces_from_lag(
-                device_name=subscription.iptrunk.iptrunk_sides[side].iptrunk_side_node.router_fqdn,
-                lag_name=subscription.iptrunk.iptrunk_sides[side].iptrunk_side_ae_iface,
+                device_name=side.iptrunk_side_node.router_fqdn,
+                lag_name=side.iptrunk_side_ae_iface,
             )
 
     return {"subscription": subscription}

gso/workflows/router/create_router.py

-from ipaddress import IPv4Network, IPv6Network
 from typing import Any
 
 # noinspection PyProtectedMember
@@ -6,7 +5,7 @@ from orchestrator.forms import FormPage
 from orchestrator.forms.validators import Choice
 from orchestrator.targets import Target
 from orchestrator.types import FormGenerator, State, SubscriptionLifecycle, UUIDstr
-from orchestrator.workflow import StepList, conditional, done, init, step, workflow
+from orchestrator.workflow import StepList, done, init, step, workflow
 from orchestrator.workflows.steps import resync, set_status, store_process_subscription
 from orchestrator.workflows.utils import wrap_create_initial_input_form
 from pydantic import validator
@@ -41,7 +40,6 @@ def initial_input_form_generator(product_name: str) -> FormGenerator:
         hostname: str
         ts_port: PortNumber
         router_role: RouterRole
-        is_ias_connected: bool | None = False
 
         @validator("hostname", allow_reuse=True)
         def hostname_must_be_available(cls, hostname: str, **kwargs: dict[str, Any]) -> str:
@@ -76,12 +74,10 @@ def initialize_subscription(
     subscription: RouterInactive,
     hostname: str,
     ts_port: PortNumber,
-    vendor: RouterVendor,
     router_site: str,
     router_role: RouterRole,
 ) -> State:
     subscription.router.router_ts_port = ts_port
-    subscription.vendor = vendor
     subscription.router.router_site = Site.from_subscription(router_site).site
     fqdn = generate_fqdn(
         hostname, subscription.router.router_site.site_name, subscription.router.router_site.site_country_code
@@ -97,31 +93,13 @@ def initialize_subscription(
 
 
 @step("Allocate loopback interfaces in IPAM")
-def ipam_allocate_loopback(subscription: RouterProvisioning, is_ias_connected: bool) -> State:
+def ipam_allocate_loopback(subscription: RouterProvisioning) -> State:
     fqdn = subscription.router.router_fqdn
     loopback_v4, loopback_v6 = infoblox.allocate_host(f"lo0.{fqdn}", "LO", [fqdn], str(subscription.subscription_id))
 
     subscription.router.router_lo_ipv4_address = loopback_v4
     subscription.router.router_lo_ipv6_address = loopback_v6
     subscription.router.router_lo_iso_address = iso_from_ipv4(subscription.router.router_lo_ipv4_address)
-    subscription.router.router_is_ias_connected = is_ias_connected
-
-    return {"subscription": subscription}
-
-
-@step("Allocate IAS connection in IPAM")
-def ipam_allocate_ias_networks(subscription: RouterProvisioning) -> State:
-    fqdn = subscription.router.router_fqdn
-
-    subscription.router.router_si_ipv4_network = infoblox.allocate_v4_network(
-        "SI", f"SI for {fqdn} - {subscription.subscription_id}"
-    )
-    subscription.router.router_ias_lt_ipv4_network = infoblox.allocate_v4_network(
-        "LT_IAS", f"LT for {fqdn} - {subscription.subscription_id}"
-    )
-    subscription.router.router_ias_lt_ipv6_network = infoblox.allocate_v6_network(
-        "LT_IAS", f"LT for {fqdn} - {subscription.subscription_id}"
-    )
 
     return {"subscription": subscription}
 
@@ -164,54 +142,21 @@ def verify_ipam_loopback(subscription: RouterProvisioning) -> State:
     return {"subscription": subscription}
 
 
-@step("Verify IPAM resources for IAS/LT networks")
-def verify_ipam_ias(subscription: RouterProvisioning) -> State:
-    si_ipv4_network = infoblox.find_network_by_cidr(IPv4Network(subscription.router.router_si_ipv4_network))
-    ias_lt_ipv4_network = infoblox.find_network_by_cidr(IPv4Network(subscription.router.router_ias_lt_ipv4_network))
-    ias_lt_ipv6_network = infoblox.find_network_by_cidr(IPv6Network(subscription.router.router_ias_lt_ipv6_network))
-
-    new_state = {}
-
-    if not si_ipv4_network or str(subscription.subscription_id) not in si_ipv4_network.comment:
-        new_state = {
-            "ipam_si_warning": f"SI IPv4 network expected at {subscription.router.router_si_ipv4_network}, "
-            f"but it was not found or misconfigured, please investigate and adjust if necessary."
-        }
-    if not ias_lt_ipv4_network or str(subscription.subscription_id) not in ias_lt_ipv4_network.comment:
-        new_state = new_state | {
-            "ipam_ias_lt_ipv4_warning": "IAS/LT IPv4 network expected at "
-            f"{subscription.router.router_ias_lt_ipv4_network}, but it was not found or misconfigured, please "
-            "investigate and adjust if necessary."
-        }
-    if not ias_lt_ipv6_network or str(subscription.subscription_id) not in ias_lt_ipv6_network.comment:
-        new_state = new_state | {
-            "ipam_ias_lt_ipv6_warning": f"IAS/LT IPv6 network expected at "
-            f"{subscription.router.router_ias_lt_ipv6_network}, but it was not found or misconfigured, please "
-            "investigate and adjust if necessary."
-        }
-
-    return new_state
-
-
 @workflow(
     "Create router",
     initial_input_form=wrap_create_initial_input_form(initial_input_form_generator),
     target=Target.CREATE,
 )
 def create_router() -> StepList:
-    should_allocate_ias = conditional(lambda state: state["is_ias_connected"])
-
     return (
         init
         >> create_subscription
         >> store_process_subscription(Target.CREATE)
         >> initialize_subscription
         >> ipam_allocate_loopback
-        >> should_allocate_ias(ipam_allocate_ias_networks)
         >> pp_interaction(provision_router_dry)
         >> pp_interaction(provision_router_real)
         >> verify_ipam_loopback
-        >> should_allocate_ias(verify_ipam_ias)
         >> create_netbox_device
         >> set_status(SubscriptionLifecycle.ACTIVE)
         >> resync

gso/workflows/router/terminate_router.py

@@ -39,21 +39,6 @@ def deprovision_loopback_ips(subscription: Router) -> dict:
     return {"subscription": subscription}
 
 
-@step("Deprovision SI interface network from IPAM")
-def deprovision_si_ips(subscription: Router) -> dict:
-    infoblox.delete_network(ipaddress.IPv4Network(subscription.router.router_si_ipv4_network))
-
-    return {"subscription": subscription}
-
-
-@step("Deprovision IAS LT interfaces from IPAM")
-def deprovision_lt_ips(subscription: Router) -> dict:
-    infoblox.delete_network(ipaddress.IPv4Network(subscription.router.router_ias_lt_ipv4_network))
-    infoblox.delete_network(ipaddress.IPv6Network(subscription.router.router_ias_lt_ipv6_network))
-
-    return {"subscription": subscription}
-
-
 @step("Remove configuration from router")
 def remove_config_from_router() -> None:
     #  FIXME: Add actual content
@@ -76,17 +61,12 @@ def remove_device_from_netbox(subscription: Router) -> dict[str, Router]:
 def terminate_router() -> StepList:
     run_ipam_steps = conditional(lambda state: state["clean_up_ipam"])
     run_config_steps = conditional(lambda state: state["remove_configuration"])
-    run_ias_removal = conditional(lambda state: state["subscription"]["router"]["router_is_ias_connected"])
-
-    ipam_steps = (
-        init >> deprovision_loopback_ips >> run_ias_removal(deprovision_si_ips) >> run_ias_removal(deprovision_lt_ips)
-    )
 
     return (
         init
         >> store_process_subscription(Target.TERMINATE)
         >> unsync
-        >> run_ipam_steps(ipam_steps)
+        >> run_ipam_steps(deprovision_loopback_ips)
         >> run_config_steps(remove_config_from_router)
         >> remove_device_from_netbox
         >> set_status(SubscriptionLifecycle.TERMINATED)

gso/workflows/tasks/import_router.py

@@ -35,10 +35,11 @@ def _get_site_by_name(site_name: str) -> Site:
 @step("Create subscription")
 def create_subscription(customer: str, router_vendor: RouterVendor) -> State:
     customer_id = get_customer_by_name(customer)["id"]
+    product_id: UUID
     if router_vendor == RouterVendor.NOKIA:
-        product_id: UUID = subscriptions.get_product_id_by_name(ProductType.NOKIA_ROUTER)
+        product_id = subscriptions.get_product_id_by_name(ProductType.NOKIA_ROUTER)
     elif router_vendor == RouterVendor.JUNIPER:
-        product_id: UUID = subscriptions.get_product_id_by_name(ProductType.JUNIPER_ROUTER)
+        product_id = subscriptions.get_product_id_by_name(ProductType.JUNIPER_ROUTER)
     else:
         raise ValueError(f"Unknown router vendor: {router_vendor}")
 
@@ -61,13 +62,9 @@ def initial_input_form_generator() -> FormGenerator:
         ts_port: int
         router_vendor: RouterVendor
         router_role: RouterRole
-        is_ias_connected: bool
         router_lo_ipv4_address: ipaddress.IPv4Address
         router_lo_ipv6_address: ipaddress.IPv6Address
         router_lo_iso_address: str
-        router_si_ipv4_network: ipaddress.IPv4Network | None = None
-        router_ias_lt_ipv4_network: ipaddress.IPv4Network | None = None
-        router_ias_lt_ipv6_network: ipaddress.IPv6Network | None = None
 
     user_input = yield ImportRouter
 
@@ -81,13 +78,9 @@ def initialize_subscription(
     ts_port: PortNumber,
     router_site: str,
     router_role: router_pb.RouterRole,
-    is_ias_connected: bool | None = None,
     router_lo_ipv4_address: ipaddress.IPv4Address | None = None,
     router_lo_ipv6_address: ipaddress.IPv6Address | None = None,
     router_lo_iso_address: str | None = None,
-    router_si_ipv4_network: ipaddress.IPv4Network | None = None,
-    router_ias_lt_ipv4_network: ipaddress.IPv4Network | None = None,
-    router_ias_lt_ipv6_network: ipaddress.IPv6Network | None = None,
 ) -> State:
     subscription.router.router_ts_port = ts_port
     subscription.router.router_site = _get_site_by_name(router_site).site
@@ -100,13 +93,9 @@ def initialize_subscription(
     subscription.router.router_role = router_role
     subscription.router.router_access_via_ts = True
     subscription.description = f"Router {fqdn}"
-    subscription.router.router_is_ias_connected = is_ias_connected
     subscription.router.router_lo_ipv4_address = router_lo_ipv4_address
     subscription.router.router_lo_ipv6_address = router_lo_ipv6_address
     subscription.router.router_lo_iso_address = router_lo_iso_address
-    subscription.router.router_si_ipv4_network = router_si_ipv4_network
-    subscription.router.router_ias_lt_ipv4_network = router_ias_lt_ipv4_network
-    subscription.router.router_ias_lt_ipv6_network = router_ias_lt_ipv6_network
 
     subscription = router.RouterProvisioning.from_other_lifecycle(subscription, SubscriptionLifecycle.PROVISIONING)
 

test/fixtures.py

@@ -79,12 +79,8 @@ def nokia_router_subscription_factory(site_subscription_factory, faker):
         router_lo_ipv4_address=None,
         router_lo_ipv6_address=None,
         router_lo_iso_address=None,
-        router_si_ipv4_network=None,
-        router_ias_lt_ipv4_network=None,
-        router_ias_lt_ipv6_network=None,
         router_role=RouterRole.PE,
         router_site=None,
-        router_is_ias_connected=True,
         status: SubscriptionLifecycle | None = None,
     ) -> UUIDstr:
         description = description or faker.text(max_nb_chars=30)
@@ -94,9 +90,6 @@ def nokia_router_subscription_factory(site_subscription_factory, faker):
         router_lo_ipv4_address = router_lo_ipv4_address or ipaddress.IPv4Address(faker.ipv4())
         router_lo_ipv6_address = router_lo_ipv6_address or ipaddress.IPv6Address(faker.ipv6())
         router_lo_iso_address = router_lo_iso_address or faker.word()
-        router_si_ipv4_network = router_si_ipv4_network or faker.ipv4_network()
-        router_ias_lt_ipv4_network = router_ias_lt_ipv4_network or faker.ipv4_network()
-        router_ias_lt_ipv6_network = router_ias_lt_ipv6_network or faker.ipv6_network()
         router_site = router_site or site_subscription_factory()
 
         product_id = subscriptions.get_product_id_by_name(ProductType.NOKIA_ROUTER)
@@ -107,12 +100,8 @@ def nokia_router_subscription_factory(site_subscription_factory, faker):
         router_subscription.router.router_lo_ipv4_address = router_lo_ipv4_address
         router_subscription.router.router_lo_ipv6_address = router_lo_ipv6_address
         router_subscription.router.router_lo_iso_address = router_lo_iso_address
-        router_subscription.router.router_si_ipv4_network = router_si_ipv4_network
-        router_subscription.router.router_ias_lt_ipv4_network = router_ias_lt_ipv4_network
-        router_subscription.router.router_ias_lt_ipv6_network = router_ias_lt_ipv6_network
         router_subscription.router.router_role = router_role
         router_subscription.router.router_site = Site.from_subscription(router_site).site
-        router_subscription.router.router_is_ias_connected = router_is_ias_connected
 
         router_subscription = SubscriptionModel.from_other_lifecycle(router_subscription, SubscriptionLifecycle.ACTIVE)
         router_subscription.description = description

test/imports/test_imports.py

@@ -107,7 +107,6 @@ def router_data(faker, site_data):
         "router_site": site_data["site_name"],
         "ts_port": 1234,
         "customer": "GÉANT",
-        "is_ias_connected": True,
         "router_lo_ipv4_address": mock_ipv4,
         "router_lo_ipv6_address": faker.ipv6(),
         "router_lo_iso_address": iso_from_ipv4(mock_ipv4),

test/workflows/iptrunk/test_migrate_iptrunk.py

@@ -40,7 +40,7 @@ def test_migrate_iptrunk_success(
     mock_provision_ip_trunk,
     mock_migrate_ip_trunk,
     iptrunk_subscription_factory,
-        nokia_router_subscription_factory,
+    nokia_router_subscription_factory,
     faker,
     data_config_filename: PathLike,
 ):

test/workflows/router/test_create_router.py

@@ -5,7 +5,7 @@ from infoblox_client import objects
 
 from gso.products import ProductType, Site
 from gso.products.product_blocks.router import RouterRole
-from gso.products.product_types.router import Router, RouterVendor
+from gso.products.product_types.router import Router
 from gso.services.crm import customer_selector, get_customer_by_name
 from gso.services.subscriptions import get_product_id_by_name
 from test.workflows import (
@@ -27,9 +27,7 @@ def router_creation_input_form_data(site_subscription_factory, faker):
         "router_site": router_site,
         "hostname": faker.pystr(),
         "ts_port": faker.pyint(),
-        "vendor": RouterVendor.NOKIA,
         "router_role": faker.random_choices(elements=(RouterRole.P, RouterRole.PE, RouterRole.AMT), length=1)[0],
-        "is_ias_connected": True,
     }
 
 
@@ -37,17 +35,11 @@ def router_creation_input_form_data(site_subscription_factory, faker):
 @patch("gso.workflows.router.create_router.provisioning_proxy.provision_router")
 @patch("gso.workflows.router.create_router.NetboxClient.create_device")
 @patch("gso.workflows.router.create_router.infoblox.hostname_available")
-@patch("gso.workflows.router.create_router.infoblox.find_network_by_cidr")
 @patch("gso.workflows.router.create_router.infoblox.find_host_by_fqdn")
-@patch("gso.workflows.router.create_router.infoblox.allocate_v6_network")
-@patch("gso.workflows.router.create_router.infoblox.allocate_v4_network")
 @patch("gso.workflows.router.create_router.infoblox.allocate_host")
 def test_create_nokia_router_success(
     mock_allocate_host,
-    mock_allocate_v4_network,
-    mock_allocate_v6_network,
     mock_find_host_by_fqdn,
-    mock_find_network_by_cidr,
     mock_hostname_available,
     mock_netbox_create_device,
     mock_provision_router,
@@ -59,7 +51,6 @@ def test_create_nokia_router_success(
     product_id = get_product_id_by_name(ProductType.NOKIA_ROUTER)
     mock_site = Site.from_subscription(router_creation_input_form_data["router_site"]).site
     mock_v4 = faker.ipv4()
-    mock_v4_net = faker.ipv4_network()
     mock_v6 = faker.ipv6()
     mock_fqdn = (
         f"{router_creation_input_form_data['hostname']}.{mock_site.site_name.lower()}."
@@ -67,8 +58,6 @@ def test_create_nokia_router_success(
     )
     mock_hostname_available.return_value = True
     mock_allocate_host.return_value = str(mock_v4), str(mock_v6)
-    mock_allocate_v4_network.return_value = mock_v4_net
-    mock_allocate_v6_network.return_value = faker.ipv6_network()
 
     #  Run workflow
     initial_router_data = [{"product": product_id}, router_creation_input_form_data]
@@ -91,13 +80,6 @@ def test_create_nokia_router_success(
         ],
         name=mock_fqdn,
     )
-    mock_find_network_by_cidr.return_value = objects.NetworkV4(
-        connector=None,
-        comment=subscription_id,
-        network=str(mock_v4_net),
-        network_view="default",
-        cidr=str(mock_v4_net),
-    )
 
     for _ in range(2):
         result, step_log = assert_pp_interaction_success(result, process_stat, step_log)
@@ -113,9 +95,7 @@ def test_create_nokia_router_success(
     assert mock_provision_router.call_count == 2
     assert mock_netbox_create_device.call_count == 1
     assert mock_find_host_by_fqdn.call_count == 1
-    assert mock_find_network_by_cidr.call_count == 3
-    for error in ["ipam_warning", "ipam_si_warning", "ipam_ias_lt_ipv4_warning", "ipam_ias_lt_ipv6_warning"]:
-        assert error not in state
+    assert "ipam_warning" not in state
 
 
 @pytest.mark.workflow

test/workflows/router/test_terminate_router.py

@@ -14,13 +14,11 @@ def router_termination_input_form_data(site_subscription_factory, faker):
 @pytest.mark.workflow
 @patch("gso.workflows.router.terminate_router.NetboxClient.delete_device")
 @patch("gso.workflows.router.terminate_router.infoblox.delete_host_by_ip")
-@patch("gso.workflows.router.terminate_router.infoblox.delete_network")
 def test_terminate_router_success(
-    mock_delete_network,
     mock_delete_host_by_ip,
     mock_delete_device,
     router_termination_input_form_data,
-        nokia_router_subscription_factory,
+    nokia_router_subscription_factory,
     faker,
     data_config_filename,
 ):
@@ -37,6 +35,5 @@ def test_terminate_router_success(
     subscription = Router.from_subscription(subscription_id)
 
     assert "terminated" == subscription.status
-    assert mock_delete_network.call_count == 3
     assert mock_delete_device.call_count == 1
     assert mock_delete_host_by_ip.call_count == 1

tox.ini

@@ -37,4 +37,4 @@ commands =
     sh -c "if [ $SKIP_ALL_TESTS -eq 1 ]; then echo 'Skipping coverage report'; else coverage report --fail-under 80; fi"
 
 allowlist_externals =
-    sh
\ No newline at end of file
+    sh