From 3bbc81a6c9398b9b5b5f3771456da128ea163a02 Mon Sep 17 00:00:00 2001
From: Karel van Klink <karel.vanklink@geant.org>
Date: Mon, 21 Oct 2024 10:58:53 +0200
Subject: [PATCH] Fix unit tests for L3 Core Services
---
gso/cli/imports.py | 40 +++---
...dd_import_edgeport_and_l3_core_service_.py | 10 +-
...6925_add_edge_port_and_l3_core_service_.py | 2 +-
.../product_types/nren_l3_core_service.py | 4 +-
.../create_imported_nren_l3_core_service.py | 18 +--
.../import_nren_l3_core_service.py | 38 ++++--
.../modify_nren_l3_core_service.py | 2 +-
test/cli/test_imports.py | 3 +-
test/conftest.py | 2 +-
test/fixtures/__init__.py | 8 +-
test/fixtures/edge_port_fixtures.py | 3 +-
...es.py => nren_l3_core_service_fixtures.py} | 76 +++++++-----
test/utils/test_helpers.py | 2 +-
.../edge_port/test_create_edge_port.py | 4 +-
.../test_create_imported_edge_port.py | 2 +-
.../edge_port/test_import_edge_port.py | 3 +-
.../edge_port/test_modify_edge_port.py | 2 +-
.../edge_port/test_terminate_edge_port.py | 2 +-
.../edge_port/test_validate_edge_port.py | 2 +-
.../geant_ip/test_import_geant_ip.py | 17 ---
.../geant_ip/test_migrate_geant_ip.py | 46 -------
.../iptrunk/test_activate_iptrunk.py | 2 +-
test/workflows/iptrunk/test_create_iptrunk.py | 3 +-
test/workflows/iptrunk/test_deploy_twamp.py | 2 +-
.../iptrunk/test_modify_isis_metric.py | 2 +-
.../iptrunk/test_modify_trunk_interface.py | 2 +-
.../iptrunk/test_terminate_iptrunk.py | 2 +-
.../__init__.py | 0
...t_create_imported_nren_l3_core_service.py} | 15 +--
.../test_create_nren_l3_core_service.py} | 12 +-
.../test_import_nren_l3_core_service.py | 23 ++++
.../test_migrate_nren_l3_core_service.py | 49 ++++++++
.../test_modify_nren_l3_core_service.py} | 117 ++++++++++--------
.../opengear/test_create_imported_opengear.py | 3 +-
test/workflows/router/test_activate_router.py | 2 +-
test/workflows/router/test_create_router.py | 3 +-
.../router/test_modify_connection_strategy.py | 2 +-
.../router/test_redeploy_base_config.py | 2 +-
.../workflows/router/test_terminate_router.py | 2 +-
.../workflows/router/test_update_ibgp_mesh.py | 2 +-
40 files changed, 305 insertions(+), 226 deletions(-)
rename test/fixtures/{geant_ip_fixtures.py => nren_l3_core_service_fixtures.py} (61%)
delete mode 100644 test/workflows/geant_ip/test_import_geant_ip.py
delete mode 100644 test/workflows/geant_ip/test_migrate_geant_ip.py
rename test/workflows/{geant_ip => nren_l3_core_service}/__init__.py (100%)
rename test/workflows/{geant_ip/test_create_imported_geant_ip.py => nren_l3_core_service/test_create_imported_nren_l3_core_service.py} (84%)
rename test/workflows/{geant_ip/test_create_geant_ip.py => nren_l3_core_service/test_create_nren_l3_core_service.py} (85%)
create mode 100644 test/workflows/nren_l3_core_service/test_import_nren_l3_core_service.py
create mode 100644 test/workflows/nren_l3_core_service/test_migrate_nren_l3_core_service.py
rename test/workflows/{geant_ip/test_modify_geant_ip.py => nren_l3_core_service/test_modify_nren_l3_core_service.py} (57%)
diff --git a/gso/cli/imports.py b/gso/cli/imports.py
index 584cbce8..5415e60c 100644
--- a/gso/cli/imports.py
+++ b/gso/cli/imports.py
@@ -23,6 +23,7 @@ from gso.products.product_blocks.edge_port import EdgePortType, EncapsulationTyp
from gso.products.product_blocks.iptrunk import IptrunkType
from gso.products.product_blocks.router import RouterRole
from gso.products.product_blocks.service_binding_port import VLAN_ID
+from gso.products.product_types.nren_l3_core_service import NRENL3CoreServiceType
from gso.services.partners import (
PartnerEmail,
PartnerName,
@@ -214,11 +215,11 @@ class EdgePortImportModel(BaseModel):
return self
-class GeantIPImportModel(BaseModel):
- """Import GEANT IP model."""
+class NRENL3CoreServiceImportModel(BaseModel):
+ """Import :term:`NREN` L3 Core Service model."""
partner: str
- service_binding_ports: list["GeantIPImportModel.ServiceBindingPort"]
+ service_binding_ports: list["NRENL3CoreServiceImportModel.ServiceBindingPort"]
class BaseBGPPeer(BaseModel):
"""Base BGP Peer model."""
@@ -250,7 +251,7 @@ class GeantIPImportModel(BaseModel):
ipv6_address: IPv6AddressType
rtbh_enabled: bool = True
is_multi_hop: bool = True
- bgp_peers: list["GeantIPImportModel.BaseBGPPeer"]
+ bgp_peers: list["NRENL3CoreServiceImportModel.BaseBGPPeer"]
T = TypeVar(
@@ -262,7 +263,7 @@ T = TypeVar(
OfficeRouterImportModel,
OpenGearImportModel,
EdgePortImportModel,
- GeantIPImportModel,
+ NRENL3CoreServiceImportModel,
)
common_filepath_option = typer.Option(
@@ -519,21 +520,22 @@ def import_partners(file_path: str = typer.Argument(..., help="Path to the CSV f
@app.command()
-def import_geant_ip(filepath: str = common_filepath_option) -> None:
- """Import GEANT IP into GSO."""
+def import_nren_l3_core_service(filepath: str = common_filepath_option) -> None:
+ """Import :term:`NREN` L3 Core Services into :term:`GSO`."""
successfully_imported_data = []
- geant_ip_list = _read_data(Path(filepath))
+ nren_l3_core_service_list = _read_data(Path(filepath))
- for geant_ip in geant_ip_list:
- partner = geant_ip["partner"]
- typer.echo(f"Creating imported GEANT IP for {partner}")
+ for nren_l3_core_service in nren_l3_core_service_list:
+ partner = nren_l3_core_service["partner"]
+ service_type = NRENL3CoreServiceType(nren_l3_core_service["service_type"])
+ typer.echo(f"Creating imported {service_type} for {partner}")
try:
- initial_data = GeantIPImportModel(**geant_ip)
- start_process("create_imported_geant_ip", [initial_data.model_dump()])
- edge_ports = [sbp["edge_port"] for sbp in geant_ip["service_binding_ports"]]
+ initial_data = NRENL3CoreServiceImportModel(**nren_l3_core_service)
+ start_process("create_imported_nren_l3_core_service", [initial_data.model_dump()])
+ edge_ports = [sbp["edge_port"] for sbp in nren_l3_core_service["service_binding_ports"]]
successfully_imported_data.append(edge_ports)
- typer.echo(f"Successfully created imported GEANT IP for {partner}")
+ typer.echo(f"Successfully created imported {service_type} for {partner}")
except ValidationError as e:
typer.echo(f"Validation error: {e}")
@@ -542,14 +544,16 @@ def import_geant_ip(filepath: str = common_filepath_option) -> None:
# Migrate new products from imported to "full" counterpart.
imported_products = get_subscriptions(
- [ProductType.IMPORTED_GEANT_IP], lifecycles=[SubscriptionLifecycle.ACTIVE], includes=["subscription_id"]
+ product_types=[ProductType.IMPORTED_GEANT_IP, ProductType.IMPORTED_IAS],
+ lifecycles=[SubscriptionLifecycle.ACTIVE],
+ includes=["subscription_id"],
)
for subscription_id in imported_products:
typer.echo(f"Importing {subscription_id}")
- start_process("import_geant_ip", [subscription_id])
+ start_process("import_nren_l3_core_service", [subscription_id])
if successfully_imported_data:
- typer.echo("Successfully created imported GEANT IPs:")
+ typer.echo("Successfully created imported NREN L3 Core Services:")
for item in successfully_imported_data:
typer.echo(f"- {item}")
diff --git a/gso/migrations/versions/2024-10-17_7412c5b7ebe4_add_import_edgeport_and_l3_core_service_.py b/gso/migrations/versions/2024-10-17_7412c5b7ebe4_add_import_edgeport_and_l3_core_service_.py
index 6f6b40e9..d3739209 100644
--- a/gso/migrations/versions/2024-10-17_7412c5b7ebe4_add_import_edgeport_and_l3_core_service_.py
+++ b/gso/migrations/versions/2024-10-17_7412c5b7ebe4_add_import_edgeport_and_l3_core_service_.py
@@ -57,31 +57,31 @@ new_workflows = [
{
"name": "create_nren_l3_core_service",
"target": "CREATE",
- "description": "Create G\u00c9ANT IP",
+ "description": "Create NREN L3 Core Service",
"product_type": "NRENL3CoreService"
},
{
"name": "modify_nren_l3_core_service",
"target": "MODIFY",
- "description": "Modify G\u00c9ANT IP",
+ "description": "Modify NREN L3 Core Service",
"product_type": "NRENL3CoreService"
},
{
"name": "create_imported_nren_l3_core_service",
"target": "CREATE",
- "description": "Import G\u00c9ANT IP",
+ "description": "Create imported NREN L3 Core Service",
"product_type": "ImportedNRENL3CoreService"
},
{
"name": "import_nren_l3_core_service",
"target": "MODIFY",
- "description": "Import G\u00c9ANT IP",
+ "description": "Import NREN L3 Core Service",
"product_type": "ImportedNRENL3CoreService"
},
{
"name": "migrate_nren_l3_core_service",
"target": "MODIFY",
- "description": "Migrate G\u00c9ANT IP",
+ "description": "Migrate NREN L3 Core Service",
"product_type": "NRENL3CoreService"
}
]
diff --git a/gso/migrations/versions/2024-10-17_e1659d366925_add_edge_port_and_l3_core_service_.py b/gso/migrations/versions/2024-10-17_e1659d366925_add_edge_port_and_l3_core_service_.py
index 7e027aa7..d033c263 100644
--- a/gso/migrations/versions/2024-10-17_e1659d366925_add_edge_port_and_l3_core_service_.py
+++ b/gso/migrations/versions/2024-10-17_e1659d366925_add_edge_port_and_l3_core_service_.py
@@ -36,7 +36,7 @@ INSERT INTO products (name, description, product_type, tag, status) VALUES ('IAS
INSERT INTO products (name, description, product_type, tag, status) VALUES ('Imported IAS', 'A pre-existing Internet Access Service that is imported into the service database', 'ImportedNRENL3CoreService', 'IMP_IAS', 'active') RETURNING products.product_id
"""))
conn.execute(sa.text("""
-INSERT INTO fixed_inputs (name, value, product_id) VALUES ('nren_l3_core_service_type', 'IAS', (SELECT products.product_id FROM products WHERE products.name IN ('IAS'))), ('nren_l3_core_service_type', 'IMPORTED_IAS', (SELECT products.product_id FROM products WHERE products.name IN ('Imported IAS'))), ('nren_l3_core_service_type', 'GEANT_IP', (SELECT products.product_id FROM products WHERE products.name IN ('GÉANT IP'))), ('nren_l3_core_service_type', 'IMPORTED_GEANT_IP', (SELECT products.product_id FROM products WHERE products.name IN ('Imported GÉANT IP')))
+INSERT INTO fixed_inputs (name, value, product_id) VALUES ('nren_l3_core_service_type', 'IAS', (SELECT products.product_id FROM products WHERE products.name IN ('IAS'))), ('nren_l3_core_service_type', 'IMPORTED IAS', (SELECT products.product_id FROM products WHERE products.name IN ('Imported IAS'))), ('nren_l3_core_service_type', 'GÉANT IP', (SELECT products.product_id FROM products WHERE products.name IN ('GÉANT IP'))), ('nren_l3_core_service_type', 'IMPORTED GÉANT IP', (SELECT products.product_id FROM products WHERE products.name IN ('Imported GÉANT IP')))
"""))
conn.execute(sa.text("""
INSERT INTO product_blocks (name, description, tag, status) VALUES ('EdgePortBlock', 'The product block with all attributes of an Edge Port', 'EP_BLOCK', 'active') RETURNING product_blocks.product_block_id
diff --git a/gso/products/product_types/nren_l3_core_service.py b/gso/products/product_types/nren_l3_core_service.py
index 8717acfe..20aa5ab2 100644
--- a/gso/products/product_types/nren_l3_core_service.py
+++ b/gso/products/product_types/nren_l3_core_service.py
@@ -17,8 +17,10 @@ class NRENL3CoreServiceType(strEnum):
The core services offered include GÉANT IP for R&E access, and the Internet Access Service.
"""
- GEANT_IP = "GEANT_IP"
+ GEANT_IP = "GÉANT IP"
+ IMPORTED_GEANT_IP = "IMPORTED GÉANT IP"
IAS = "IAS"
+ IMPORTED_IAS = "IMPORTED IAS"
class NRENL3CoreServiceInactive(SubscriptionModel, is_base=True):
diff --git a/gso/workflows/nren_l3_core_service/create_imported_nren_l3_core_service.py b/gso/workflows/nren_l3_core_service/create_imported_nren_l3_core_service.py
index b39d094c..0bad4c86 100644
--- a/gso/workflows/nren_l3_core_service/create_imported_nren_l3_core_service.py
+++ b/gso/workflows/nren_l3_core_service/create_imported_nren_l3_core_service.py
@@ -16,7 +16,7 @@ from gso.products.product_blocks.bgp_session import BGPSession, IPFamily
from gso.products.product_blocks.nren_l3_core_service import NRENAccessPortInactive
from gso.products.product_blocks.service_binding_port import VLAN_ID, ServiceBindingPortInactive
from gso.products.product_types.edge_port import EdgePort
-from gso.products.product_types.nren_l3_core_service import ImportedNRENL3CoreServiceInactive
+from gso.products.product_types.nren_l3_core_service import ImportedNRENL3CoreServiceInactive, NRENL3CoreServiceType
from gso.services.partners import get_partner_by_name
from gso.services.subscriptions import get_product_id_by_name
from gso.utils.shared_enums import SBPType
@@ -56,20 +56,24 @@ def initial_input_form_generator() -> FormGenerator:
is_multi_hop: bool = True
bgp_peers: list[BaseBGPPeer]
- class ImportGeantIPForm(FormPage):
+ class ImportNRENL3CoreServiceForm(FormPage):
partner: str
service_binding_ports: list[ServiceBindingPort]
+ nren_l3_core_service_type: NRENL3CoreServiceType
- user_input = yield ImportGeantIPForm
+ user_input = yield ImportNRENL3CoreServiceForm
return user_input.model_dump()
@step("Create subscription")
-def create_subscription(partner: str) -> dict:
+def create_subscription(partner: str, nren_l3_core_service_type: NRENL3CoreServiceType) -> dict:
"""Create a new subscription object in the database."""
partner_id = get_partner_by_name(partner)["partner_id"]
- product_id = get_product_id_by_name(ProductName.IMPORTED_GEANT_IP)
+ if nren_l3_core_service_type == NRENL3CoreServiceType.GEANT_IP:
+ product_id = get_product_id_by_name(ProductName.IMPORTED_GEANT_IP)
+ elif nren_l3_core_service_type == NRENL3CoreServiceType.IAS:
+ product_id = get_product_id_by_name(ProductName.IMPORTED_IAS)
subscription = ImportedNRENL3CoreServiceInactive.from_product_id(product_id, partner_id)
return {"subscription": subscription, "subscription_id": subscription.subscription_id}
@@ -96,13 +100,13 @@ def initialize_subscription(subscription: ImportedNRENL3CoreServiceInactive, ser
)
)
- subscription.description = "GEANT IP service"
+ subscription.description = f"{subscription.product} service"
return {"subscription": subscription}
@workflow(
- "Import GÉANT IP",
+ "Create imported NREN L3 Core Service",
initial_input_form=initial_input_form_generator,
target=Target.CREATE,
)
diff --git a/gso/workflows/nren_l3_core_service/import_nren_l3_core_service.py b/gso/workflows/nren_l3_core_service/import_nren_l3_core_service.py
index 4d314997..39519246 100644
--- a/gso/workflows/nren_l3_core_service/import_nren_l3_core_service.py
+++ b/gso/workflows/nren_l3_core_service/import_nren_l3_core_service.py
@@ -2,26 +2,44 @@
from orchestrator.targets import Target
from orchestrator.types import State, UUIDstr
+from orchestrator.utils.errors import ProcessFailureError
from orchestrator.workflow import StepList, done, init, step, workflow
from orchestrator.workflows.steps import resync, store_process_subscription, unsync
from orchestrator.workflows.utils import wrap_modify_initial_input_form
from gso.products import ProductName
-from gso.products.product_types.nren_l3_core_service import ImportedNRENL3CoreService, NRENL3CoreService
+from gso.products.product_types.nren_l3_core_service import (
+ ImportedNRENL3CoreService,
+ NRENL3CoreService,
+ NRENL3CoreServiceType,
+)
from gso.services.subscriptions import get_product_id_by_name
-@step("Create new IP trunk subscription")
-def import_geant_ip_subscription(subscription_id: UUIDstr) -> State:
- """Take an ImportedGeantIP subscription, and turn it into an GeantIP subscription."""
- old_geant_ip = ImportedNRENL3CoreService.from_subscription(subscription_id)
- new_subscription_id = get_product_id_by_name(ProductName.GEANT_IP)
- new_subscription = NRENL3CoreService.from_other_product(old_geant_ip, new_subscription_id) # type: ignore[arg-type]
+@step("Create imported subscription")
+def import_nren_l3_core_service_subscription(subscription_id: UUIDstr) -> State:
+ """Take an imported subscription, and turn it into an :term:`NREN` L3 Core Service subscription."""
+ old_nren_l3_core_service = ImportedNRENL3CoreService.from_subscription(subscription_id)
+ if old_nren_l3_core_service.nren_l3_core_service_type == NRENL3CoreServiceType.IMPORTED_GEANT_IP:
+ new_subscription_id = get_product_id_by_name(ProductName.GEANT_IP)
+ elif old_nren_l3_core_service.nren_l3_core_service_type == NRENL3CoreServiceType.IMPORTED_IAS:
+ new_subscription_id = get_product_id_by_name(ProductName.IAS)
+ else:
+ msg = f"This {old_nren_l3_core_service.nren_l3_core_service_type} is already imported, nothing to do."
+ raise ProcessFailureError(message=msg, details=old_nren_l3_core_service)
+ new_subscription = NRENL3CoreService.from_other_product(old_nren_l3_core_service, new_subscription_id) # type: ignore[arg-type]
return {"subscription": new_subscription}
-@workflow("Import GÉANT IP", target=Target.MODIFY, initial_input_form=wrap_modify_initial_input_form(None))
+@workflow("Import NREN L3 Core Service", target=Target.MODIFY, initial_input_form=wrap_modify_initial_input_form(None))
def import_nren_l3_core_service() -> StepList:
- """Modify an ImportedGeantIP subscription into an GeantIP subscription to complete the import."""
- return init >> store_process_subscription(Target.MODIFY) >> unsync >> import_geant_ip_subscription >> resync >> done
+ """Modify an imported subscription into an :term:`NREN` L3 Core Service subscription to complete the import."""
+ return (
+ init
+ >> store_process_subscription(Target.MODIFY)
+ >> unsync
+ >> import_nren_l3_core_service_subscription
+ >> resync
+ >> done
+ )
diff --git a/gso/workflows/nren_l3_core_service/modify_nren_l3_core_service.py b/gso/workflows/nren_l3_core_service/modify_nren_l3_core_service.py
index 12c809b7..c3cef39f 100644
--- a/gso/workflows/nren_l3_core_service/modify_nren_l3_core_service.py
+++ b/gso/workflows/nren_l3_core_service/modify_nren_l3_core_service.py
@@ -280,7 +280,7 @@ def create_new_sbp_blocks(subscription: NRENL3CoreService, added_service_binding
@workflow(
- "Modify :term:`NREN` L3 Core Service",
+ "Modify NREN L3 Core Service",
initial_input_form=wrap_modify_initial_input_form(initial_input_form_generator),
target=Target.MODIFY,
)
diff --git a/test/cli/test_imports.py b/test/cli/test_imports.py
index ea089f91..55f7db15 100644
--- a/test/cli/test_imports.py
+++ b/test/cli/test_imports.py
@@ -13,11 +13,12 @@ from gso.cli.imports import (
import_sites,
import_super_pop_switches,
)
-from gso.products import Router, Site
from gso.products.product_blocks.edge_port import EdgePortType, EncapsulationType
from gso.products.product_blocks.iptrunk import IptrunkType
from gso.products.product_blocks.router import RouterRole
from gso.products.product_blocks.site import SiteTier
+from gso.products.product_types.router import Router
+from gso.products.product_types.site import Site
from gso.utils.helpers import iso_from_ipv4
from gso.utils.shared_enums import Vendor
from gso.utils.types.interfaces import PhysicalPortCapacity
diff --git a/test/conftest.py b/test/conftest.py
index e8d0a283..88e43569 100644
--- a/test/conftest.py
+++ b/test/conftest.py
@@ -38,10 +38,10 @@ from gso.utils.types.interfaces import LAGMember, LAGMemberList
from test.fixtures import ( # noqa: F401
bgp_session_subscription_factory,
edge_port_subscription_factory,
- geant_ip_subscription_factory,
iptrunk_side_subscription_factory,
iptrunk_subscription_factory,
nren_access_port_factory,
+ nren_l3_core_service_subscription_factory,
office_router_subscription_factory,
opengear_subscription_factory,
router_subscription_factory,
diff --git a/test/fixtures/__init__.py b/test/fixtures/__init__.py
index fbedac94..645e6d20 100644
--- a/test/fixtures/__init__.py
+++ b/test/fixtures/__init__.py
@@ -1,11 +1,11 @@
from test.fixtures.edge_port_fixtures import edge_port_subscription_factory
-from test.fixtures.geant_ip_fixtures import (
+from test.fixtures.iptrunk_fixtures import iptrunk_side_subscription_factory, iptrunk_subscription_factory
+from test.fixtures.nren_l3_core_service_fixtures import (
bgp_session_subscription_factory,
- geant_ip_subscription_factory,
nren_access_port_factory,
+ nren_l3_core_service_subscription_factory,
service_binding_port_factory,
)
-from test.fixtures.iptrunk_fixtures import iptrunk_side_subscription_factory, iptrunk_subscription_factory
from test.fixtures.office_router_fixtures import office_router_subscription_factory
from test.fixtures.opengear_fixtures import opengear_subscription_factory
from test.fixtures.router_fixtures import juniper_router_subscription_factory, nokia_router_subscription_factory
@@ -15,12 +15,12 @@ from test.fixtures.super_pop_switch_fixtures import super_pop_switch_subscriptio
__all__ = [
"bgp_session_subscription_factory",
"edge_port_subscription_factory",
- "geant_ip_subscription_factory",
"iptrunk_side_subscription_factory",
"iptrunk_subscription_factory",
"juniper_router_subscription_factory",
"nokia_router_subscription_factory",
"nren_access_port_factory",
+ "nren_l3_core_service_subscription_factory",
"office_router_subscription_factory",
"opengear_subscription_factory",
"service_binding_port_factory",
diff --git a/test/fixtures/edge_port_fixtures.py b/test/fixtures/edge_port_fixtures.py
index c8e73f09..f689c191 100644
--- a/test/fixtures/edge_port_fixtures.py
+++ b/test/fixtures/edge_port_fixtures.py
@@ -5,13 +5,14 @@ from orchestrator.db import (
from orchestrator.domain import SubscriptionModel
from orchestrator.types import SubscriptionLifecycle, UUIDstr
-from gso.products import ProductName, Router
+from gso.products import ProductName
from gso.products.product_blocks.edge_port import (
EdgePortAEMemberBlock,
EdgePortType,
EncapsulationType,
)
from gso.products.product_types.edge_port import EdgePortInactive, ImportedEdgePortInactive
+from gso.products.product_types.router import Router
from gso.services import subscriptions
from gso.utils.types.interfaces import PhysicalPortCapacity
diff --git a/test/fixtures/geant_ip_fixtures.py b/test/fixtures/nren_l3_core_service_fixtures.py
similarity index 61%
rename from test/fixtures/geant_ip_fixtures.py
rename to test/fixtures/nren_l3_core_service_fixtures.py
index cb864966..059f2d37 100644
--- a/test/fixtures/geant_ip_fixtures.py
+++ b/test/fixtures/nren_l3_core_service_fixtures.py
@@ -6,11 +6,16 @@ from orchestrator.db import db
from orchestrator.domain import SubscriptionModel
from orchestrator.types import SubscriptionLifecycle, UUIDstr
-from gso.products import EdgePort, ProductName
+from gso.products import ProductName
from gso.products.product_blocks.bgp_session import BGPSession, IPFamily
from gso.products.product_blocks.nren_l3_core_service import NRENAccessPort
from gso.products.product_blocks.service_binding_port import ServiceBindingPort
-from gso.products.product_types.nren_l3_core_service import ImportedNRENL3CoreService, NRENL3CoreServiceInactive
+from gso.products.product_types.edge_port import EdgePort
+from gso.products.product_types.nren_l3_core_service import (
+ ImportedNRENL3CoreService,
+ NRENL3CoreServiceInactive,
+ NRENL3CoreServiceType,
+)
from gso.services import subscriptions
from gso.utils.shared_enums import APType, SBPType
from gso.utils.types.ip_address import IPAddress
@@ -55,7 +60,7 @@ def bgp_session_subscription_factory(faker):
@pytest.fixture()
def service_binding_port_factory(faker, bgp_session_subscription_factory, edge_port_subscription_factory):
def create_service_binding_port(
- sbp_bgp_session_list: list | None = None,
+ bgp_session_list: list[BGPSession] | None = None,
geant_sid: str | None = None,
sbp_type: SBPType = SBPType.L3,
ipv4_address: str | None = None,
@@ -79,7 +84,7 @@ def service_binding_port_factory(faker, bgp_session_subscription_factory, edge_p
ipv6_mask=ipv6_mask or faker.ipv6_netmask(),
custom_firewall_filters=custom_firewall_filters,
geant_sid=geant_sid or faker.geant_sid(),
- sbp_bgp_session_list=sbp_bgp_session_list
+ bgp_session_list=bgp_session_list
or [
bgp_session_subscription_factory(families=[IPFamily.V4UNICAST]),
bgp_session_subscription_factory(families=[IPFamily.V6UNICAST], peer_address=faker.ipv6()),
@@ -98,58 +103,71 @@ def nren_access_port_factory(faker, service_binding_port_factory):
):
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(),
+ ap_type=nren_ap_type or random.choice(list(APType)), # noqa: S311
+ sbp=service_binding_port or service_binding_port_factory(),
)
return create_nren_access_port
@pytest.fixture()
-def geant_ip_subscription_factory(
+def nren_l3_core_service_subscription_factory(
faker,
partner_factory,
nren_access_port_factory,
):
- def create_geant_ip_subscription(
+ def create_nren_l3_core_service_subscription(
+ nren_l3_core_service_type: NRENL3CoreServiceType,
description=None,
partner: dict | None = None,
nren_ap_list: list[NRENAccessPort] | None = None,
start_date="2023-05-24T00:00:00+00:00",
status: SubscriptionLifecycle | None = None,
- *,
- is_imported=True,
) -> UUIDstr:
partner = partner or partner_factory()
- if is_imported:
- product_id = subscriptions.get_product_id_by_name(ProductName.GEANT_IP)
- geant_ip_subscription = NRENL3CoreServiceInactive.from_product_id(
- product_id, customer_id=partner["partner_id"], insync=True
- )
- else:
- product_id = subscriptions.get_product_id_by_name(ProductName.IMPORTED_GEANT_IP)
- geant_ip_subscription = ImportedNRENL3CoreService.from_product_id(
- product_id, customer_id=partner["partner_id"], insync=True
- )
+ match nren_l3_core_service_type:
+ case NRENL3CoreServiceType.GEANT_IP:
+ product_id = subscriptions.get_product_id_by_name(ProductName.GEANT_IP)
+ nren_l3_core_service_subscription = NRENL3CoreServiceInactive.from_product_id(
+ product_id, customer_id=partner["partner_id"], insync=True
+ )
+ case NRENL3CoreServiceType.IMPORTED_GEANT_IP:
+ product_id = subscriptions.get_product_id_by_name(ProductName.IMPORTED_GEANT_IP)
+ nren_l3_core_service_subscription = ImportedNRENL3CoreService.from_product_id(
+ product_id, customer_id=partner["partner_id"], insync=True
+ )
+ case NRENL3CoreServiceType.IAS:
+ product_id = subscriptions.get_product_id_by_name(ProductName.IAS)
+ nren_l3_core_service_subscription = NRENL3CoreServiceInactive.from_product_id(
+ product_id, customer_id=partner["partner_id"], insync=True
+ )
+ case NRENL3CoreServiceType.IMPORTED_IAS:
+ product_id = subscriptions.get_product_id_by_name(ProductName.IMPORTED_IAS)
+ nren_l3_core_service_subscription = ImportedNRENL3CoreService.from_product_id(
+ product_id, customer_id=partner["partner_id"], insync=True
+ )
+ case _:
+ msg = f"NREN L3 Core Service type not found: {nren_l3_core_service_type}"
+ raise ValueError(msg)
# Default nren_ap_list creation with primary and backup access ports
- geant_ip_subscription.geant_ip.geant_ip_ap_list = nren_ap_list or [
+ nren_l3_core_service_subscription.nren_l3_core_service.nren_ap_list = nren_ap_list or [
nren_access_port_factory(nren_ap_type=APType.PRIMARY),
nren_access_port_factory(nren_ap_type=APType.BACKUP),
]
# Update subscription with description, start date, and status
- geant_ip_subscription = SubscriptionModel.from_other_lifecycle(
- geant_ip_subscription,
+ nren_l3_core_service_subscription = SubscriptionModel.from_other_lifecycle(
+ nren_l3_core_service_subscription,
SubscriptionLifecycle.ACTIVE,
)
- 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()
+ nren_l3_core_service_subscription.description = description or faker.sentence()
+ nren_l3_core_service_subscription.start_date = start_date
+ nren_l3_core_service_subscription.status = status or SubscriptionLifecycle.ACTIVE
+ nren_l3_core_service_subscription.save()
db.session.commit()
- return str(geant_ip_subscription.subscription_id)
+ return str(nren_l3_core_service_subscription.subscription_id)
- return create_geant_ip_subscription
+ return create_nren_l3_core_service_subscription
diff --git a/test/utils/test_helpers.py b/test/utils/test_helpers.py
index 0df4ced9..880893b6 100644
--- a/test/utils/test_helpers.py
+++ b/test/utils/test_helpers.py
@@ -3,9 +3,9 @@ from unittest.mock import patch
import pytest
from orchestrator.types import SubscriptionLifecycle
-from gso.products import Router
from gso.products.product_blocks.iptrunk import IptrunkInterfaceBlock
from gso.products.product_blocks.router import RouterRole
+from gso.products.product_types.router import Router
from gso.utils.helpers import (
available_interfaces_choices_including_current_members,
generate_inventory_for_active_routers,
diff --git a/test/workflows/edge_port/test_create_edge_port.py b/test/workflows/edge_port/test_create_edge_port.py
index f6a13146..d087134f 100644
--- a/test/workflows/edge_port/test_create_edge_port.py
+++ b/test/workflows/edge_port/test_create_edge_port.py
@@ -4,8 +4,10 @@ from unittest.mock import patch
import pytest
from pydantic_forms.exceptions import FormValidationError
-from gso.products import EdgePort, ProductName, Router
+from gso.products import ProductName
from gso.products.product_blocks.edge_port import EdgePortType, EncapsulationType
+from gso.products.product_types.edge_port import EdgePort
+from gso.products.product_types.router import Router
from gso.services.subscriptions import get_product_id_by_name
from gso.utils.types.interfaces import PhysicalPortCapacity
from test.services.conftest import MockedNetboxClient
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 6f494366..8ebd12a1 100644
--- a/test/workflows/edge_port/test_create_imported_edge_port.py
+++ b/test/workflows/edge_port/test_create_imported_edge_port.py
@@ -1,8 +1,8 @@
import pytest
from orchestrator.types import SubscriptionLifecycle
-from gso.products import ImportedEdgePort
from gso.products.product_blocks.edge_port import EdgePortType, EncapsulationType
+from gso.products.product_types.edge_port import ImportedEdgePort
from gso.utils.types.interfaces import PhysicalPortCapacity
from test.workflows import assert_complete, extract_state, run_workflow
diff --git a/test/workflows/edge_port/test_import_edge_port.py b/test/workflows/edge_port/test_import_edge_port.py
index 46405d50..6d79e89d 100644
--- a/test/workflows/edge_port/test_import_edge_port.py
+++ b/test/workflows/edge_port/test_import_edge_port.py
@@ -1,7 +1,8 @@
import pytest
from orchestrator.types import SubscriptionLifecycle
-from gso.products import EdgePort, ProductName
+from gso.products import ProductName
+from gso.products.product_types.edge_port import EdgePort
from test.workflows import assert_complete, run_workflow
diff --git a/test/workflows/edge_port/test_modify_edge_port.py b/test/workflows/edge_port/test_modify_edge_port.py
index 68635d14..a7c82428 100644
--- a/test/workflows/edge_port/test_modify_edge_port.py
+++ b/test/workflows/edge_port/test_modify_edge_port.py
@@ -2,7 +2,7 @@ from unittest.mock import patch
import pytest
-from gso.products import EdgePort
+from gso.products.product_types.edge_port import EdgePort
from gso.utils.types.interfaces import PhysicalPortCapacity
from test.workflows import (
assert_complete,
diff --git a/test/workflows/edge_port/test_terminate_edge_port.py b/test/workflows/edge_port/test_terminate_edge_port.py
index ac26fa3e..b5a16866 100644
--- a/test/workflows/edge_port/test_terminate_edge_port.py
+++ b/test/workflows/edge_port/test_terminate_edge_port.py
@@ -2,7 +2,7 @@ from unittest.mock import patch
import pytest
-from gso.products import EdgePort
+from gso.products.product_types.edge_port import EdgePort
from test.services.conftest import MockedNetboxClient
from test.workflows import (
assert_complete,
diff --git a/test/workflows/edge_port/test_validate_edge_port.py b/test/workflows/edge_port/test_validate_edge_port.py
index 9559f7f4..3aff8081 100644
--- a/test/workflows/edge_port/test_validate_edge_port.py
+++ b/test/workflows/edge_port/test_validate_edge_port.py
@@ -2,7 +2,7 @@ from unittest.mock import patch
import pytest
-from gso.products import EdgePort
+from gso.products.product_types.edge_port import EdgePort
from test.services.conftest import MockedNetboxClient
from test.workflows import (
assert_complete,
diff --git a/test/workflows/geant_ip/test_import_geant_ip.py b/test/workflows/geant_ip/test_import_geant_ip.py
deleted file mode 100644
index 55ce0cce..00000000
--- a/test/workflows/geant_ip/test_import_geant_ip.py
+++ /dev/null
@@ -1,17 +0,0 @@
-import pytest
-from orchestrator.types import SubscriptionLifecycle
-
-from gso.products import NRENL3CoreService, ProductName
-from test.workflows import assert_complete, run_workflow
-
-
-@pytest.mark.workflow()
-def test_import_edge_port_success(geant_ip_subscription_factory):
- imported_geant_ip = geant_ip_subscription_factory(is_imported=False)
- result, _, _ = run_workflow("import_geant_ip", [{"subscription_id": imported_geant_ip}])
- subscription = NRENL3CoreService.from_subscription(imported_geant_ip)
-
- assert_complete(result)
- assert subscription.product.name == ProductName.GEANT_IP
- assert subscription.status == SubscriptionLifecycle.ACTIVE
- assert subscription.insync
diff --git a/test/workflows/geant_ip/test_migrate_geant_ip.py b/test/workflows/geant_ip/test_migrate_geant_ip.py
deleted file mode 100644
index 949082a2..00000000
--- a/test/workflows/geant_ip/test_migrate_geant_ip.py
+++ /dev/null
@@ -1,46 +0,0 @@
-import pytest
-
-from gso.products.product_types.nren_l3_core_service import NRENL3CoreService
-from test.workflows import assert_complete, extract_state, run_workflow
-
-
-@pytest.mark.workflow()
-def test_migrate_geant_ip_success(
- faker, edge_port_subscription_factory, partner_factory, geant_ip_subscription_factory
-):
- partner = partner_factory()
- subscription_id = geant_ip_subscription_factory(partner=partner)
- new_edge_port_1 = edge_port_subscription_factory(partner=partner)
- new_edge_port_2 = edge_port_subscription_factory(partner=partner)
- subscription = NRENL3CoreService.from_subscription(subscription_id)
-
- form_input_data = [
- {"subscription_id": subscription_id},
- {
- "tt_number": faker.tt_number(),
- "edge_port_selection": [
- {
- "old_edge_port": subscription.geant_ip.geant_ip_ap_list[0].geant_ip_sbp.edge_port.description,
- "new_edge_port": new_edge_port_1,
- },
- {
- "old_edge_port": subscription.geant_ip.geant_ip_ap_list[1].geant_ip_sbp.edge_port.description,
- "new_edge_port": new_edge_port_2,
- },
- ],
- },
- ]
-
- result, _, _ = run_workflow("migrate_geant_ip", form_input_data)
-
- assert_complete(result)
- state = extract_state(result)
- subscription = NRENL3CoreService.from_subscription(state["subscription_id"])
- assert subscription.insync
- assert len(subscription.geant_ip.geant_ip_ap_list) == 2
- assert (
- str(subscription.geant_ip.geant_ip_ap_list[0].geant_ip_sbp.edge_port.owner_subscription_id) == new_edge_port_1
- )
- assert (
- str(subscription.geant_ip.geant_ip_ap_list[1].geant_ip_sbp.edge_port.owner_subscription_id) == new_edge_port_2
- )
diff --git a/test/workflows/iptrunk/test_activate_iptrunk.py b/test/workflows/iptrunk/test_activate_iptrunk.py
index 837f340c..d1ed9b33 100644
--- a/test/workflows/iptrunk/test_activate_iptrunk.py
+++ b/test/workflows/iptrunk/test_activate_iptrunk.py
@@ -1,6 +1,6 @@
import pytest
-from gso.products import Iptrunk
+from gso.products.product_types.iptrunk import Iptrunk
from test.workflows import (
assert_complete,
assert_suspended,
diff --git a/test/workflows/iptrunk/test_create_iptrunk.py b/test/workflows/iptrunk/test_create_iptrunk.py
index 6be42fae..c21f26cc 100644
--- a/test/workflows/iptrunk/test_create_iptrunk.py
+++ b/test/workflows/iptrunk/test_create_iptrunk.py
@@ -4,8 +4,9 @@ from unittest.mock import patch
import pytest
from infoblox_client.objects import HostRecord
-from gso.products import Iptrunk, ProductName
+from gso.products import ProductName
from gso.products.product_blocks.iptrunk import IptrunkType
+from gso.products.product_types.iptrunk import Iptrunk
from gso.services.subscriptions import get_product_id_by_name
from gso.utils.shared_enums import Vendor
from gso.utils.types.interfaces import PhysicalPortCapacity
diff --git a/test/workflows/iptrunk/test_deploy_twamp.py b/test/workflows/iptrunk/test_deploy_twamp.py
index e65feac7..1b475140 100644
--- a/test/workflows/iptrunk/test_deploy_twamp.py
+++ b/test/workflows/iptrunk/test_deploy_twamp.py
@@ -2,7 +2,7 @@ from unittest.mock import patch
import pytest
-from gso.products import Iptrunk
+from gso.products.product_types.iptrunk import Iptrunk
from test.workflows import (
assert_complete,
assert_lso_interaction_success,
diff --git a/test/workflows/iptrunk/test_modify_isis_metric.py b/test/workflows/iptrunk/test_modify_isis_metric.py
index 26a9bbd4..1c739b74 100644
--- a/test/workflows/iptrunk/test_modify_isis_metric.py
+++ b/test/workflows/iptrunk/test_modify_isis_metric.py
@@ -2,7 +2,7 @@ from unittest.mock import patch
import pytest
-from gso.products import Iptrunk
+from gso.products.product_types.iptrunk import Iptrunk
from test.workflows import (
assert_complete,
assert_lso_interaction_success,
diff --git a/test/workflows/iptrunk/test_modify_trunk_interface.py b/test/workflows/iptrunk/test_modify_trunk_interface.py
index 359308cd..77470e79 100644
--- a/test/workflows/iptrunk/test_modify_trunk_interface.py
+++ b/test/workflows/iptrunk/test_modify_trunk_interface.py
@@ -2,8 +2,8 @@ from unittest.mock import patch
import pytest
-from gso.products import Iptrunk
from gso.products.product_blocks.iptrunk import IptrunkType
+from gso.products.product_types.iptrunk import Iptrunk
from gso.utils.shared_enums import Vendor
from gso.utils.types.interfaces import LAGMemberList, PhysicalPortCapacity
from test.conftest import UseJuniperSide
diff --git a/test/workflows/iptrunk/test_terminate_iptrunk.py b/test/workflows/iptrunk/test_terminate_iptrunk.py
index c745a615..83e0324a 100644
--- a/test/workflows/iptrunk/test_terminate_iptrunk.py
+++ b/test/workflows/iptrunk/test_terminate_iptrunk.py
@@ -2,8 +2,8 @@ from unittest.mock import patch
import pytest
-from gso.products import Iptrunk
from gso.products.product_blocks.router import RouterRole
+from gso.products.product_types.iptrunk import Iptrunk
from gso.settings import load_oss_params
from test.services.conftest import MockedNetboxClient
from test.workflows import (
diff --git a/test/workflows/geant_ip/__init__.py b/test/workflows/nren_l3_core_service/__init__.py
similarity index 100%
rename from test/workflows/geant_ip/__init__.py
rename to test/workflows/nren_l3_core_service/__init__.py
diff --git a/test/workflows/geant_ip/test_create_imported_geant_ip.py b/test/workflows/nren_l3_core_service/test_create_imported_nren_l3_core_service.py
similarity index 84%
rename from test/workflows/geant_ip/test_create_imported_geant_ip.py
rename to test/workflows/nren_l3_core_service/test_create_imported_nren_l3_core_service.py
index 4ce636ea..05982ef1 100644
--- a/test/workflows/geant_ip/test_create_imported_geant_ip.py
+++ b/test/workflows/nren_l3_core_service/test_create_imported_nren_l3_core_service.py
@@ -1,16 +1,19 @@
import pytest
from orchestrator.types import SubscriptionLifecycle
-from gso.products import ImportedNRENL3CoreService
from gso.products.product_blocks.bgp_session import IPFamily
+from gso.products.product_types.nren_l3_core_service import ImportedNRENL3CoreService, NRENL3CoreServiceType
from gso.utils.shared_enums import SBPType
from test.workflows import assert_complete, extract_state, run_workflow
-@pytest.fixture()
-def imported_geant_ip_creation_input_form_data(edge_port_subscription_factory, partner_factory, faker):
- return {
+@pytest.mark.parametrize("l3_core_service_type", [NRENL3CoreServiceType.GEANT_IP, NRENL3CoreServiceType.IAS])
+def test_create_imported_nren_l3_core_service_success(
+ faker, partner_factory, edge_port_subscription_factory, l3_core_service_type
+):
+ creation_form_input_data = {
"partner": partner_factory()["name"],
+ "nren_l3_core_service_type": l3_core_service_type,
"service_binding_ports": [
{
"edge_port": edge_port_subscription_factory(),
@@ -58,9 +61,7 @@ def imported_geant_ip_creation_input_form_data(edge_port_subscription_factory, p
],
}
-
-def test_create_imported_geant_ip_success(faker, imported_geant_ip_creation_input_form_data):
- result, _, _ = run_workflow("create_imported_nren_l3_core_service", [imported_geant_ip_creation_input_form_data])
+ result, _, _ = run_workflow("create_imported_nren_l3_core_service", [creation_form_input_data])
state = extract_state(result)
subscription = ImportedNRENL3CoreService.from_subscription(state["subscription_id"])
assert_complete(result)
diff --git a/test/workflows/geant_ip/test_create_geant_ip.py b/test/workflows/nren_l3_core_service/test_create_nren_l3_core_service.py
similarity index 85%
rename from test/workflows/geant_ip/test_create_geant_ip.py
rename to test/workflows/nren_l3_core_service/test_create_nren_l3_core_service.py
index e158b46e..1287bd01 100644
--- a/test/workflows/geant_ip/test_create_geant_ip.py
+++ b/test/workflows/nren_l3_core_service/test_create_nren_l3_core_service.py
@@ -28,10 +28,12 @@ def base_bgp_peer_input(faker):
return _base_bgp_peer_input
+@pytest.mark.parametrize("l3_core_type", [ProductName.GEANT_IP, ProductName.IAS])
@pytest.mark.workflow()
@patch("gso.services.lso_client._send_request")
-def test_create_geant_ip_success(
+def test_create_nren_l3_core_service_success(
mock_lso_client,
+ l3_core_type,
responses,
faker,
partner_factory,
@@ -40,7 +42,7 @@ def test_create_geant_ip_success(
data_config_filename,
):
partner = partner_factory()
- product_id = get_product_id_by_name(ProductName.GEANT_IP)
+ product_id = get_product_id_by_name(l3_core_type)
edge_port_a = edge_port_subscription_factory(partner=partner)
form_input_data = [
@@ -62,7 +64,7 @@ def test_create_geant_ip_success(
]
lso_interaction_count = 6
- result, process_stat, step_log = run_workflow("create_geant_ip", form_input_data)
+ result, process_stat, step_log = run_workflow("create_nren_l3_core_service", form_input_data)
for _ in range(lso_interaction_count):
result, step_log = assert_lso_interaction_success(result, process_stat, step_log)
@@ -72,8 +74,8 @@ def test_create_geant_ip_success(
subscription = NRENL3CoreService.from_subscription(state["subscription_id"])
assert mock_lso_client.call_count == lso_interaction_count
assert subscription.status == SubscriptionLifecycle.ACTIVE
- assert len(subscription.geant_ip.geant_ip_ap_list) == 1
+ assert len(subscription.nren_l3_core_service.nren_ap_list) == 1
assert (
- str(subscription.geant_ip.geant_ip_ap_list[0].geant_ip_sbp.edge_port.owner_subscription_id)
+ str(subscription.nren_l3_core_service.nren_ap_list[0].sbp.edge_port.owner_subscription_id)
== form_input_data[2]["edge_ports"][0]["edge_port"]
)
diff --git a/test/workflows/nren_l3_core_service/test_import_nren_l3_core_service.py b/test/workflows/nren_l3_core_service/test_import_nren_l3_core_service.py
new file mode 100644
index 00000000..b155bb72
--- /dev/null
+++ b/test/workflows/nren_l3_core_service/test_import_nren_l3_core_service.py
@@ -0,0 +1,23 @@
+import pytest
+from orchestrator.types import SubscriptionLifecycle
+
+from gso.products.product_types.nren_l3_core_service import NRENL3CoreService, NRENL3CoreServiceType
+from test.workflows import assert_complete, run_workflow
+
+
+@pytest.mark.parametrize(
+ "l3_core_service_type", [NRENL3CoreServiceType.IMPORTED_GEANT_IP, NRENL3CoreServiceType.IMPORTED_IAS]
+)
+@pytest.mark.workflow()
+def test_import_nren_l3_core_service_success(nren_l3_core_service_subscription_factory, l3_core_service_type):
+ imported_nren_l3_core_service = nren_l3_core_service_subscription_factory(
+ nren_l3_core_service_type=l3_core_service_type
+ )
+ result, _, _ = run_workflow("import_nren_l3_core_service", [{"subscription_id": imported_nren_l3_core_service}])
+ subscription = NRENL3CoreService.from_subscription(imported_nren_l3_core_service)
+
+ assert_complete(result)
+ # Remove the "IMPORTED_" prefix with ``[9:]``
+ assert subscription.nren_l3_core_service_type == NRENL3CoreServiceType(l3_core_service_type.value[9:])
+ assert subscription.status == SubscriptionLifecycle.ACTIVE
+ assert subscription.insync
diff --git a/test/workflows/nren_l3_core_service/test_migrate_nren_l3_core_service.py b/test/workflows/nren_l3_core_service/test_migrate_nren_l3_core_service.py
new file mode 100644
index 00000000..95d6d68b
--- /dev/null
+++ b/test/workflows/nren_l3_core_service/test_migrate_nren_l3_core_service.py
@@ -0,0 +1,49 @@
+import pytest
+
+from gso.products.product_types.nren_l3_core_service import NRENL3CoreService, NRENL3CoreServiceType
+from test.workflows import assert_complete, extract_state, run_workflow
+
+
+@pytest.mark.parametrize("l3_core_service_type", [NRENL3CoreServiceType.GEANT_IP, NRENL3CoreServiceType.IAS])
+@pytest.mark.workflow()
+def test_migrate_nren_l3_core_service_success(
+ faker,
+ edge_port_subscription_factory,
+ partner_factory,
+ nren_l3_core_service_subscription_factory,
+ l3_core_service_type,
+):
+ partner = partner_factory()
+ subscription_id = nren_l3_core_service_subscription_factory(
+ partner=partner, nren_l3_core_service_type=l3_core_service_type
+ )
+ new_edge_port_1 = edge_port_subscription_factory(partner=partner)
+ new_edge_port_2 = edge_port_subscription_factory(partner=partner)
+ subscription = NRENL3CoreService.from_subscription(subscription_id)
+
+ form_input_data = [
+ {"subscription_id": subscription_id},
+ {
+ "tt_number": faker.tt_number(),
+ "edge_port_selection": [
+ {
+ "old_edge_port": subscription.nren_l3_core_service.nren_ap_list[0].sbp.edge_port.description,
+ "new_edge_port": new_edge_port_1,
+ },
+ {
+ "old_edge_port": subscription.nren_l3_core_service.nren_ap_list[1].sbp.edge_port.description,
+ "new_edge_port": new_edge_port_2,
+ },
+ ],
+ },
+ ]
+
+ result, _, _ = run_workflow("migrate_nren_l3_core_service", form_input_data)
+
+ assert_complete(result)
+ state = extract_state(result)
+ subscription = NRENL3CoreService.from_subscription(state["subscription_id"])
+ assert subscription.insync
+ assert len(subscription.nren_l3_core_service.nren_ap_list) == 2
+ assert str(subscription.nren_l3_core_service.nren_ap_list[0].sbp.edge_port.owner_subscription_id) == new_edge_port_1
+ assert str(subscription.nren_l3_core_service.nren_ap_list[1].sbp.edge_port.owner_subscription_id) == new_edge_port_2
diff --git a/test/workflows/geant_ip/test_modify_geant_ip.py b/test/workflows/nren_l3_core_service/test_modify_nren_l3_core_service.py
similarity index 57%
rename from test/workflows/geant_ip/test_modify_geant_ip.py
rename to test/workflows/nren_l3_core_service/test_modify_nren_l3_core_service.py
index 0cd7f24a..86831534 100644
--- a/test/workflows/geant_ip/test_modify_geant_ip.py
+++ b/test/workflows/nren_l3_core_service/test_modify_nren_l3_core_service.py
@@ -1,59 +1,69 @@
import pytest
from gso.products.product_blocks.bgp_session import IPFamily
-from gso.products.product_types.nren_l3_core_service import NRENL3CoreService
+from gso.products.product_types.nren_l3_core_service import NRENL3CoreService, NRENL3CoreServiceType
from gso.utils.shared_enums import APType
from test.workflows import extract_state, run_workflow
+@pytest.mark.parametrize("l3_core_service_type", [NRENL3CoreServiceType.GEANT_IP, NRENL3CoreServiceType.IAS])
@pytest.mark.workflow()
-def test_modify_geant_ip_remove_edge_port_success(geant_ip_subscription_factory):
- subscription_id = geant_ip_subscription_factory()
+def test_modify_nren_l3_core_service_remove_edge_port_success(
+ nren_l3_core_service_subscription_factory, l3_core_service_type
+):
+ subscription_id = nren_l3_core_service_subscription_factory(nren_l3_core_service_type=l3_core_service_type)
subscription = NRENL3CoreService.from_subscription(subscription_id)
- access_port = subscription.geant_ip.geant_ip_ap_list[0]
+ access_port = subscription.nren_l3_core_service.nren_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,
+ "edge_port": str(access_port.sbp.edge_port.owner_subscription_id),
+ "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)
+ result, _, _ = run_workflow("modify_nren_l3_core_service", input_form_data)
state = extract_state(result)
subscription = NRENL3CoreService.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
+ assert len(subscription.nren_l3_core_service.nren_ap_list) == 1
+ assert subscription.nren_l3_core_service.nren_ap_list[0].ap_type == APType.LOAD_BALANCED
+@pytest.mark.parametrize("l3_core_service_type", [NRENL3CoreServiceType.GEANT_IP, NRENL3CoreServiceType.IAS])
@pytest.mark.workflow()
-def test_modify_geant_ip_add_new_edge_port_success(
- geant_ip_subscription_factory, edge_port_subscription_factory, partner_factory, faker
+def test_modify_nren_l3_core_service_add_new_edge_port_success(
+ nren_l3_core_service_subscription_factory,
+ edge_port_subscription_factory,
+ partner_factory,
+ faker,
+ l3_core_service_type,
):
partner = partner_factory()
new_edge_port = edge_port_subscription_factory(partner=partner)
- subscription_id = geant_ip_subscription_factory(partner=partner)
+ subscription_id = nren_l3_core_service_subscription_factory(
+ partner=partner, nren_l3_core_service_type=l3_core_service_type
+ )
subscription = NRENL3CoreService.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,
+ "edge_port": str(port.sbp.edge_port.owner_subscription_id),
+ "ap_type": port.ap_type,
}
- for port in subscription.geant_ip.geant_ip_ap_list
+ for port in subscription.nren_l3_core_service.nren_ap_list
]
+ [
{
- "geant_ip_ep": str(new_edge_port),
- "nren_ap_type": APType.BACKUP,
+ "edge_port": str(new_edge_port),
+ "ap_type": APType.BACKUP,
}
]
},
@@ -75,16 +85,19 @@ def test_modify_geant_ip_add_new_edge_port_success(
},
]
- result, _, _ = run_workflow("modify_geant_ip", input_form_data)
+ result, _, _ = run_workflow("modify_nren_l3_core_service", input_form_data)
state = extract_state(result)
subscription = NRENL3CoreService.from_subscription(state["subscription_id"])
- assert len(subscription.geant_ip.geant_ip_ap_list) == 3
+ assert len(subscription.nren_l3_core_service.nren_ap_list) == 3
+@pytest.mark.parametrize("l3_core_service_type", [NRENL3CoreServiceType.GEANT_IP, NRENL3CoreServiceType.IAS])
@pytest.mark.workflow()
-def test_modify_geant_ip_modify_edge_port_success(faker, geant_ip_subscription_factory):
- subscription_id = geant_ip_subscription_factory()
+def test_modify_nren_l3_core_service_modify_edge_port_success(
+ faker, nren_l3_core_service_subscription_factory, l3_core_service_type
+):
+ subscription_id = nren_l3_core_service_subscription_factory(nren_l3_core_service_type=l3_core_service_type)
subscription = NRENL3CoreService.from_subscription(subscription_id)
new_sbp_data = [
{
@@ -158,121 +171,121 @@ def test_modify_geant_ip_modify_edge_port_success(faker, geant_ip_subscription_f
{
"access_ports": [
{
- "geant_ip_ep": str(port.geant_ip_sbp.edge_port.owner_subscription_id),
- "nren_ap_type": port.nren_ap_type,
+ "edge_port": str(port.sbp.edge_port.owner_subscription_id),
+ "ap_type": port.ap_type,
}
- for port in subscription.geant_ip.geant_ip_ap_list
+ for port in subscription.nren_l3_core_service.nren_ap_list
]
},
{**new_sbp_data[0]},
{**new_sbp_data[1]},
]
- result, _, _ = run_workflow("modify_geant_ip", input_form_data)
+ result, _, _ = run_workflow("modify_nren_l3_core_service", input_form_data)
state = extract_state(result)
subscription = NRENL3CoreService.from_subscription(state["subscription_id"])
- assert len(subscription.geant_ip.geant_ip_ap_list) == 2
+ assert len(subscription.nren_l3_core_service.nren_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 subscription.nren_l3_core_service.nren_ap_list[i].sbp.geant_sid == new_sbp_data[i]["geant_sid"]
+ assert subscription.nren_l3_core_service.nren_ap_list[i].sbp.is_tagged == new_sbp_data[i]["is_tagged"]
+ assert subscription.nren_l3_core_service.nren_ap_list[i].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"]
+ str(subscription.nren_l3_core_service.nren_ap_list[i].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"]
+ str(subscription.nren_l3_core_service.nren_ap_list[i].sbp.ipv6_address) == new_sbp_data[i]["ipv6_address"]
)
assert (
- subscription.geant_ip.geant_ip_ap_list[i].geant_ip_sbp.custom_firewall_filters
+ subscription.nren_l3_core_service.nren_ap_list[i].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
+ subscription.nren_l3_core_service.nren_ap_list[i].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
+ subscription.nren_l3_core_service.nren_ap_list[i].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
+ subscription.nren_l3_core_service.nren_ap_list[i].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
+ subscription.nren_l3_core_service.nren_ap_list[i].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
+ subscription.nren_l3_core_service.nren_ap_list[i].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
+ subscription.nren_l3_core_service.nren_ap_list[i].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
+ subscription.nren_l3_core_service.nren_ap_list[i].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
+ subscription.nren_l3_core_service.nren_ap_list[i].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)
+ str(subscription.nren_l3_core_service.nren_ap_list[i].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
+ in subscription.nren_l3_core_service.nren_ap_list[i].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
+ subscription.nren_l3_core_service.nren_ap_list[i].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
+ subscription.nren_l3_core_service.nren_ap_list[i].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
+ subscription.nren_l3_core_service.nren_ap_list[i].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
+ subscription.nren_l3_core_service.nren_ap_list[i].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
+ subscription.nren_l3_core_service.nren_ap_list[i].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
+ subscription.nren_l3_core_service.nren_ap_list[i].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
+ subscription.nren_l3_core_service.nren_ap_list[i].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
+ subscription.nren_l3_core_service.nren_ap_list[i].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)
+ str(subscription.nren_l3_core_service.nren_ap_list[i].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
+ in subscription.nren_l3_core_service.nren_ap_list[i].sbp.bgp_session_list[1].families
)
== new_sbp_data[i]["v6_bgp_peer"]["add_v6_multicast"]
)
diff --git a/test/workflows/opengear/test_create_imported_opengear.py b/test/workflows/opengear/test_create_imported_opengear.py
index 0f6d9bf1..9058c602 100644
--- a/test/workflows/opengear/test_create_imported_opengear.py
+++ b/test/workflows/opengear/test_create_imported_opengear.py
@@ -1,7 +1,8 @@
import pytest
from orchestrator.types import SubscriptionLifecycle
-from gso.products import ImportedOpengear, ProductName
+from gso.products import ProductName
+from gso.products.product_types.opengear import ImportedOpengear
from gso.products.product_types.site import Site
from test.workflows import (
assert_complete,
diff --git a/test/workflows/router/test_activate_router.py b/test/workflows/router/test_activate_router.py
index da0c24a6..c118dcaf 100644
--- a/test/workflows/router/test_activate_router.py
+++ b/test/workflows/router/test_activate_router.py
@@ -1,6 +1,6 @@
import pytest
-from gso.products import Router
+from gso.products.product_types.router import Router
from test.workflows import (
assert_complete,
assert_suspended,
diff --git a/test/workflows/router/test_create_router.py b/test/workflows/router/test_create_router.py
index a6bf456f..104bea75 100644
--- a/test/workflows/router/test_create_router.py
+++ b/test/workflows/router/test_create_router.py
@@ -3,9 +3,10 @@ from unittest.mock import patch
import pytest
from infoblox_client import objects
-from gso.products import ProductName, Site
+from gso.products import ProductName
from gso.products.product_blocks.router import RouterRole
from gso.products.product_types.router import Router
+from gso.products.product_types.site import Site
from gso.services.subscriptions import get_product_id_by_name
from gso.utils.shared_enums import Vendor
from test import USER_CONFIRM_EMPTY_FORM
diff --git a/test/workflows/router/test_modify_connection_strategy.py b/test/workflows/router/test_modify_connection_strategy.py
index e460f2e9..fd3a0135 100644
--- a/test/workflows/router/test_modify_connection_strategy.py
+++ b/test/workflows/router/test_modify_connection_strategy.py
@@ -1,6 +1,6 @@
import pytest
-from gso.products import Router
+from gso.products.product_types.router import Router
from gso.utils.shared_enums import ConnectionStrategy
from test.workflows import assert_complete, run_workflow
diff --git a/test/workflows/router/test_redeploy_base_config.py b/test/workflows/router/test_redeploy_base_config.py
index c9697cb7..2023ec04 100644
--- a/test/workflows/router/test_redeploy_base_config.py
+++ b/test/workflows/router/test_redeploy_base_config.py
@@ -2,7 +2,7 @@ from unittest.mock import patch
import pytest
-from gso.products import Router
+from gso.products.product_types.router import Router
from test.workflows import (
assert_complete,
assert_lso_interaction_success,
diff --git a/test/workflows/router/test_terminate_router.py b/test/workflows/router/test_terminate_router.py
index a182cc74..d2dca96f 100644
--- a/test/workflows/router/test_terminate_router.py
+++ b/test/workflows/router/test_terminate_router.py
@@ -2,8 +2,8 @@ from unittest.mock import patch
import pytest
-from gso.products import Router
from gso.products.product_blocks.router import RouterRole
+from gso.products.product_types.router import Router
from test.services.conftest import MockedKentikClient
from test.workflows import assert_complete, assert_lso_interaction_success, extract_state, run_workflow
diff --git a/test/workflows/router/test_update_ibgp_mesh.py b/test/workflows/router/test_update_ibgp_mesh.py
index 023cd2cb..693bfd67 100644
--- a/test/workflows/router/test_update_ibgp_mesh.py
+++ b/test/workflows/router/test_update_ibgp_mesh.py
@@ -5,8 +5,8 @@ from orchestrator.types import SubscriptionLifecycle
from orchestrator.workflow import StepStatus
from pydantic_forms.exceptions import FormValidationError
-from gso.products import Iptrunk
from gso.products.product_blocks.router import RouterRole
+from gso.products.product_types.iptrunk import Iptrunk
from test import USER_CONFIRM_EMPTY_FORM
from test.workflows import (
assert_lso_interaction_success,
--
GitLab