Skip to content
Snippets Groups Projects
Verified Commit 182e4ac4 authored by Karel van Klink's avatar Karel van Klink :smiley_cat:
Browse files

Add switch fixture and unit test for lan interconnect product

parent 8091226c
No related branches found
No related tags found
No related merge requests found
This commit is part of merge request !300. Comments created here will be created in the context of that merge request.
......@@ -56,6 +56,11 @@ LazyWorkflowInstance("gso.workflows.switch.activate_switch", "activate_switch")
LazyWorkflowInstance("gso.workflows.switch.terminate_switch", "terminate_switch")
LazyWorkflowInstance("gso.workflows.switch.validate_switch", "validate_switch")
# LAN Switch Interconnect workflows
LazyWorkflowInstance(
"gso.workflows.lan_switch_interconnect.create_lan_switch_interconnect", "create_lan_switch_interconnect"
)
# Site workflows
LazyWorkflowInstance("gso.workflows.site.create_site", "create_site")
LazyWorkflowInstance("gso.workflows.site.modify_site", "modify_site")
......
......@@ -48,6 +48,7 @@ from test.fixtures import ( # noqa: F401
service_binding_port_factory,
site_subscription_factory,
super_pop_switch_subscription_factory,
switch_subscription_factory,
)
logging.getLogger("faker.factory").setLevel(logging.WARNING)
......@@ -110,6 +111,12 @@ class FakerProvider(BaseProvider):
def network_interface(self) -> str:
return self.generator.numerify("ge-@#/@#/@#")
def juniper_ae_interface_name(self) -> str:
return self.generator.numerify("ae@#")
def nokia_lag_interface_name(self) -> str:
return self.generator.numerify("lag-@#")
def link_members_juniper(self) -> LAGMemberList[LAGMember]:
iface_amount = self.generator.random_int(min=2, max=5)
interface_names = [f"{prefix}{i}" for prefix in ["xe-1/0/", "ge-3/0/", "xe-2/1/"] for i in range(iface_amount)]
......
......@@ -11,6 +11,7 @@ from test.fixtures.opengear_fixtures import opengear_subscription_factory
from test.fixtures.router_fixtures import router_subscription_factory
from test.fixtures.site_fixtures import site_subscription_factory
from test.fixtures.super_pop_switch_fixtures import super_pop_switch_subscription_factory
from test.fixtures.switch_fixtures import switch_subscription_factory
__all__ = [
"bgp_session_subscription_factory",
......@@ -25,4 +26,5 @@ __all__ = [
"service_binding_port_factory",
"site_subscription_factory",
"super_pop_switch_subscription_factory",
"switch_subscription_factory",
]
import pytest
from orchestrator.db import db
from orchestrator.domain import SubscriptionModel
from orchestrator.types import SubscriptionLifecycle
from pydantic_forms.types import UUIDstr
from gso.products import ProductName
from gso.products.product_blocks.switch import SwitchModel
from gso.products.product_types.site import Site
from gso.products.product_types.switch import SwitchInactive
from gso.services.subscriptions import get_product_id_by_name
from gso.utils.shared_enums import Vendor
from gso.utils.types.ip_address import PortNumber
@pytest.fixture()
def switch_subscription_factory(faker, geant_partner, site_subscription_factory):
def subscription_create(
partner: dict | None = None,
description: str | None = None,
start_date: str | None = "2024-01-01T10:20:30+01:02",
fqdn: str | None = None,
ts_port: PortNumber | None = None,
site: UUIDstr | None = None,
switch_vendor: Vendor | None = None,
switch_model: SwitchModel | None = None,
status: SubscriptionLifecycle | None = None,
*,
is_imported: bool = True,
) -> UUIDstr:
if partner is None:
partner = geant_partner
if is_imported:
product_id = get_product_id_by_name(ProductName.SWITCH)
switch_subscription = SwitchInactive.from_product_id(product_id, partner["partner_id"])
else:
product_id = get_product_id_by_name(ProductName.IMPORTED_SWITCH)
raise NotImplemented
switch_subscription = ImportedSwitchInactive.from_product_id(product_id, partner["partner_id"])
switch_subscription.switch.fqdn = fqdn or faker.domain_name(levels=4)
switch_subscription.switch.ts_port = ts_port or faker.port_number(is_user=True)
switch_subscription.switch.site = site or Site.from_subscription(site_subscription_factory()).site
switch_subscription.switch.switch_vendor = switch_vendor or Vendor.JUNIPER
switch_subscription.switch.switch_model = switch_model or SwitchModel.EX3400
switch_subscription = SubscriptionModel.from_other_lifecycle(switch_subscription, SubscriptionLifecycle.ACTIVE)
switch_subscription.insync = True
switch_subscription.description = description or faker.sentence()
switch_subscription.start_date = start_date
if status:
switch_subscription.status = status
switch_subscription.save()
db.session.commit()
return str(switch_subscription.subscription_id)
return subscription_create
......@@ -64,6 +64,9 @@ class MockedNetboxClient:
def delete_interface():
return None
def get_available_lags_in_range(self):
return self.get_available_lags()
class MockedSharePointClient:
class BaseMockObject:
......
from unittest.mock import patch
import pytest
from orchestrator.types import SubscriptionLifecycle
from gso.products import ProductName
from gso.products.product_blocks.lan_switch_interconnect import LanSwitchInterconnectAddressSpace
from gso.products.product_types.lan_switch_interconnect import LanSwitchInterconnect
from gso.services.subscriptions import get_product_id_by_name
from test.services.conftest import MockedNetboxClient
from test.workflows import assert_complete, extract_state, run_workflow
@pytest.fixture()
def _netbox_client_mock():
# Mock NetboxClient methods
with (
patch("gso.services.netbox_client.NetboxClient.get_device_by_name") as mock_get_device_by_name,
patch("gso.services.netbox_client.NetboxClient.get_available_interfaces") as mock_get_available_interfaces,
patch("gso.services.netbox_client.NetboxClient.get_available_lags_in_range") as mock_available_lags_in_range,
):
mock_get_device_by_name.return_value = MockedNetboxClient().get_device_by_name()
mock_get_available_interfaces.return_value = MockedNetboxClient().get_available_interfaces()
mock_available_lags_in_range.return_value = MockedNetboxClient().get_available_lags_in_range()
yield
@pytest.fixture()
def input_form_data(faker, router_subscription_factory, switch_subscription_factory):
def _generate_form_data(address_space: LanSwitchInterconnectAddressSpace):
return [
{
"product": get_product_id_by_name(ProductName.LAN_SWITCH_INTERCONNECT),
},
{
"tt_number": faker.tt_number(),
"router_side": router_subscription_factory(),
"switch_side": switch_subscription_factory(),
"address_space": address_space,
"description": faker.sentence(),
"minimum_link_count": 2,
"vlan_id": 111, # VLAN ID for new interconnections is always 111
},
{
"router_side_iface": "lag-4",
"router_side_ae_members": faker.link_members_nokia()[:2],
},
{
"switch_side_iface": "ae9",
"switch_side_ae_members": faker.link_members_juniper()[:2],
},
]
return _generate_form_data
@pytest.mark.parametrize(
"address_space", [LanSwitchInterconnectAddressSpace.PRIVATE, LanSwitchInterconnectAddressSpace.PUBLIC]
)
@pytest.mark.workflow()
def test_create_lan_switch_interconnect_success(
address_space,
input_form_data,
_netbox_client_mock, # noqa: PT019
):
initial_data = input_form_data(address_space)
result, _, _ = run_workflow("create_lan_switch_interconnect", initial_data)
assert_complete(result)
state = extract_state(result)
subscription_id = state["subscription_id"]
subscription = LanSwitchInterconnect.from_subscription(subscription_id)
assert subscription.status == SubscriptionLifecycle.ACTIVE
0% Loading or .
You are about to add 0 people to the discussion. Proceed with caution.
Please register or to comment