From d0617eaf450dd7da5c424574c448ae491a8fe8a6 Mon Sep 17 00:00:00 2001
From: Neda Moeini <neda.moeini@geant.org>
Date: Mon, 4 Nov 2024 14:58:25 +0100
Subject: [PATCH] Add unittest for create L2Circuit WF.
---
.../l2_circuit/test_create_layer_2_circuit.py | 70 +++++++++++++++++++
1 file changed, 70 insertions(+)
create mode 100644 test/workflows/l2_circuit/test_create_layer_2_circuit.py
diff --git a/test/workflows/l2_circuit/test_create_layer_2_circuit.py b/test/workflows/l2_circuit/test_create_layer_2_circuit.py
new file mode 100644
index 00000000..cc1d33dc
--- /dev/null
+++ b/test/workflows/l2_circuit/test_create_layer_2_circuit.py
@@ -0,0 +1,70 @@
+import pytest
+from orchestrator.types import SubscriptionLifecycle
+
+from gso.products import ProductName
+from gso.products.product_blocks.layer_2_circuit import Layer2CircuitType
+from gso.products.product_types.layer_2_circuit import Layer2Circuit
+from gso.services.subscriptions import get_product_id_by_name
+from test.workflows import assert_complete, extract_state, run_workflow
+
+
+@pytest.fixture()
+def layer_2_circuit_input(faker, partner_factory, edge_port_subscription_factory, layer_2_circuit_service_type):
+ partner = partner_factory()
+ product_id = get_product_id_by_name(layer_2_circuit_service_type)
+ edge_port_a = edge_port_subscription_factory(partner=partner)
+ edge_port_b = edge_port_subscription_factory(partner=partner)
+ policer_enabled = faker.boolean()
+
+
+ return [
+ {"product": product_id},
+ {
+ "tt_number": faker.tt_number(),
+ "partner": partner["partner_id"],
+ "layer_2_circuit_type": Layer2CircuitType.TAGGED,
+ "policer_enabled": policer_enabled,
+ },
+ {
+ "vlan_range_lower_bound": faker.vlan_id(),
+ "vlan_range_upper_bound": faker.vlan_id(),
+ "policer_bandwidth": faker.bandwidth() if policer_enabled else None,
+ "geant_sid": faker.geant_sid(),
+ "layer_2_circuit_side_a": {"edge_port": edge_port_a, "vlan_id": faker.vlan_id()},
+ "layer_2_circuit_side_b": {"edge_port": edge_port_b, "vlan_id": faker.vlan_id()},
+ },
+ ]
+
+@pytest.mark.parametrize(
+ "layer_2_circuit_service_type", [ProductName.GEANT_PLUS, ProductName.EXPRESSROUTE]
+)
+@pytest.mark.workflow()
+def test_create_layer_2_circuit_success(
+ layer_2_circuit_service_type,
+ layer_2_circuit_input,
+ faker,
+ partner_factory,
+ data_config_filename,
+):
+
+ result, process_stat, step_log = run_workflow("create_layer_2_circuit", layer_2_circuit_input)
+ assert_complete(result)
+ state = extract_state(result)
+ subscription = Layer2Circuit.from_subscription(state["subscription_id"])
+ assert subscription.status == SubscriptionLifecycle.ACTIVE
+ assert subscription.layer_2_circuit.virtual_circuit_id is not None
+ assert len(subscription.layer_2_circuit.layer_2_circuit_sides) == 2
+ assert (
+ str(subscription.layer_2_circuit.layer_2_circuit_sides[0].sbp.edge_port.owner_subscription_id)
+ == layer_2_circuit_input[2]["layer_2_circuit_side_a"]["edge_port"]
+ )
+ assert subscription.layer_2_circuit.layer_2_circuit_sides[0].sbp.is_tagged is True
+ assert subscription.layer_2_circuit.layer_2_circuit_sides[0].sbp.geant_sid == layer_2_circuit_input[2]["geant_sid"]
+ assert (
+ str(subscription.layer_2_circuit.layer_2_circuit_sides[1].sbp.edge_port.owner_subscription_id)
+ == layer_2_circuit_input[2]["layer_2_circuit_side_b"]["edge_port"]
+ )
+ assert subscription.layer_2_circuit.layer_2_circuit_sides[1].sbp.is_tagged is True
+ assert subscription.layer_2_circuit.layer_2_circuit_sides[1].sbp.geant_sid == layer_2_circuit_input[2]["geant_sid"]
+ assert subscription.layer_2_circuit.layer_2_circuit_type == Layer2CircuitType.TAGGED
+ assert subscription.layer_2_circuit.policer_enabled == layer_2_circuit_input[1]["policer_enabled"]
--
GitLab