Rename NRENL3CoreService and update BGP peer

• Rename to L3CoreService
• Add Copernicus, LHCOne, and GWS as new L3 core services
• Add prefix limit to BGP peer

gso/cli/imports.py

@@ -13,7 +13,7 @@ import yaml
 from orchestrator.db import db
 from orchestrator.services.processes import start_process
 from orchestrator.types import SubscriptionLifecycle, UUIDstr
-from pydantic import BaseModel, ValidationError, field_validator, model_validator
+from pydantic import BaseModel, NonNegativeInt, ValidationError, field_validator, model_validator
 from sqlalchemy.exc import SQLAlchemyError
 
 from gso.db.models import PartnerTable
@@ -238,8 +238,8 @@ class EdgePortImportModel(BaseModel):
         return self
 
 
-class NRENL3CoreServiceImportModel(BaseModel):
-    """Import :term:`NREN` L3 Core Service model."""
+class L3CoreServiceImportModel(BaseModel):
+    """Import L3 Core Service model."""
 
     class BaseBGPPeer(BaseModel):
         """Base BGP Peer model."""
@@ -254,6 +254,7 @@ class NRENL3CoreServiceImportModel(BaseModel):
         families: list[IPFamily]
         is_multi_hop: bool
         rtbh_enabled: bool  # whether Remote Triggered Blackhole is enabled
+        prefix_limit: NonNegativeInt | None = None
 
     class BFDSettingsModel(BaseModel):
         """BFD Settings model."""
@@ -278,9 +279,9 @@ class NRENL3CoreServiceImportModel(BaseModel):
         ipv6_address: IPv6AddressType
         ipv6_mask: IPV6Netmask
         is_multi_hop: bool = True
-        bgp_peers: list["NRENL3CoreServiceImportModel.BaseBGPPeer"]
-        v4_bfd_settings: "NRENL3CoreServiceImportModel.BFDSettingsModel"
-        v6_bfd_settings: "NRENL3CoreServiceImportModel.BFDSettingsModel"
+        bgp_peers: list["L3CoreServiceImportModel.BaseBGPPeer"]
+        v4_bfd_settings: "L3CoreServiceImportModel.BFDSettingsModel"
+        v6_bfd_settings: "L3CoreServiceImportModel.BFDSettingsModel"
 
     partner: str
     service_binding_ports: list[ServiceBindingPort]
@@ -390,7 +391,7 @@ T = TypeVar(
     OfficeRouterImportModel,
     OpenGearImportModel,
     EdgePortImportModel,
-    NRENL3CoreServiceImportModel,
+    L3CoreServiceImportModel,
     LanSwitchInterconnectImportModel,
     Layer2CircuitServiceImportModel,
 )
@@ -655,20 +656,20 @@ def import_partners(file_path: str = typer.Argument(..., help="Path to the CSV f
 
 
 @app.command()
-def import_nren_l3_core_service(filepath: str = common_filepath_option) -> None:
-    """Import :term:`NREN` L3 Core Services into :term:`GSO`."""
+def import_l3_core_service(filepath: str = common_filepath_option) -> None:
+    """Import L3 Core Services into :term:`GSO`."""
     successfully_imported_data = []
-    nren_l3_core_service_list = _read_data(Path(filepath))
+    l3_core_service_list = _read_data(Path(filepath))
 
-    for nren_l3_core_service in nren_l3_core_service_list:
-        partner = nren_l3_core_service["partner"]
-        service_type = nren_l3_core_service["service_type"]
+    for l3_core_service in l3_core_service_list:
+        partner = l3_core_service["partner"]
+        service_type = l3_core_service["service_type"]
         typer.echo(f"Creating imported {service_type} for {partner}")
 
         try:
-            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"]]
+            initial_data = L3CoreServiceImportModel(**l3_core_service)
+            start_process("create_imported_l3_core_service", [initial_data.model_dump()])
+            edge_ports = [sbp["edge_port"] for sbp in l3_core_service["service_binding_ports"]]
             successfully_imported_data.append(edge_ports)
             typer.echo(f"Successfully created imported {service_type} for {partner}")
         except ValidationError as e:
@@ -679,17 +680,22 @@ def import_nren_l3_core_service(filepath: str = common_filepath_option) -> None:
 
     # Migrate new products from imported to "full" counterpart.
     imported_products = get_subscriptions(
-        product_types=[ProductType.IMPORTED_GEANT_IP, ProductType.IMPORTED_IAS],
+        product_types=[
+            ProductType.IMPORTED_GEANT_IP,
+            ProductType.IMPORTED_IAS,
+            ProductType.IMPORTED_LHCONE,
+            ProductType.IMPORTED_COPERNICUS,
+        ],
         lifecycles=[SubscriptionLifecycle.ACTIVE],
         includes=["subscription_id"],
     )
 
     for subscription_id in imported_products:
         typer.echo(f"Importing {subscription_id}")
-        start_process("import_nren_l3_core_service", [subscription_id])
+        start_process("import_l3_core_service", [subscription_id])
 
     if successfully_imported_data:
-        typer.echo("Successfully created imported NREN L3 Core Services:")
+        typer.echo("Successfully created imported L3 Core Services:")
         for item in successfully_imported_data:
             typer.echo(f"- {item}")