Skip to content
Snippets Groups Projects
Verified Commit 2b83e40e authored by Neda Moeini's avatar Neda Moeini Committed by Karel van Klink
Browse files

Update create L3 core service and L2 Circuit WFs

parent e0fccf00
No related branches found
No related tags found
No related merge requests found
This commit is part of merge request !324. Comments created here will be created in the context of that merge request.
...@@ -19,6 +19,7 @@ from gso.products.product_blocks.service_binding_port import ServiceBindingPortI ...@@ -19,6 +19,7 @@ from gso.products.product_blocks.service_binding_port import ServiceBindingPortI
from gso.products.product_types.edge_port import EdgePort from gso.products.product_types.edge_port import EdgePort
from gso.products.product_types.layer_2_circuit import Layer2Circuit, Layer2CircuitInactive from gso.products.product_types.layer_2_circuit import Layer2Circuit, Layer2CircuitInactive
from gso.services.partners import get_partner_by_name from gso.services.partners import get_partner_by_name
from gso.services.subscriptions import generate_unique_gs_id
from gso.utils.helpers import active_edge_port_selector, generate_unique_vc_id, partner_choice from gso.utils.helpers import active_edge_port_selector, generate_unique_vc_id, partner_choice
from gso.utils.shared_enums import SBPType from gso.utils.shared_enums import SBPType
from gso.utils.types.interfaces import BandwidthString from gso.utils.types.interfaces import BandwidthString
...@@ -69,7 +70,6 @@ def initial_input_generator(product_name: str) -> FormGenerator: ...@@ -69,7 +70,6 @@ def initial_input_generator(product_name: str) -> FormGenerator:
vlan_divider: Divider = Field(None, exclude=True) vlan_divider: Divider = Field(None, exclude=True)
policer_bandwidth: _policer_field(policer_enabled=initial_user_input.policer_enabled) # type: ignore[valid-type] policer_bandwidth: _policer_field(policer_enabled=initial_user_input.policer_enabled) # type: ignore[valid-type]
policer_burst_rate: _policer_field(policer_enabled=initial_user_input.policer_enabled) # type: ignore[valid-type] policer_burst_rate: _policer_field(policer_enabled=initial_user_input.policer_enabled) # type: ignore[valid-type]
gs_id: str
layer_2_circuit_side_a: Layer2CircuitSideSelection layer_2_circuit_side_a: Layer2CircuitSideSelection
side_divider: Divider = Field(None, exclude=True) side_divider: Divider = Field(None, exclude=True)
layer_2_circuit_side_b: Layer2CircuitSideSelection layer_2_circuit_side_b: Layer2CircuitSideSelection
...@@ -105,10 +105,10 @@ def initialize_subscription( ...@@ -105,10 +105,10 @@ def initialize_subscription(
policer_enabled: bool, # noqa: FBT001 policer_enabled: bool, # noqa: FBT001
policer_bandwidth: BandwidthString | None, policer_bandwidth: BandwidthString | None,
policer_burst_rate: BandwidthString | None, policer_burst_rate: BandwidthString | None,
gs_id: str,
) -> State: ) -> State:
"""Build a subscription object from all user input.""" """Build a subscription object from all user input."""
layer_2_circuit_sides = [] layer_2_circuit_sides = []
gs_id = generate_unique_gs_id()
for circuit_side_data in [layer_2_circuit_side_a, layer_2_circuit_side_b]: for circuit_side_data in [layer_2_circuit_side_a, layer_2_circuit_side_b]:
sbp = ServiceBindingPortInactive.new( sbp = ServiceBindingPortInactive.new(
uuid4(), uuid4(),
......
...@@ -20,6 +20,7 @@ from gso.products.product_types.edge_port import EdgePort ...@@ -20,6 +20,7 @@ from gso.products.product_types.edge_port import EdgePort
from gso.products.product_types.l3_core_service import L3CoreService, L3CoreServiceInactive from gso.products.product_types.l3_core_service import L3CoreService, L3CoreServiceInactive
from gso.services.lso_client import LSOState, lso_interaction from gso.services.lso_client import LSOState, lso_interaction
from gso.services.partners import get_partner_by_id from gso.services.partners import get_partner_by_id
from gso.services.subscriptions import generate_unique_gs_id
from gso.utils.helpers import ( from gso.utils.helpers import (
active_edge_port_selector, active_edge_port_selector,
partner_choice, partner_choice,
...@@ -113,7 +114,6 @@ def initial_input_form_generator(product_name: str) -> FormGenerator: ...@@ -113,7 +114,6 @@ def initial_input_form_generator(product_name: str) -> FormGenerator:
exclude=True, exclude=True,
) )
gs_id: str
is_tagged: bool = False is_tagged: bool = False
vlan_id: VLAN_ID vlan_id: VLAN_ID
custom_firewall_filters: bool = False custom_firewall_filters: bool = False
...@@ -165,6 +165,7 @@ def initialize_subscription( ...@@ -165,6 +165,7 @@ def initialize_subscription(
BGPSession.new(subscription_id=uuid4(), rtbh_enabled=True, is_multi_hop=True, **session) BGPSession.new(subscription_id=uuid4(), rtbh_enabled=True, is_multi_hop=True, **session)
for session in binding_port_input["bgp_peers"] for session in binding_port_input["bgp_peers"]
] ]
sbp_gs_id = generate_unique_gs_id()
service_binding_port = ServiceBindingPortInactive.new( service_binding_port = ServiceBindingPortInactive.new(
subscription_id=uuid4(), subscription_id=uuid4(),
v4_bfd_settings=BFDSettings.new(subscription_id=uuid4(), **(binding_port_input.pop("v4_bfd_settings"))), v4_bfd_settings=BFDSettings.new(subscription_id=uuid4(), **(binding_port_input.pop("v4_bfd_settings"))),
...@@ -173,6 +174,7 @@ def initialize_subscription( ...@@ -173,6 +174,7 @@ def initialize_subscription(
bgp_session_list=sbp_bgp_session_list, bgp_session_list=sbp_bgp_session_list,
sbp_type=SBPType.L3, sbp_type=SBPType.L3,
edge_port=edge_port_subscription.edge_port, edge_port=edge_port_subscription.edge_port,
gs_id=sbp_gs_id,
) )
subscription.l3_core_service.ap_list.append( subscription.l3_core_service.ap_list.append(
AccessPortInactive.new( AccessPortInactive.new(
......
...@@ -28,7 +28,6 @@ def layer_2_circuit_input(faker, partner_factory, edge_port_subscription_factory ...@@ -28,7 +28,6 @@ def layer_2_circuit_input(faker, partner_factory, edge_port_subscription_factory
"vlan_range_upper_bound": faker.vlan_id(), "vlan_range_upper_bound": faker.vlan_id(),
"policer_bandwidth": faker.bandwidth() if policer_enabled else None, "policer_bandwidth": faker.bandwidth() if policer_enabled else None,
"policer_burst_rate": faker.bandwidth() if policer_enabled else None, "policer_burst_rate": faker.bandwidth() if policer_enabled else None,
"gs_id": faker.geant_sid(),
"layer_2_circuit_side_a": {"edge_port": edge_port_a, "vlan_id": faker.vlan_id()}, "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()}, "layer_2_circuit_side_b": {"edge_port": edge_port_b, "vlan_id": faker.vlan_id()},
}, },
...@@ -56,13 +55,15 @@ def test_create_layer_2_circuit_success( ...@@ -56,13 +55,15 @@ def test_create_layer_2_circuit_success(
== layer_2_circuit_input[2]["layer_2_circuit_side_a"]["edge_port"] == 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.is_tagged is True
assert subscription.layer_2_circuit.layer_2_circuit_sides[0].sbp.gs_id == layer_2_circuit_input[2]["gs_id"] assert (
subscription.layer_2_circuit.layer_2_circuit_sides[0].sbp.gs_id
== subscription.layer_2_circuit.layer_2_circuit_sides[1].sbp.gs_id
)
assert ( assert (
str(subscription.layer_2_circuit.layer_2_circuit_sides[1].sbp.edge_port.owner_subscription_id) 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"] == 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.is_tagged is True
assert subscription.layer_2_circuit.layer_2_circuit_sides[1].sbp.gs_id == layer_2_circuit_input[2]["gs_id"]
assert subscription.layer_2_circuit.layer_2_circuit_type == Layer2CircuitType.TAGGED assert subscription.layer_2_circuit.layer_2_circuit_type == Layer2CircuitType.TAGGED
assert subscription.layer_2_circuit.vlan_range_lower_bound == layer_2_circuit_input[2]["vlan_range_lower_bound"] assert subscription.layer_2_circuit.vlan_range_lower_bound == layer_2_circuit_input[2]["vlan_range_lower_bound"]
assert subscription.layer_2_circuit.vlan_range_upper_bound == layer_2_circuit_input[2]["vlan_range_upper_bound"] assert subscription.layer_2_circuit.vlan_range_upper_bound == layer_2_circuit_input[2]["vlan_range_upper_bound"]
......
...@@ -49,7 +49,6 @@ def test_create_l3_core_service_success( ...@@ -49,7 +49,6 @@ def test_create_l3_core_service_success(
{"tt_number": faker.tt_number(), "partner": partner["partner_id"]}, {"tt_number": faker.tt_number(), "partner": partner["partner_id"]},
{"edge_port": {"edge_port": edge_port_a, "ap_type": APType.PRIMARY}}, {"edge_port": {"edge_port": edge_port_a, "ap_type": APType.PRIMARY}},
{ {
"gs_id": faker.geant_sid(),
"is_tagged": faker.boolean(), "is_tagged": faker.boolean(),
"vlan_id": faker.vlan_id(), "vlan_id": faker.vlan_id(),
"ipv4_address": faker.ipv4(), "ipv4_address": faker.ipv4(),
......
0% Loading or .
You are about to add 0 people to the discussion. Proceed with caution.
Please register or to comment