From 3f4397227da05eb890ff068c54efe4230dad15b2 Mon Sep 17 00:00:00 2001
From: Karel van Klink <karel.vanklink@geant.org>
Date: Tue, 8 Oct 2024 14:17:35 +0200
Subject: [PATCH] =?UTF-8?q?Add=20unit=20test=20for=20G=C3=89ANT=20IP=20cre?=
=?UTF-8?q?ation?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit
---
gso/products/__init__.py | 4 +--
.../geant_ip/test_create_geant_ip.py | 30 +++++++++++++++++--
2 files changed, 29 insertions(+), 5 deletions(-)
diff --git a/gso/products/__init__.py b/gso/products/__init__.py
index 88ea8b18..09d0333b 100644
--- a/gso/products/__init__.py
+++ b/gso/products/__init__.py
@@ -88,7 +88,7 @@ SUBSCRIPTION_MODEL_REGISTRY.update(
ProductName.IMPORTED_OPENGEAR.value: ImportedOpengear,
ProductName.EDGE_PORT.value: EdgePort,
ProductName.IMPORTED_EDGE_PORT.value: ImportedEdgePort,
- ProductType.GEANT_IP.value: GeantIP,
- ProductType.IMPORTED_GEANT_IP.value: ImportedGeantIP,
+ ProductName.GEANT_IP.value: GeantIP,
+ ProductName.IMPORTED_GEANT_IP.value: ImportedGeantIP,
},
)
diff --git a/test/workflows/geant_ip/test_create_geant_ip.py b/test/workflows/geant_ip/test_create_geant_ip.py
index 2d030660..90e237e9 100644
--- a/test/workflows/geant_ip/test_create_geant_ip.py
+++ b/test/workflows/geant_ip/test_create_geant_ip.py
@@ -1,9 +1,13 @@
+from unittest.mock import patch
+
import pytest
+from orchestrator.types import SubscriptionLifecycle
from gso.products import ProductName
+from gso.products.product_types.geant_ip import GeantIP
from gso.services.subscriptions import get_product_id_by_name
from gso.utils.shared_enums import APType
-from test.workflows import assert_complete, run_workflow
+from test.workflows import assert_complete, assert_lso_interaction_success, extract_state, run_workflow
@pytest.fixture()
@@ -25,8 +29,15 @@ def base_bgp_peer_input(faker):
@pytest.mark.workflow()
+@patch("gso.services.lso_client._send_request")
def test_create_geant_ip_success(
- responses, faker, partner_factory, edge_port_subscription_factory, base_bgp_peer_input
+ mock_lso_client,
+ responses,
+ faker,
+ partner_factory,
+ edge_port_subscription_factory,
+ base_bgp_peer_input,
+ data_config_filename,
):
partner = partner_factory(name=faker.company(), email=faker.email())
product_id = get_product_id_by_name(ProductName.GEANT_IP)
@@ -47,7 +58,20 @@ def test_create_geant_ip_success(
"v6_bgp_peer": base_bgp_peer_input() | {"add_v6_multicast": faker.boolean(), "peer_address": faker.ipv6()},
},
]
+ lso_interaction_count = 6
result, process_stat, step_log = run_workflow("create_geant_ip", form_input_data)
- assert process_stat, step_log
+
+ for _ in range(lso_interaction_count):
+ result, step_log = assert_lso_interaction_success(result, process_stat, step_log)
+
assert_complete(result)
+ state = extract_state(result)
+ subscription = GeantIP.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 (
+ str(subscription.geant_ip.geant_ip_ap_list[0].geant_ip_sbp.edge_port.owner_subscription_id)
+ == form_input_data[2]["edge_ports"][0]["edge_port"]
+ )
--
GitLab