Skip to content
Snippets Groups Projects
Commit 9f4a6fba authored by JORGE SASIAIN's avatar JORGE SASIAIN Committed by Neda Moeini
Browse files

NAT-244: apply merge review suggestions

parent 0a708d56
No related branches found
No related tags found
1 merge request!94Feature/Netbox integration terminate ip trunk
......@@ -67,7 +67,7 @@ class NetboxClient:
return self.netbox.dcim.devices.get(name=device_name)
def get_interfaces_by_device(self, device_name: str, speed: str) -> list[Interfaces]:
"""Get all interfaces of a device by name and speed that are free and not allocated."""
"""Get all interfaces of a device by name and speed that are not reserved and not allocated."""
device = self.get_device_by_name(device_name)
return list(
self.netbox.dcim.interfaces.filter(device_id=device.id, enabled=False, mark_connected=False, speed=speed)
......@@ -201,7 +201,6 @@ class NetboxClient:
iface.description = description
iface.save()
return iface
def reserve_interface(self, device_name: str, iface_name: str) -> Interfaces:
......
......@@ -11,7 +11,7 @@ from orchestrator.workflows.utils import wrap_modify_initial_input_form
from gso.products.product_blocks.router import RouterVendor
from gso.products.product_types.iptrunk import Iptrunk
from gso.services import infoblox, provisioning_proxy
from gso.services.netbox_client import NetBoxClient
from gso.services.netbox_client import NetboxClient
from gso.services.provisioning_proxy import pp_interaction
from gso.utils.helpers import set_isis_to_90000
......@@ -61,31 +61,17 @@ def deprovision_ip_trunk_real(subscription: Iptrunk, process_id: UUIDstr, tt_num
@step("Remove IP Trunk from NetBox")
def remove_iptrunk_from_netbox(subscription: Iptrunk) -> State:
_router_sideA = subscription.iptrunk.iptrunk_sides[0].iptrunk_side_node
_router_sideB = subscription.iptrunk.iptrunk_sides[1].iptrunk_side_node
router_sideA = _router_sideA.router_fqdn
router_sideB = _router_sideB.router_fqdn
router_sideA_vendor = _router_sideA.router_vendor
router_sideB_vendor = _router_sideB.router_vendor
sideA_members = subscription.iptrunk.iptrunk_sides[0].iptrunk_side_ae_members
sideB_members = subscription.iptrunk.iptrunk_sides[1].iptrunk_side_ae_members
sideA_ae_iface = subscription.iptrunk.iptrunk_sides[0].iptrunk_side_ae_iface
sideB_ae_iface = subscription.iptrunk.iptrunk_sides[1].iptrunk_side_ae_iface
# Remove physical interfaces from LAGs
if router_sideA_vendor == RouterVendor.NOKIA:
for sideA_member in sideA_members:
NetBoxClient().clear_interface(router_sideA, sideA_member)
if router_sideB_vendor == RouterVendor.NOKIA:
for sideB_member in sideB_members:
NetBoxClient().clear_interface(router_sideB, sideB_member)
# Delete LAGs
if router_sideA_vendor == RouterVendor.NOKIA:
NetBoxClient().delete_interface(router_sideA, sideA_ae_iface)
if router_sideB_vendor == RouterVendor.NOKIA:
NetBoxClient().delete_interface(router_sideB, sideB_ae_iface)
def free_interfaces_in_netbox(subscription: Iptrunk) -> State:
for side in [0, 1]:
router = subscription.iptrunk.iptrunk_sides[side].iptrunk_side_node
router_fqdn = router.router_fqdn
if router.router_vendor == RouterVendor.NOKIA:
nbclient = NetboxClient()
# Remove physical interfaces from LAGs
for member in subscription.iptrunk.iptrunk_sides[side].iptrunk_side_ae_members:
nbclient.free_interface(router_fqdn, member.interface_name)
# Delete LAGs
nbclient.delete_interface(router_fqdn, subscription.iptrunk.iptrunk_sides[side].iptrunk_side_ae_iface)
return {"subscription": subscription}
......@@ -126,7 +112,7 @@ def terminate_iptrunk() -> StepList:
>> store_process_subscription(Target.TERMINATE)
>> unsync
>> run_config_steps(config_steps)
>> remove_iptrunk_from_netbox
>> free_interfaces_in_netbox
>> run_ipam_steps(ipam_steps)
>> set_status(SubscriptionLifecycle.TERMINATED)
>> resync
......
......@@ -247,7 +247,7 @@ def test_delete_device(mock_api, device, data_config_filename: PathLike):
def test_get_interfaces_by_device(mock_api, device, interface, data_config_filename: PathLike):
"""Test if a interface is returned for a device."""
# Setup interface speed
speed = 1000
speed = "1000"
# Mock netbox api
mock_api.return_value.dcim.devices.get.return_value = device
......@@ -303,7 +303,7 @@ def test_free_interface(mock_api, device, interface):
# Create a NetboxClient instance
netbox_client = NetboxClient()
# Test clear_interface method on success
# Test free_interface method on success
interface.mark_connected = True
interface.enabled = True
cleared_interface = netbox_client.free_interface(device_name, interface_name)
......
......@@ -28,7 +28,7 @@ class MockedNetboxClient:
def delete_interface(self):
return None
def clear_interface(self):
def free_interface(self):
return self.BaseMockObject(id=1, name="test")
......@@ -41,14 +41,14 @@ def netbox_client_mock():
"gso.services.netbox_client.NetboxClient.get_interface_by_name_and_by_device_id"
) as mock_get_interface_by_name_and_by_device_id,
patch("gso.services.netbox_client.NetboxClient.delete_interface") as mock_delete_interface,
patch("gso.services.netbox_client.NetboxClient.clear_interface") as mock_clear_interface,
patch("gso.services.netbox_client.NetboxClient.free_interface") as mock_free_interface,
):
mock_get_device_by_name.return_value = MockedNetboxClient().get_device_by_name()
mock_get_interface_by_name_and_by_device_id.return_value = (
MockedNetboxClient().get_interface_by_name_and_by_device_id()
)
mock_delete_interface.return_value = MockedNetboxClient().delete_interface()
mock_clear_interface.return_value = MockedNetboxClient().clear_interface()
mock_free_interface.return_value = MockedNetboxClient().free_interface()
yield
......
......@@ -179,6 +179,7 @@ def free_interface(fqdn: str, iface: str) -> None:
freed_iface = NetboxClient().free_interface(fqdn, iface)
click.echo(freed_iface)
@action.command()
@click.option("--fqdn", help="Device name from where to get interface to edit")
@click.option("--iface", help="Interface name to edit")
......@@ -207,11 +208,22 @@ def attach_interface_to_lag(fqdn: str, lag: str, iface: str) -> None:
click.echo(attached_iface)
@action.command()
@click.option("--fqdn", help="Device name from where to get physical interface to detach LAG")
@click.option("--lag", help="LAG name to detach from physical interface")
@click.option("--iface", help="Interface name to detach LAG from")
def detach_interface_from_lag(fqdn: str, lag: str, iface: str) -> None:
click.echo(f"Detaching LAG from physical interface: device={fqdn}, LAG name={lag}, interface name={iface}")
NetboxClient().detach_interfaces_from_lag(fqdn, lag)
click.echo(f"Detached LAG from physical interface: device={fqdn}, LAG name={lag}, interface name={iface}")
action.add_command(reserve_interface)
action.add_command(free_interface)
action.add_command(allocate_interface)
action.add_command(deallocate_interface)
action.add_command(attach_interface_to_lag)
action.add_command(detach_interface_from_lag)
if __name__ == "__main__":
......
0% Loading or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment