Add unit tests and input validation for NREN L3 CLI command

# Conflicts:
#	tox.ini

.pylintrc

-[MAIN]
-extension-pkg-whitelist=pydantic
-
-[MISCELLANEOUS]
-
-# List of note tags to take in consideration, separated by a comma.
-# Note that it does not contain TODO, only the default FIXME and XXX
-notes=FIXME,
-      XXX

Changelog.md

@@ -23,12 +23,6 @@
 ## [2.18] - 2024-10-01
 -  Use solo pool for Celery workers
 
-## [2.17] - 2024-09-30
-- NOTHING IS HERE (JENKINS ISSUE)
-
-## [2.16] - 2024-09-30
-- NOTHING IS HERE (JENKINS ISSUE)
-
 ## [2.15] - 2024-09-30
 - Show current license usage when updating Kentik license of a router
 - Fix the bug of clearing all the AE members and creating new objects instead of updating it.

gso/cli/imports.py

@@ -31,6 +31,7 @@ from gso.services.partners import (
     get_partner_by_name,
 )
 from gso.services.subscriptions import (
+    get_active_edge_port_subscriptions,
     get_active_router_subscriptions,
     get_active_subscriptions_by_field_and_value,
     get_subscriptions,
@@ -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.ip_address import IPAddress, IPv4AddressType, IPv6AddressType, PortNumber
+from gso.utils.types.ip_address import IPAddress, IPv4AddressType, IPV4Netmask, IPv6AddressType, IPV6Netmask, PortNumber
 
 app: typer.Typer = typer.Typer()
 
@@ -218,9 +219,6 @@ class EdgePortImportModel(BaseModel):
 class NRENL3CoreServiceImportModel(BaseModel):
     """Import :term:`NREN` L3 Core Service model."""
 
-    partner: str
-    service_binding_ports: list["NRENL3CoreServiceImportModel.ServiceBindingPort"]
-
     class BaseBGPPeer(BaseModel):
         """Base BGP Peer model."""
 
@@ -248,11 +246,37 @@ class NRENL3CoreServiceImportModel(BaseModel):
         vlan_id: VLAN_ID
         custom_firewall_filters: bool = False
         ipv4_address: IPv4AddressType
+        ipv4_mask: IPV4Netmask
         ipv6_address: IPv6AddressType
-        rtbh_enabled: bool = True
+        ipv6_mask: IPV6Netmask
         is_multi_hop: bool = True
         bgp_peers: list["NRENL3CoreServiceImportModel.BaseBGPPeer"]
 
+    partner: str
+    service_binding_ports: list[ServiceBindingPort]
+
+    @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
+
+    @field_validator("service_binding_ports")
+    def validate_node(cls, value: list[ServiceBindingPort]) -> list[ServiceBindingPort]:
+        """Check if the Service Binding Ports are valid."""
+        edge_ports = [str(subscription["subscription_id"]) for subscription in get_active_edge_port_subscriptions()]
+        for sbp in value:
+            if sbp.edge_port not in edge_ports:
+                msg = f"Edge Port {sbp.edge_port} not found"
+                raise ValueError(msg)
+
+        return value
+
 
 T = TypeVar(
     "T",

gso/workflows/edge_port/create_edge_port.py

@@ -29,7 +29,6 @@ from gso.utils.helpers import (
     partner_choice,
     validate_edge_port_number_of_members_based_on_lacp,
 )
-from gso.services.partners import get_partner_by_id
 from gso.utils.types.interfaces import LAGMember, PhysicalPortCapacity
 from gso.utils.types.tt_number import TTNumber
 
@@ -185,7 +184,9 @@ def allocate_interfaces_in_netbox(subscription: EdgePortProvisioning) -> None:
 
 
 @step("[DRY RUN] Create edge port")
-def create_edge_port_dry(subscription: dict[str, Any], tt_number: str, process_id: UUIDstr, partner_name: str) -> LSOState:
+def create_edge_port_dry(
+    subscription: dict[str, Any], tt_number: str, process_id: UUIDstr, partner_name: str
+) -> LSOState:
     """Create a new edge port in the network as a dry run."""
     extra_vars = {
         "dry_run": True,
@@ -203,7 +204,9 @@ def create_edge_port_dry(subscription: dict[str, Any], tt_number: str, process_i
 
 
 @step("[FOR REAL] Create edge port")
-def create_edge_port_real(subscription: dict[str, Any], tt_number: str, process_id: UUIDstr, partner_name: str) -> LSOState:
+def create_edge_port_real(
+    subscription: dict[str, Any], tt_number: str, process_id: UUIDstr, partner_name: str
+) -> LSOState:
     """Create a new edge port in the network for real."""
     extra_vars = {
         "dry_run": False,

gso/workflows/edge_port/create_imported_edge_port.py

@@ -106,7 +106,7 @@ def initialize_subscription(
     target=Target.CREATE,
 )
 def create_imported_edge_port() -> StepList:
-    """Import a Edge Port without provisioning it."""
+    """Import an Edge Port without provisioning it."""
     return (
         begin
         >> create_subscription

gso/workflows/nren_l3_core_service/create_imported_nren_l3_core_service.py

@@ -59,7 +59,7 @@ def initial_input_form_generator() -> FormGenerator:
     class ImportNRENL3CoreServiceForm(FormPage):
         partner: str
         service_binding_ports: list[ServiceBindingPort]
-        nren_l3_core_service_type: NRENL3CoreServiceType
+        service_type: NRENL3CoreServiceType
 
     user_input = yield ImportNRENL3CoreServiceForm
 
@@ -67,12 +67,12 @@ def initial_input_form_generator() -> FormGenerator:
 
 
 @step("Create subscription")
-def create_subscription(partner: str, nren_l3_core_service_type: NRENL3CoreServiceType) -> dict:
+def create_subscription(partner: str, service_type: NRENL3CoreServiceType) -> dict:
     """Create a new subscription object in the database."""
     partner_id = get_partner_by_name(partner)["partner_id"]
-    if nren_l3_core_service_type == NRENL3CoreServiceType.GEANT_IP:
+    if service_type == NRENL3CoreServiceType.GEANT_IP:
         product_id = get_product_id_by_name(ProductName.IMPORTED_GEANT_IP)
-    elif nren_l3_core_service_type == NRENL3CoreServiceType.IAS:
+    elif 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}

pyproject.toml

@@ -116,3 +116,6 @@ filterwarnings = [
     "ignore",
     "default:::gso",
 ]
+
+[tool.coverage.run]
+omit = ["gso/migrations/*"]

test/cli/test_imports.py

@@ -7,12 +7,14 @@ import pytest
 from gso.cli.imports import (
     import_edge_port,
     import_iptrunks,
+    import_nren_l3_core_service,
     import_office_routers,
     import_opengear,
     import_routers,
     import_sites,
     import_super_pop_switches,
 )
+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.router import RouterRole
@@ -233,6 +235,103 @@ def edge_port_data(temp_file, faker, nokia_router_subscription_factory, partner_
     return _edge_port_data
 
 
+@pytest.fixture()
+def nren_l3_core_service_data(temp_file, faker, partner_factory, edge_port_subscription_factory):
+    def _nren_l3_core_service_data(**kwargs):
+        nren_l3_core_service_data = {
+            "partner": partner_factory()["name"],
+            "service_type": "IMPORTED IAS",
+            "service_binding_ports": [
+                {
+                    "edge_port": edge_port_subscription_factory(),
+                    "ap_type": "PRIMARY",
+                    "geant_sid": faker.geant_sid(),
+                    "vlan_id": faker.vlan_id(),
+                    "ipv4_address": faker.ipv4(),
+                    "ipv4_mask": faker.ipv4_netmask(),
+                    "ipv6_address": faker.ipv6(),
+                    "ipv6_mask": faker.ipv6_netmask(),
+                    "bgp_peers": [
+                        {
+                            "bfd_enabled": True,
+                            "bfd_interval": faker.pyint(),
+                            "bfd_multiplier": faker.pyint(),
+                            "has_custom_policies": True,
+                            "authentication_key": faker.password(),
+                            "multipath_enabled": False,
+                            "send_default_route": True,
+                            "is_passive": True,
+                            "peer_address": faker.ipv4(),
+                            "families": [IPFamily.V4UNICAST, IPFamily.V4MULTICAST],
+                            "is_multi_hop": False,
+                            "rtbh_enabled": True,
+                        },
+                        {
+                            "bfd_enabled": True,
+                            "bfd_interval": faker.pyint(),
+                            "bfd_multiplier": faker.pyint(),
+                            "has_custom_policies": True,
+                            "authentication_key": faker.password(),
+                            "multipath_enabled": False,
+                            "send_default_route": True,
+                            "is_passive": True,
+                            "peer_address": faker.ipv6(),
+                            "families": [IPFamily.V6UNICAST],
+                            "is_multi_hop": False,
+                            "rtbh_enabled": True,
+                        },
+                    ],
+                },
+                {
+                    "edge_port": edge_port_subscription_factory(),
+                    "ap_type": "BACKUP",
+                    "geant_sid": faker.geant_sid(),
+                    "vlan_id": faker.vlan_id(),
+                    "ipv4_address": faker.ipv4(),
+                    "ipv4_mask": faker.ipv4_netmask(),
+                    "ipv6_address": faker.ipv6(),
+                    "ipv6_mask": faker.ipv6_netmask(),
+                    "bgp_peers": [
+                        {
+                            "bfd_enabled": True,
+                            "bfd_interval": faker.pyint(),
+                            "bfd_multiplier": faker.pyint(),
+                            "has_custom_policies": True,
+                            "authentication_key": faker.password(),
+                            "multipath_enabled": False,
+                            "send_default_route": True,
+                            "is_passive": True,
+                            "peer_address": faker.ipv4(),
+                            "families": [IPFamily.V4UNICAST, IPFamily.V4MULTICAST],
+                            "is_multi_hop": False,
+                            "rtbh_enabled": True,
+                        },
+                        {
+                            "bfd_enabled": True,
+                            "bfd_interval": faker.pyint(),
+                            "bfd_multiplier": faker.pyint(),
+                            "has_custom_policies": True,
+                            "authentication_key": faker.password(),
+                            "multipath_enabled": False,
+                            "send_default_route": True,
+                            "is_passive": True,
+                            "peer_address": faker.ipv6(),
+                            "families": [IPFamily.V6UNICAST],
+                            "is_multi_hop": False,
+                            "rtbh_enabled": True,
+                        },
+                    ],
+                },
+            ],
+        }
+        nren_l3_core_service_data.update(**kwargs)
+
+        temp_file.write_text(json.dumps([nren_l3_core_service_data]))
+        return {"path": str(temp_file), "data": nren_l3_core_service_data}
+
+    return _nren_l3_core_service_data
+
+
 ###########
 #  TESTS  #
 ###########
@@ -447,3 +546,96 @@ def test_import_edge_port_with_invalid_partner(mock_start_process, mock_sleep, e
     captured_output, _ = capfd.readouterr()
     assert "Partner INVALID not found" in captured_output
     assert mock_start_process.call_count == 0
+
+
+@patch("gso.cli.imports.time.sleep")
+@patch("gso.cli.imports.start_process")
+def test_import_nren_l3_core_service_success(mock_start_process, mock_sleep, nren_l3_core_service_data, capfd):
+    import_nren_l3_core_service(nren_l3_core_service_data()["path"])
+    assert mock_start_process.call_count == 1
+
+
+@patch("gso.cli.imports.time.sleep")
+@patch("gso.cli.imports.start_process")
+def test_import_nren_l3_core_service_with_invalid_partner(
+    mock_start_process, mock_sleep, nren_l3_core_service_data, capfd
+):
+    broken_data = nren_l3_core_service_data(partner="INVALID")
+    import_nren_l3_core_service(broken_data["path"])
+
+    captured_output, _ = capfd.readouterr()
+    assert "Partner INVALID not found" in captured_output
+    assert mock_start_process.call_count == 0
+
+
+@patch("gso.cli.imports.time.sleep")
+@patch("gso.cli.imports.start_process")
+def test_import_nren_l3_core_service_with_invalid_edge_port(
+    mock_start_process, mock_sleep, faker, nren_l3_core_service_data, edge_port_subscription_factory, capfd
+):
+    fake_uuid = faker.uuid4()
+    broken_data = nren_l3_core_service_data(
+        service_binding_ports=[
+            {
+                "edge_port": fake_uuid,
+                "ap_type": "PRIMARY",
+                "geant_sid": faker.geant_sid(),
+                "vlan_id": faker.vlan_id(),
+                "ipv4_address": faker.ipv4(),
+                "ipv4_mask": faker.ipv4_netmask(),
+                "ipv6_address": faker.ipv6(),
+                "ipv6_mask": faker.ipv6_netmask(),
+                "bgp_peers": [
+                    {
+                        "bfd_enabled": False,
+                        "authentication_key": faker.password(),
+                        "peer_address": faker.ipv4(),
+                        "families": [IPFamily.V4UNICAST],
+                        "is_multi_hop": False,
+                        "rtbh_enabled": True,
+                    },
+                    {
+                        "bfd_enabled": False,
+                        "authentication_key": faker.password(),
+                        "peer_address": faker.ipv6(),
+                        "families": [IPFamily.V6UNICAST],
+                        "is_multi_hop": False,
+                        "rtbh_enabled": True,
+                    },
+                ],
+            },
+            {
+                "edge_port": edge_port_subscription_factory(),
+                "ap_type": "BACKUP",
+                "geant_sid": faker.geant_sid(),
+                "vlan_id": faker.vlan_id(),
+                "ipv4_address": faker.ipv4(),
+                "ipv4_mask": faker.ipv4_netmask(),
+                "ipv6_address": faker.ipv6(),
+                "ipv6_mask": faker.ipv6_netmask(),
+                "bgp_peers": [
+                    {
+                        "bfd_enabled": False,
+                        "authentication_key": faker.password(),
+                        "peer_address": faker.ipv4(),
+                        "families": [IPFamily.V4UNICAST],
+                        "is_multi_hop": False,
+                        "rtbh_enabled": True,
+                    },
+                    {
+                        "bfd_enabled": False,
+                        "authentication_key": faker.password(),
+                        "peer_address": faker.ipv6(),
+                        "families": [IPFamily.V6UNICAST],
+                        "is_multi_hop": False,
+                        "rtbh_enabled": True,
+                    },
+                ],
+            },
+        ]
+    )
+    import_nren_l3_core_service(broken_data["path"])
+
+    captured_output, _ = capfd.readouterr()
+    assert f"Edge Port {fake_uuid} not found" in captured_output
+    assert mock_start_process.call_count == 0

test/fixtures/nren_l3_core_service_fixtures.py

@@ -76,7 +76,7 @@ def service_binding_port_factory(faker, bgp_session_subscription_factory, edge_p
         return ServiceBindingPort.new(
             subscription_id=uuid4(),
             is_tagged=is_tagged,
-            vlan_id=vlan_id or faker.pyint(min_value=1, max_value=4096),
+            vlan_id=vlan_id or faker.vlan_id(),
             sbp_type=sbp_type,
             ipv4_address=ipv4_address or faker.ipv4(),
             ipv4_mask=ipv4_mask or faker.ipv4_netmask(),

test/workflows/nren_l3_core_service/test_create_imported_nren_l3_core_service.py

@@ -13,7 +13,7 @@ def test_create_imported_nren_l3_core_service_success(
 ):
     creation_form_input_data = {
         "partner": partner_factory()["name"],
-        "nren_l3_core_service_type": l3_core_service_type,
+        "service_type": l3_core_service_type,
         "service_binding_ports": [
             {
                 "edge_port": edge_port_subscription_factory(),