diff --git a/gso/workflows/geant_ip/modify_geant_ip.py b/gso/workflows/geant_ip/modify_geant_ip.py
index db56802f1199ff114f62522f051657e0d8ec20b1..d568147be1ce54f1206632be5b519a7a87a49f7b 100644
--- a/gso/workflows/geant_ip/modify_geant_ip.py
+++ b/gso/workflows/geant_ip/modify_geant_ip.py
@@ -29,12 +29,12 @@ def initial_input_form_generator(subscription_id: UUIDstr) -> FormGenerator:
subscription = GeantIP.from_subscription(subscription_id)
class AccessPortSelection(BaseModel):
- geant_ip_ep: active_edge_port_selector(partner_id=subscription.customer_id) # type: ignore[valid-type]
+ geant_ip_ep: active_edge_port_selector(partner_id=subscription.customer_id) | str # type: ignore[valid-type]
nren_ap_type: APType
def validate_edge_ports_are_unique(access_ports: list[AccessPortSelection]) -> list[AccessPortSelection]:
"""Verify if interfaces are unique."""
- edge_ports = [port.geant_ip_ep.name for port in access_ports]
+ edge_ports = [str(port.geant_ip_ep) for port in access_ports]
if len(edge_ports) != len(set(edge_ports)):
msg = "Edge Ports must be unique."
raise ValueError(msg)
diff --git a/test/cli/test_imports.py b/test/cli/test_imports.py
index 58352ea7eed4d92acd3d270befd8cbd272dccef6..ea089f91f92ae03f001530ef2a546eab7a608bdb 100644
--- a/test/cli/test_imports.py
+++ b/test/cli/test_imports.py
@@ -209,7 +209,7 @@ def edge_port_data(temp_file, faker, nokia_router_subscription_factory, partner_
"minimum_links": 2,
"geant_ga_id": faker.geant_gid(),
"mac_address": faker.mac_address(),
- "partner": partner_factory(name="GAAR", email=faker.email())["name"],
+ "partner": partner_factory()["name"],
"enable_lacp": True,
"ignore_if_down": False,
"ae_members": [
diff --git a/test/conftest.py b/test/conftest.py
index ccae3c61668deace3f08dfbb4bdd25daadd03270..1083a001572bb3b12617f20cd2fae1b2d7fcd45f 100644
--- a/test/conftest.py
+++ b/test/conftest.py
@@ -120,7 +120,7 @@ class FakerProvider(BaseProvider):
]
def vlan_id(self) -> int:
- return self.generator.random_int(min=1, max=4096)
+ return self.generator.random_int(min=1, max=4095)
@pytest.fixture(scope="session")
@@ -272,15 +272,15 @@ def test_client(fastapi_app):
@pytest.fixture(scope="session")
-def partner_factory():
+def partner_factory(faker):
def _create_partner(
- name: str,
- email: str,
+ name: str | None = None,
+ email: str | None = None,
) -> dict:
return create_partner(
PartnerSchema(
- name=name,
- email=email,
+ name=name or faker.company(),
+ email=email or faker.email(),
)
)
diff --git a/test/fixtures/edge_port_fixtures.py b/test/fixtures/edge_port_fixtures.py
index 58d7220376541dc7c8acd83be7df1890fde9f5f9..274f4052d7d68afe74ecc204a98e3ea1d91e3b1d 100644
--- a/test/fixtures/edge_port_fixtures.py
+++ b/test/fixtures/edge_port_fixtures.py
@@ -38,7 +38,7 @@ def edge_port_subscription_factory(faker, partner_factory, nokia_router_subscrip
ignore_if_down=False,
is_imported=True,
) -> UUIDstr:
- partner = partner or partner_factory(name=faker.company(), email=faker.email())
+ partner = partner or partner_factory()
edge_port_node = Router.from_subscription(nokia_router_subscription_factory()).router
if is_imported:
product_id = subscriptions.get_product_id_by_name(ProductName.EDGE_PORT)
diff --git a/test/fixtures/geant_ip_fixtures.py b/test/fixtures/geant_ip_fixtures.py
index d024e4050a9c4cfed77c012b22c15b881abbc543..084cc927765b3d0d918cdfda55358aa3493b56d0 100644
--- a/test/fixtures/geant_ip_fixtures.py
+++ b/test/fixtures/geant_ip_fixtures.py
@@ -30,6 +30,8 @@ def bgp_session_subscription_factory(faker):
bfd_enabled: bool = True,
multipath_enabled: bool | None = True,
send_default_route: bool | None = True,
+ is_passive: bool | None = False,
+ rtbh_enabled: bool | None = False,
):
return BGPSession.new(
subscription_id=uuid4(),
@@ -43,6 +45,8 @@ def bgp_session_subscription_factory(faker):
is_multi_hop=is_multi_hop,
bfd_interval=bfd_interval,
bfd_multiplier=bfd_multiplier,
+ rtbh_enabled=rtbh_enabled,
+ is_passive=is_passive,
)
return create_bgp_session
@@ -71,19 +75,27 @@ def service_binding_port_factory(faker, bgp_session_subscription_factory, edge_p
ipv6_address=ipv6_address or faker.ipv6(),
custom_firewall_filters=custom_firewall_filters,
geant_sid=geant_sid or faker.geant_sid(),
- sbp_bgp_session_list=sbp_bgp_session_list or [bgp_session_subscription_factory() for _ in range(2)],
- edge_port=edge_port or edge_port_subscription_factory(),
+ sbp_bgp_session_list=sbp_bgp_session_list
+ or [
+ bgp_session_subscription_factory(families=[IPFamily.V4UNICAST]),
+ bgp_session_subscription_factory(families=[IPFamily.V6UNICAST], peer_address=faker.ipv6()),
+ ],
+ edge_port=edge_port or EdgePort.from_subscription(edge_port_subscription_factory()).edge_port,
)
return create_service_binding_port
@pytest.fixture()
-def nren_access_port_factory(faker):
- def create_nren_access_port(nren_ap_type: APType | None = None):
+def nren_access_port_factory(faker, service_binding_port_factory):
+ def create_nren_access_port(
+ nren_ap_type: APType | None = None,
+ service_binding_port: ServiceBindingPort | None = None,
+ ):
return NRENAccessPort.new(
subscription_id=uuid4(),
nren_ap_type=nren_ap_type or random.choice(list(APType)), # noqa: S311
+ geant_ip_sbp=service_binding_port or service_binding_port_factory(),
)
return create_nren_access_port
@@ -93,9 +105,6 @@ def nren_access_port_factory(faker):
def geant_ip_subscription_factory(
faker,
partner_factory,
- edge_port_subscription_factory,
- bgp_session_subscription_factory,
- service_binding_port_factory,
nren_access_port_factory,
):
def create_geant_ip_subscription(
@@ -106,7 +115,7 @@ def geant_ip_subscription_factory(
status: SubscriptionLifecycle | None = None,
) -> UUIDstr:
product_id = subscriptions.get_product_id_by_name(ProductName.GEANT_IP)
- partner = partner or partner_factory(name=faker.company(), email=faker.email())
+ partner = partner or partner_factory()
# Create GEANT IP subscription with product_id and partner details
geant_ip_subscription = GeantIPInactive.from_product_id(
@@ -114,32 +123,17 @@ def geant_ip_subscription_factory(
)
# Default nren_ap_list creation with primary and backup access ports
- nren_ap_list = nren_ap_list or [
- nren_access_port_factory(
- nren_ap_type=APType.PRIMARY, edge_port=edge_port_subscription_factory(partner=partner)
- ),
- nren_access_port_factory(
- nren_ap_type=APType.BACKUP, edge_port=edge_port_subscription_factory(partner=partner)
- ),
+ geant_ip_subscription.geant_ip.geant_ip_ap_list = nren_ap_list or [
+ nren_access_port_factory(nren_ap_type=APType.PRIMARY),
+ nren_access_port_factory(nren_ap_type=APType.BACKUP),
]
- # Assign and save edge port and service binding ports
- for nren_ap in nren_ap_list:
- edge_port = nren_ap.geant_ip_ep
- edge_port.edge_port_sbp_list = [
- service_binding_port_factory(),
- service_binding_port_factory(),
- ]
-
# Update subscription with description, start date, and status
geant_ip_subscription = SubscriptionModel.from_other_lifecycle(
geant_ip_subscription,
SubscriptionLifecycle.ACTIVE,
)
-
- if status:
- geant_ip_subscription.status = status
- geant_ip_subscription.description = description or faker.text(max_nb_chars=30)
+ geant_ip_subscription.description = description or faker.sentence()
geant_ip_subscription.start_date = start_date
geant_ip_subscription.status = status or SubscriptionLifecycle.ACTIVE
geant_ip_subscription.save()
diff --git a/test/workflows/edge_port/test_create_imported_edge_port.py b/test/workflows/edge_port/test_create_imported_edge_port.py
index 21d9f829dffb420a362bec165ad3f889a350a870..6f49436696b7d879f807e92e98889da7cce0ef6d 100644
--- a/test/workflows/edge_port/test_create_imported_edge_port.py
+++ b/test/workflows/edge_port/test_create_imported_edge_port.py
@@ -18,7 +18,7 @@ def imported_edge_port_creation_input_form_data(nokia_router_subscription_factor
"minimum_links": 2,
"geant_ga_id": faker.geant_gid(),
"mac_address": faker.mac_address(),
- "partner": partner_factory(name="GAAR", email=faker.email())["name"],
+ "partner": partner_factory()["name"],
"enable_lacp": True,
"ignore_if_down": False,
"ae_members": [
diff --git a/test/workflows/geant_ip/test_create_geant_ip.py b/test/workflows/geant_ip/test_create_geant_ip.py
index 90e237e9f1f5aabaaa8134acab9fd3a231bd42c6..a93698ed4327cd0f810ee03fbdabc0473233dc8b 100644
--- a/test/workflows/geant_ip/test_create_geant_ip.py
+++ b/test/workflows/geant_ip/test_create_geant_ip.py
@@ -39,7 +39,7 @@ def test_create_geant_ip_success(
base_bgp_peer_input,
data_config_filename,
):
- partner = partner_factory(name=faker.company(), email=faker.email())
+ partner = partner_factory()
product_id = get_product_id_by_name(ProductName.GEANT_IP)
edge_port_a = edge_port_subscription_factory(partner=partner)
diff --git a/test/workflows/geant_ip/test_modify_geant_ip.py b/test/workflows/geant_ip/test_modify_geant_ip.py
new file mode 100644
index 0000000000000000000000000000000000000000..937f95cb9a88bef62e7bcaaed26461a1a786f170
--- /dev/null
+++ b/test/workflows/geant_ip/test_modify_geant_ip.py
@@ -0,0 +1,278 @@
+import pytest
+
+from gso.products.product_blocks.bgp_session import IPFamily
+from gso.products.product_types.geant_ip import GeantIP
+from gso.utils.shared_enums import APType
+from test.workflows import extract_state, run_workflow
+
+
+@pytest.mark.workflow()
+def test_modify_geant_ip_remove_edge_port_success(geant_ip_subscription_factory):
+ subscription_id = geant_ip_subscription_factory()
+ subscription = GeantIP.from_subscription(subscription_id)
+ access_port = subscription.geant_ip.geant_ip_ap_list[0]
+ input_form_data = [
+ {"subscription_id": subscription_id},
+ {
+ "access_ports": [
+ {
+ "geant_ip_ep": str(access_port.geant_ip_sbp.edge_port.owner_subscription_id),
+ "nren_ap_type": APType.LOAD_BALANCED,
+ }
+ ] # The factory generates a subscription with two Access Ports, this will remove the second one.
+ },
+ {},
+ ]
+
+ result, _, _ = run_workflow("modify_geant_ip", input_form_data)
+
+ state = extract_state(result)
+ subscription = GeantIP.from_subscription(state["subscription_id"])
+ assert len(subscription.geant_ip.geant_ip_ap_list) == 1
+ assert subscription.geant_ip.geant_ip_ap_list[0].nren_ap_type == APType.LOAD_BALANCED
+
+
+@pytest.mark.workflow()
+def test_modify_geant_ip_add_new_edge_port_success(
+ geant_ip_subscription_factory, edge_port_subscription_factory, partner_factory, faker
+):
+ partner = partner_factory()
+ new_edge_port = edge_port_subscription_factory(partner=partner)
+ subscription_id = geant_ip_subscription_factory(partner=partner)
+ subscription = GeantIP.from_subscription(subscription_id)
+ input_form_data = [
+ {"subscription_id": subscription_id},
+ {
+ "access_ports": [
+ {
+ "geant_ip_ep": str(port.geant_ip_sbp.edge_port.owner_subscription_id),
+ "nren_ap_type": port.nren_ap_type,
+ }
+ for port in subscription.geant_ip.geant_ip_ap_list
+ ]
+ + [
+ {
+ "geant_ip_ep": str(new_edge_port),
+ "nren_ap_type": APType.BACKUP,
+ }
+ ]
+ },
+ {}, # The existing SBPs are unchanged
+ {},
+ { # Adding configuration for the new SBP
+ "geant_sid": faker.geant_sid(),
+ "vlan_id": faker.vlan_id(),
+ "ipv4_address": faker.ipv4(),
+ "ipv6_address": faker.ipv6(),
+ "v4_bgp_peer": {
+ "authentication_key": faker.password(),
+ "peer_address": faker.ipv4(),
+ },
+ "v6_bgp_peer": {
+ "authentication_key": faker.password(),
+ "peer_address": faker.ipv6(),
+ },
+ },
+ ]
+
+ result, _, _ = run_workflow("modify_geant_ip", input_form_data)
+
+ state = extract_state(result)
+ subscription = GeantIP.from_subscription(state["subscription_id"])
+ assert len(subscription.geant_ip.geant_ip_ap_list) == 3
+
+
+@pytest.mark.workflow()
+def test_modify_geant_ip_modify_edge_port_success(faker, geant_ip_subscription_factory):
+ subscription_id = geant_ip_subscription_factory()
+ subscription = GeantIP.from_subscription(subscription_id)
+ new_sbp_data = [
+ {
+ "geant_sid": faker.geant_sid(),
+ "is_tagged": True,
+ "vlan_id": faker.vlan_id(),
+ "ipv4_address": faker.ipv4(),
+ "ipv6_address": faker.ipv6(),
+ "custom_firewall_filters": True,
+ "v4_bgp_peer": {
+ "bfd_enabled": True,
+ "bfd_interval": faker.pyint(),
+ "bfd_multiplier": faker.pyint(),
+ "has_custom_policies": True,
+ "authentication_key": faker.password(),
+ "multipath_enabled": True,
+ "send_default_route": True,
+ "is_passive": True,
+ "peer_address": faker.ipv4(),
+ "add_v4_multicast": True,
+ },
+ "v6_bgp_peer": {
+ "bfd_enabled": True,
+ "bfd_interval": faker.pyint(),
+ "bfd_multiplier": faker.pyint(),
+ "has_custom_policies": True,
+ "authentication_key": faker.password(),
+ "multipath_enabled": True,
+ "send_default_route": True,
+ "is_passive": True,
+ "peer_address": faker.ipv6(),
+ "add_v6_multicast": True,
+ },
+ },
+ {
+ "geant_sid": faker.geant_sid(),
+ "is_tagged": True,
+ "vlan_id": faker.vlan_id(),
+ "ipv4_address": faker.ipv4(),
+ "ipv6_address": faker.ipv6(),
+ "custom_firewall_filters": True,
+ "v4_bgp_peer": {
+ "bfd_enabled": True,
+ "bfd_interval": faker.pyint(),
+ "bfd_multiplier": faker.pyint(),
+ "has_custom_policies": True,
+ "authentication_key": faker.password(),
+ "multipath_enabled": True,
+ "send_default_route": True,
+ "is_passive": True,
+ "peer_address": faker.ipv4(),
+ "add_v4_multicast": True,
+ },
+ "v6_bgp_peer": {
+ "bfd_enabled": True,
+ "bfd_interval": faker.pyint(),
+ "bfd_multiplier": faker.pyint(),
+ "has_custom_policies": True,
+ "authentication_key": faker.password(),
+ "multipath_enabled": True,
+ "send_default_route": True,
+ "is_passive": True,
+ "peer_address": faker.ipv6(),
+ "add_v6_multicast": True,
+ },
+ },
+ ]
+
+ input_form_data = [
+ {"subscription_id": subscription_id},
+ {
+ "access_ports": [
+ {
+ "geant_ip_ep": str(port.geant_ip_sbp.edge_port.owner_subscription_id),
+ "nren_ap_type": port.nren_ap_type,
+ }
+ for port in subscription.geant_ip.geant_ip_ap_list
+ ]
+ },
+ {**new_sbp_data[0]},
+ {**new_sbp_data[1]},
+ ]
+
+ result, _, _ = run_workflow("modify_geant_ip", input_form_data)
+
+ state = extract_state(result)
+ subscription = GeantIP.from_subscription(state["subscription_id"])
+ assert len(subscription.geant_ip.geant_ip_ap_list) == 2
+
+ for i in range(2):
+ assert subscription.geant_ip.geant_ip_ap_list[i].geant_ip_sbp.geant_sid == new_sbp_data[i]["geant_sid"]
+ assert subscription.geant_ip.geant_ip_ap_list[i].geant_ip_sbp.is_tagged == new_sbp_data[i]["is_tagged"]
+ assert subscription.geant_ip.geant_ip_ap_list[i].geant_ip_sbp.vlan_id == new_sbp_data[i]["vlan_id"]
+ assert (
+ str(subscription.geant_ip.geant_ip_ap_list[i].geant_ip_sbp.ipv4_address) == new_sbp_data[i]["ipv4_address"]
+ )
+ assert (
+ str(subscription.geant_ip.geant_ip_ap_list[i].geant_ip_sbp.ipv6_address) == new_sbp_data[i]["ipv6_address"]
+ )
+ assert (
+ subscription.geant_ip.geant_ip_ap_list[i].geant_ip_sbp.custom_firewall_filters
+ == new_sbp_data[i]["custom_firewall_filters"]
+ )
+
+ assert (
+ subscription.geant_ip.geant_ip_ap_list[i].geant_ip_sbp.sbp_bgp_session_list[0].bfd_enabled
+ == new_sbp_data[i]["v4_bgp_peer"]["bfd_enabled"]
+ )
+ assert (
+ subscription.geant_ip.geant_ip_ap_list[i].geant_ip_sbp.sbp_bgp_session_list[0].bfd_interval
+ == new_sbp_data[i]["v4_bgp_peer"]["bfd_interval"]
+ )
+ assert (
+ subscription.geant_ip.geant_ip_ap_list[i].geant_ip_sbp.sbp_bgp_session_list[0].bfd_multiplier
+ == new_sbp_data[i]["v4_bgp_peer"]["bfd_multiplier"]
+ )
+ assert (
+ subscription.geant_ip.geant_ip_ap_list[i].geant_ip_sbp.sbp_bgp_session_list[0].has_custom_policies
+ == new_sbp_data[i]["v4_bgp_peer"]["has_custom_policies"]
+ )
+ assert (
+ subscription.geant_ip.geant_ip_ap_list[i].geant_ip_sbp.sbp_bgp_session_list[0].authentication_key
+ == new_sbp_data[i]["v4_bgp_peer"]["authentication_key"]
+ )
+ assert (
+ subscription.geant_ip.geant_ip_ap_list[i].geant_ip_sbp.sbp_bgp_session_list[0].multipath_enabled
+ == new_sbp_data[i]["v4_bgp_peer"]["multipath_enabled"]
+ )
+ assert (
+ subscription.geant_ip.geant_ip_ap_list[i].geant_ip_sbp.sbp_bgp_session_list[0].send_default_route
+ == new_sbp_data[i]["v4_bgp_peer"]["send_default_route"]
+ )
+ assert (
+ subscription.geant_ip.geant_ip_ap_list[i].geant_ip_sbp.sbp_bgp_session_list[0].is_passive
+ == new_sbp_data[i]["v4_bgp_peer"]["is_passive"]
+ )
+ assert (
+ str(subscription.geant_ip.geant_ip_ap_list[i].geant_ip_sbp.sbp_bgp_session_list[0].peer_address)
+ == new_sbp_data[i]["v4_bgp_peer"]["peer_address"]
+ )
+ assert (
+ bool(
+ IPFamily.V4MULTICAST
+ in subscription.geant_ip.geant_ip_ap_list[i].geant_ip_sbp.sbp_bgp_session_list[0].families
+ )
+ == new_sbp_data[i]["v4_bgp_peer"]["add_v4_multicast"]
+ )
+
+ assert (
+ subscription.geant_ip.geant_ip_ap_list[i].geant_ip_sbp.sbp_bgp_session_list[1].bfd_enabled
+ == new_sbp_data[i]["v6_bgp_peer"]["bfd_enabled"]
+ )
+ assert (
+ subscription.geant_ip.geant_ip_ap_list[i].geant_ip_sbp.sbp_bgp_session_list[1].bfd_interval
+ == new_sbp_data[i]["v6_bgp_peer"]["bfd_interval"]
+ )
+ assert (
+ subscription.geant_ip.geant_ip_ap_list[i].geant_ip_sbp.sbp_bgp_session_list[1].bfd_multiplier
+ == new_sbp_data[i]["v6_bgp_peer"]["bfd_multiplier"]
+ )
+ assert (
+ subscription.geant_ip.geant_ip_ap_list[i].geant_ip_sbp.sbp_bgp_session_list[1].has_custom_policies
+ == new_sbp_data[i]["v6_bgp_peer"]["has_custom_policies"]
+ )
+ assert (
+ subscription.geant_ip.geant_ip_ap_list[i].geant_ip_sbp.sbp_bgp_session_list[1].authentication_key
+ == new_sbp_data[i]["v6_bgp_peer"]["authentication_key"]
+ )
+ assert (
+ subscription.geant_ip.geant_ip_ap_list[i].geant_ip_sbp.sbp_bgp_session_list[1].multipath_enabled
+ == new_sbp_data[i]["v6_bgp_peer"]["multipath_enabled"]
+ )
+ assert (
+ subscription.geant_ip.geant_ip_ap_list[i].geant_ip_sbp.sbp_bgp_session_list[1].send_default_route
+ == new_sbp_data[i]["v6_bgp_peer"]["send_default_route"]
+ )
+ assert (
+ subscription.geant_ip.geant_ip_ap_list[i].geant_ip_sbp.sbp_bgp_session_list[1].is_passive
+ == new_sbp_data[i]["v6_bgp_peer"]["is_passive"]
+ )
+ assert (
+ str(subscription.geant_ip.geant_ip_ap_list[i].geant_ip_sbp.sbp_bgp_session_list[1].peer_address)
+ == new_sbp_data[i]["v6_bgp_peer"]["peer_address"]
+ )
+ assert (
+ bool(
+ IPFamily.V6MULTICAST
+ in subscription.geant_ip.geant_ip_ap_list[i].geant_ip_sbp.sbp_bgp_session_list[1].families
+ )
+ == new_sbp_data[i]["v6_bgp_peer"]["add_v6_multicast"]
+ )
diff --git a/test/workflows/router/test_modify_connection_stratey.py b/test/workflows/router/test_modify_connection_strategy.py
similarity index 100%
rename from test/workflows/router/test_modify_connection_stratey.py
rename to test/workflows/router/test_modify_connection_strategy.py