From be1627dd487639fbde45d939c69c96be4a8267fc Mon Sep 17 00:00:00 2001
From: Karel van Klink <karel.vanklink@geant.org>
Date: Mon, 21 Oct 2024 14:35:31 +0200
Subject: [PATCH] Add unit tests and input validation for NREN L3 CLI command
# Conflicts:
# tox.ini
---
.pylintrc | 9 -
Changelog.md | 6 -
gso/cli/imports.py | 34 +++-
gso/workflows/edge_port/create_edge_port.py | 9 +-
.../edge_port/create_imported_edge_port.py | 2 +-
.../create_imported_nren_l3_core_service.py | 8 +-
pyproject.toml | 3 +
test/cli/test_imports.py | 192 ++++++++++++++++++
.../fixtures/nren_l3_core_service_fixtures.py | 2 +-
...st_create_imported_nren_l3_core_service.py | 2 +-
10 files changed, 237 insertions(+), 30 deletions(-)
delete mode 100644 .pylintrc
diff --git a/.pylintrc b/.pylintrc
deleted file mode 100644
index d5b83130..00000000
--- a/.pylintrc
+++ /dev/null
@@ -1,9 +0,0 @@
-[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
diff --git a/Changelog.md b/Changelog.md
index b0b8daef..07061198 100644
--- a/Changelog.md
+++ b/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.
diff --git a/gso/cli/imports.py b/gso/cli/imports.py
index 5415e60c..f1553d22 100644
--- a/gso/cli/imports.py
+++ b/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",
diff --git a/gso/workflows/edge_port/create_edge_port.py b/gso/workflows/edge_port/create_edge_port.py
index 97b95577..40ebd4b0 100644
--- a/gso/workflows/edge_port/create_edge_port.py
+++ b/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,
diff --git a/gso/workflows/edge_port/create_imported_edge_port.py b/gso/workflows/edge_port/create_imported_edge_port.py
index 5f9fe262..b932175f 100644
--- a/gso/workflows/edge_port/create_imported_edge_port.py
+++ b/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
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 0bad4c86..f5a2e2ea 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
@@ -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}
diff --git a/pyproject.toml b/pyproject.toml
index e9979242..2ae64df6 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -116,3 +116,6 @@ filterwarnings = [
"ignore",
"default:::gso",
]
+
+[tool.coverage.run]
+omit = ["gso/migrations/*"]
diff --git a/test/cli/test_imports.py b/test/cli/test_imports.py
index 55f7db15..b0ec0adb 100644
--- a/test/cli/test_imports.py
+++ b/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
diff --git a/test/fixtures/nren_l3_core_service_fixtures.py b/test/fixtures/nren_l3_core_service_fixtures.py
index 059f2d37..d0fd069a 100644
--- a/test/fixtures/nren_l3_core_service_fixtures.py
+++ b/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(),
diff --git a/test/workflows/nren_l3_core_service/test_create_imported_nren_l3_core_service.py b/test/workflows/nren_l3_core_service/test_create_imported_nren_l3_core_service.py
index 05982ef1..63aab373 100644
--- a/test/workflows/nren_l3_core_service/test_create_imported_nren_l3_core_service.py
+++ b/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(),
--
GitLab