Feature/l2circuits

gso/cli/imports.py

@@ -17,13 +17,14 @@ from pydantic import BaseModel, ValidationError, field_validator, model_validato
 from sqlalchemy.exc import SQLAlchemyError
 
 from gso.db.models import PartnerTable
-from gso.products import ProductType
+from gso.products import Layer2CircuitServiceType, ProductType
 from gso.products.product_blocks.bgp_session import IPFamily
 from gso.products.product_blocks.edge_port import EdgePortType, EncapsulationType
 from gso.products.product_blocks.iptrunk import IptrunkType
+from gso.products.product_blocks.layer_2_circuit import Layer2CircuitType
 from gso.products.product_blocks.router import RouterRole
 from gso.products.product_blocks.switch import SwitchModel
-from gso.products.product_types.nren_l3_core_service import NRENL3CoreServiceType
+from gso.products.product_types.edge_port import EdgePort
 from gso.services.partners import (
     PartnerEmail,
     PartnerName,
@@ -38,7 +39,7 @@ from gso.services.subscriptions import (
 )
 from gso.utils.shared_enums import SBPType, Vendor
 from gso.utils.types.base_site import BaseSiteValidatorModel
-from gso.utils.types.interfaces import LAGMember, LAGMemberList, PhysicalPortCapacity
+from gso.utils.types.interfaces import BandwidthString, LAGMember, LAGMemberList, PhysicalPortCapacity
 from gso.utils.types.ip_address import (
     AddressSpace,
     IPAddress,
@@ -49,7 +50,7 @@ from gso.utils.types.ip_address import (
     IPV6Netmask,
     PortNumber,
 )
-from gso.utils.types.virtual_identifiers import VLAN_ID
+from gso.utils.types.virtual_identifiers import VC_ID, VLAN_ID
 
 app: typer.Typer = typer.Typer()
 
@@ -327,6 +328,49 @@ class LanSwitchInterconnectImportModel(BaseModel):
     switch_side: LanSwitchInterconnectSwitchSideImportModel
 
 
+class Layer2CircuitServiceImportModel(BaseModel):
+    """Import Layer 2 Circuit Service model."""
+
+    class ServiceBindingPortInput(BaseModel):
+        """Service Binding Port model."""
+
+        edge_port: UUIDstr
+        vlan_id: VLAN_ID
+
+    service_type: Layer2CircuitServiceType
+    partner: str
+    geant_sid: str
+    vc_id: VC_ID
+    layer_2_circuit_side_a: ServiceBindingPortInput
+    layer_2_circuit_side_b: ServiceBindingPortInput
+    layer_2_circuit_type: Layer2CircuitType
+    vlan_range_lower_bound: VLAN_ID | None = None
+    vlan_range_upper_bound: VLAN_ID | None = None
+    policer_enabled: bool = False
+    policer_bandwidth: BandwidthString | None = None
+    policer_burst_rate: BandwidthString | None = None
+
+    @field_validator("partner")
+    def check_if_partner_exists(cls, value: str) -> str:
+        """Validate that the partner exists."""
+        try:
+            get_partner_by_name(value)
+        except PartnerNotFoundError as e:
+            msg = f"Partner {value} not found"
+            raise ValueError(msg) from e
+
+        return value
+
+    @model_validator(mode="after")
+    def check_if_edge_ports_exist(self) -> Self:
+        """Check if the edge ports exist."""
+        for side in [self.layer_2_circuit_side_a, self.layer_2_circuit_side_b]:
+            if not EdgePort.from_subscription(side.edge_port):
+                msg = f"Edge Port {side.edge_port} not found"
+                raise ValueError(msg)
+        return self
+
+
 T = TypeVar(
     "T",
     SiteImportModel,
@@ -339,6 +383,7 @@ T = TypeVar(
     EdgePortImportModel,
     NRENL3CoreServiceImportModel,
     LanSwitchInterconnectImportModel,
+    Layer2CircuitServiceImportModel,
 )
 
 common_filepath_option = typer.Option(
@@ -608,7 +653,7 @@ def import_nren_l3_core_service(filepath: str = common_filepath_option) -> None:
 
     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"])
+        service_type = nren_l3_core_service["service_type"]
         typer.echo(f"Creating imported {service_type} for {partner}")
 
         try:
@@ -650,3 +695,44 @@ def import_lan_switch_interconnect(filepath: str = common_filepath_option) -> No
         "lan_switch_interconnect_description",
         LanSwitchInterconnectImportModel,
     )
+
+
+@app.command()
+def import_layer_2_circuit_service(filepath: str = common_filepath_option) -> None:
+    """Import Layer 2 Circuit services into GSO."""
+    successfully_imported_data = []
+    layer_2_circuit_service_list = _read_data(Path(filepath))
+
+    for layer_2_circuit_service in layer_2_circuit_service_list:
+        partner = layer_2_circuit_service["partner"]
+        service_type = Layer2CircuitServiceType(layer_2_circuit_service["service_type"])
+        typer.echo(f"Creating imported {service_type} for {partner}")
+
+        try:
+            initial_data = Layer2CircuitServiceImportModel(**layer_2_circuit_service)
+            start_process("create_imported_layer_2_circuit", [initial_data.model_dump()])
+            successfully_imported_data.append(initial_data.vc_id)
+            typer.echo(
+                f"Successfully created imported {service_type} with virtual circuit ID {initial_data.vc_id}"
+                f" for {partner}"
+            )
+        except ValidationError as e:
+            typer.echo(f"Validation error: {e}")
+    typer.echo("Waiting for the dust to settle before importing new products...")
+    time.sleep(1)
+
+    # Migrate new products from imported to "full" counterpart.
+    imported_products = get_subscriptions(
+        product_types=[ProductType.IMPORTED_EXPRESSROUTE, ProductType.IMPORTED_GEANT_PLUS],
+        lifecycles=[SubscriptionLifecycle.ACTIVE],
+        includes=["subscription_id"],
+    )
+
+    for subscription_id in imported_products:
+        typer.echo(f"Importing {subscription_id}")
+        start_process("import_layer_2_circuit", [subscription_id])
+
+    if successfully_imported_data:
+        typer.echo("Successfully created imported Layer 2 Circuit services:")
+        for item in successfully_imported_data:
+            typer.echo(f"- {item}")