Skip to content
Snippets Groups Projects
Verified Commit 501326b1 authored by Karel van Klink's avatar Karel van Klink :smiley_cat:
Browse files

Rename NRENL3CoreService 

Rename to L3CoreService, and add Copernicus, LHCOne, and GWS as new L3 core services
parent d27d1e42
No related branches found
No related tags found
2 merge requests!313Rename NRENL3CoreService and update BGP peer,!312Rename NRENL3CoreService
Pipeline #90566 failed
Stage: tox
Stage: documentation
Stage: sonarqube
Hide whitespace changes
Inline Side-by-side
Showing
with 191 additions and 84 deletions
test/workflows/nren_l3_core_service/test_create_imported_nren_l3_core_service.py → test/workflows/l3_core_service/test_create_imported_l3_core_service.py
+ 14
− 5
View file @ 501326b1
...@@ -2,13 +2,22 @@ import pytest ...@@ -2,13 +2,22 @@ import pytest
from orchestrator.types import SubscriptionLifecycle from orchestrator.types import SubscriptionLifecycle
from gso.products.product_blocks.bgp_session import IPFamily from gso.products.product_blocks.bgp_session import IPFamily
from gso.products.product_types.nren_l3_core_service import ImportedNRENL3CoreService, NRENL3CoreServiceType from gso.products.product_types.l3_core_service import ImportedL3CoreService, L3CoreServiceType
from gso.utils.shared_enums import SBPType from gso.utils.shared_enums import SBPType
from test.workflows import assert_complete, extract_state, run_workflow from test.workflows import assert_complete, extract_state, run_workflow
@pytest.mark.parametrize("l3_core_service_type", [NRENL3CoreServiceType.GEANT_IP, NRENL3CoreServiceType.IAS]) @pytest.mark.parametrize(
def test_create_imported_nren_l3_core_service_success( "l3_core_service_type",
[
L3CoreServiceType.GEANT_IP,
L3CoreServiceType.IAS,
L3CoreServiceType.GWS,
L3CoreServiceType.LHCONE,
L3CoreServiceType.COPERNICUS,
],
)
def test_create_imported_l3_core_service_success(
faker, partner_factory, edge_port_subscription_factory, l3_core_service_type faker, partner_factory, edge_port_subscription_factory, l3_core_service_type
): ):
creation_form_input_data = { creation_form_input_data = {
...@@ -69,8 +78,8 @@ def test_create_imported_nren_l3_core_service_success( ...@@ -69,8 +78,8 @@ def test_create_imported_nren_l3_core_service_success(
], ],
} }
result, _, _ = run_workflow("create_imported_nren_l3_core_service", [creation_form_input_data]) result, _, _ = run_workflow("create_imported_l3_core_service", [creation_form_input_data])
state = extract_state(result) state = extract_state(result)
subscription = ImportedNRENL3CoreService.from_subscription(state["subscription_id"]) subscription = ImportedL3CoreService.from_subscription(state["subscription_id"])
assert_complete(result) assert_complete(result)
assert subscription.status == SubscriptionLifecycle.ACTIVE assert subscription.status == SubscriptionLifecycle.ACTIVE
test/workflows/nren_l3_core_service/test_create_nren_l3_core_service.py → test/workflows/l3_core_service/test_create_l3_core_service.py
+ 9
− 7
View file @ 501326b1
...@@ -4,7 +4,7 @@ import pytest ...@@ -4,7 +4,7 @@ import pytest
from orchestrator.types import SubscriptionLifecycle from orchestrator.types import SubscriptionLifecycle
from gso.products import ProductName from gso.products import ProductName
from gso.products.product_types.nren_l3_core_service import NRENL3CoreService from gso.products.product_types.l3_core_service import L3CoreService
from gso.services.subscriptions import get_product_id_by_name from gso.services.subscriptions import get_product_id_by_name
from gso.utils.shared_enums import APType from gso.utils.shared_enums import APType
from test.workflows import assert_complete, assert_lso_interaction_success, extract_state, run_workflow from test.workflows import assert_complete, assert_lso_interaction_success, extract_state, run_workflow
...@@ -26,10 +26,12 @@ def base_bgp_peer_input(faker): ...@@ -26,10 +26,12 @@ def base_bgp_peer_input(faker):
return _base_bgp_peer_input return _base_bgp_peer_input
@pytest.mark.parametrize("l3_core_type", [ProductName.GEANT_IP, ProductName.IAS]) @pytest.mark.parametrize(
"l3_core_type", [ProductName.GEANT_IP, ProductName.IAS, ProductName.GWS, ProductName.LHCONE, ProductName.COPERNICUS]
)
@pytest.mark.workflow() @pytest.mark.workflow()
@patch("gso.services.lso_client._send_request") @patch("gso.services.lso_client._send_request")
def test_create_nren_l3_core_service_success( def test_create_l3_core_service_success(
mock_lso_client, mock_lso_client,
l3_core_type, l3_core_type,
faker, faker,
...@@ -73,18 +75,18 @@ def test_create_nren_l3_core_service_success( ...@@ -73,18 +75,18 @@ def test_create_nren_l3_core_service_success(
] ]
lso_interaction_count = 6 lso_interaction_count = 6
result, process_stat, step_log = run_workflow("create_nren_l3_core_service", form_input_data) result, process_stat, step_log = run_workflow("create_l3_core_service", form_input_data)
for _ in range(lso_interaction_count): for _ in range(lso_interaction_count):
result, step_log = assert_lso_interaction_success(result, process_stat, step_log) result, step_log = assert_lso_interaction_success(result, process_stat, step_log)
assert_complete(result) assert_complete(result)
state = extract_state(result) state = extract_state(result)
subscription = NRENL3CoreService.from_subscription(state["subscription_id"]) subscription = L3CoreService.from_subscription(state["subscription_id"])
assert mock_lso_client.call_count == lso_interaction_count assert mock_lso_client.call_count == lso_interaction_count
assert subscription.status == SubscriptionLifecycle.ACTIVE assert subscription.status == SubscriptionLifecycle.ACTIVE
assert len(subscription.nren_l3_core_service.nren_ap_list) == 1 assert len(subscription.l3_core_service.ap_list) == 1
assert ( assert (
str(subscription.nren_l3_core_service.nren_ap_list[0].sbp.edge_port.owner_subscription_id) str(subscription.l3_core_service.ap_list[0].sbp.edge_port.owner_subscription_id)
== form_input_data[2]["edge_port"]["edge_port"] == form_input_data[2]["edge_port"]["edge_port"]
) )
test/workflows/nren_l3_core_service/test_import_nren_l3_core_service.py → test/workflows/l3_core_service/test_import_l3_core_service.py
+ 27
− 0
View file @ 501326b1
import pytest import pytest
from orchestrator.types import SubscriptionLifecycle from orchestrator.types import SubscriptionLifecycle
from gso.products.product_types.nren_l3_core_service import NRENL3CoreService, NRENL3CoreServiceType from gso.products.product_types.l3_core_service import L3CoreService, L3CoreServiceType
from test.workflows import assert_complete, run_workflow from test.workflows import assert_complete, run_workflow
@pytest.mark.parametrize( @pytest.mark.parametrize(
"l3_core_service_type", [NRENL3CoreServiceType.IMPORTED_GEANT_IP, NRENL3CoreServiceType.IMPORTED_IAS] "l3_core_service_type",
[
L3CoreServiceType.IMPORTED_GEANT_IP,
L3CoreServiceType.IMPORTED_IAS,
L3CoreServiceType.IMPORTED_LHCONE,
L3CoreServiceType.IMPORTED_COPERNICUS,
],
) )
@pytest.mark.workflow() @pytest.mark.workflow()
def test_import_nren_l3_core_service_success(nren_l3_core_service_subscription_factory, l3_core_service_type): def test_import_l3_core_service_success(l3_core_service_subscription_factory, l3_core_service_type):
imported_nren_l3_core_service = nren_l3_core_service_subscription_factory( imported_l3_core_service = l3_core_service_subscription_factory(l3_core_service_type=l3_core_service_type)
nren_l3_core_service_type=l3_core_service_type result, _, _ = run_workflow("import_l3_core_service", [{"subscription_id": imported_l3_core_service}])
) subscription = L3CoreService.from_subscription(imported_l3_core_service)
result, _, _ = run_workflow("import_nren_l3_core_service", [{"subscription_id": imported_nren_l3_core_service}])
subscription = NRENL3CoreService.from_subscription(imported_nren_l3_core_service)
assert_complete(result) assert_complete(result)
# Remove the "IMPORTED_" prefix with ``[9:]`` # Remove the "IMPORTED_" prefix with ``[9:]``
assert subscription.nren_l3_core_service_type == NRENL3CoreServiceType(l3_core_service_type.value[9:]) assert subscription.l3_core_service_type == L3CoreServiceType(l3_core_service_type.value[9:])
assert subscription.status == SubscriptionLifecycle.ACTIVE assert subscription.status == SubscriptionLifecycle.ACTIVE
assert subscription.insync is True assert subscription.insync is True
test/workflows/nren_l3_core_service/test_migrate_nren_l3_core_service.py → test/workflows/l3_core_service/test_migrate_l3_core_service.py
+ 56
− 0
View file @ 501326b1
import pytest import pytest
from gso.products.product_types.nren_l3_core_service import NRENL3CoreService, NRENL3CoreServiceType from gso.products.product_types.l3_core_service import L3CoreService, L3CoreServiceType
from test.workflows import assert_complete, extract_state, run_workflow from test.workflows import assert_complete, extract_state, run_workflow
@pytest.mark.parametrize("l3_core_service_type", [NRENL3CoreServiceType.GEANT_IP, NRENL3CoreServiceType.IAS]) @pytest.mark.parametrize(
"l3_core_service_type",
[
L3CoreServiceType.GEANT_IP,
L3CoreServiceType.IAS,
L3CoreServiceType.GWS,
L3CoreServiceType.LHCONE,
L3CoreServiceType.COPERNICUS,
],
)
@pytest.mark.workflow() @pytest.mark.workflow()
def test_migrate_nren_l3_core_service_success( def test_migrate_l3_core_service_success(
faker, faker,
edge_port_subscription_factory, edge_port_subscription_factory,
partner_factory, partner_factory,
nren_l3_core_service_subscription_factory, l3_core_service_subscription_factory,
l3_core_service_type, l3_core_service_type,
): ):
partner = partner_factory() partner = partner_factory()
subscription_id = nren_l3_core_service_subscription_factory( subscription_id = l3_core_service_subscription_factory(partner=partner, l3_core_service_type=l3_core_service_type)
partner=partner, nren_l3_core_service_type=l3_core_service_type
)
new_edge_port_1 = edge_port_subscription_factory(partner=partner) new_edge_port_1 = edge_port_subscription_factory(partner=partner)
new_edge_port_2 = edge_port_subscription_factory(partner=partner) new_edge_port_2 = edge_port_subscription_factory(partner=partner)
subscription = NRENL3CoreService.from_subscription(subscription_id) subscription = L3CoreService.from_subscription(subscription_id)
form_input_data = [ form_input_data = [
{"subscription_id": subscription_id}, {"subscription_id": subscription_id},
...@@ -27,23 +34,23 @@ def test_migrate_nren_l3_core_service_success( ...@@ -27,23 +34,23 @@ def test_migrate_nren_l3_core_service_success(
"tt_number": faker.tt_number(), "tt_number": faker.tt_number(),
"edge_port_selection": [ "edge_port_selection": [
{ {
"old_edge_port": subscription.nren_l3_core_service.nren_ap_list[0].sbp.edge_port.description, "old_edge_port": subscription.l3_core_service.ap_list[0].sbp.edge_port.description,
"new_edge_port": new_edge_port_1, "new_edge_port": new_edge_port_1,
}, },
{ {
"old_edge_port": subscription.nren_l3_core_service.nren_ap_list[1].sbp.edge_port.description, "old_edge_port": subscription.l3_core_service.ap_list[1].sbp.edge_port.description,
"new_edge_port": new_edge_port_2, "new_edge_port": new_edge_port_2,
}, },
], ],
}, },
] ]
result, _, _ = run_workflow("migrate_nren_l3_core_service", form_input_data) result, _, _ = run_workflow("migrate_l3_core_service", form_input_data)
assert_complete(result) assert_complete(result)
state = extract_state(result) state = extract_state(result)
subscription = NRENL3CoreService.from_subscription(state["subscription_id"]) subscription = L3CoreService.from_subscription(state["subscription_id"])
assert subscription.insync is True assert subscription.insync is True
assert len(subscription.nren_l3_core_service.nren_ap_list) == 2 assert len(subscription.l3_core_service.ap_list) == 2
assert str(subscription.nren_l3_core_service.nren_ap_list[0].sbp.edge_port.owner_subscription_id) == new_edge_port_1 assert str(subscription.l3_core_service.ap_list[0].sbp.edge_port.owner_subscription_id) == new_edge_port_1
assert str(subscription.nren_l3_core_service.nren_ap_list[1].sbp.edge_port.owner_subscription_id) == new_edge_port_2 assert str(subscription.l3_core_service.ap_list[1].sbp.edge_port.owner_subscription_id) == new_edge_port_2
test/workflows/nren_l3_core_service/test_modify_nren_l3_core_service.py → test/workflows/l3_core_service/test_modify_l3_core_service.py
+ 85
− 72
View file @ 501326b1
import pytest import pytest
from gso.products.product_blocks.bgp_session import IPFamily from gso.products.product_blocks.bgp_session import IPFamily
from gso.products.product_types.nren_l3_core_service import NRENL3CoreService, NRENL3CoreServiceType from gso.products.product_types.l3_core_service import L3CoreService, L3CoreServiceType
from gso.utils.shared_enums import APType from gso.utils.shared_enums import APType
from test.workflows import extract_state, run_workflow from test.workflows import extract_state, run_workflow
@pytest.mark.parametrize("l3_core_service_type", [NRENL3CoreServiceType.GEANT_IP, NRENL3CoreServiceType.IAS]) @pytest.mark.parametrize(
"l3_core_service_type",
[
L3CoreServiceType.GEANT_IP,
L3CoreServiceType.IAS,
L3CoreServiceType.GWS,
L3CoreServiceType.LHCONE,
L3CoreServiceType.COPERNICUS,
],
)
@pytest.mark.workflow() @pytest.mark.workflow()
def test_modify_nren_l3_core_service_remove_edge_port_success( def test_modify_l3_core_service_remove_edge_port_success(l3_core_service_subscription_factory, l3_core_service_type):
nren_l3_core_service_subscription_factory, l3_core_service_type subscription_id = l3_core_service_subscription_factory(l3_core_service_type=l3_core_service_type)
): subscription = L3CoreService.from_subscription(subscription_id)
subscription_id = nren_l3_core_service_subscription_factory(nren_l3_core_service_type=l3_core_service_type) access_port = subscription.l3_core_service.ap_list[0]
subscription = NRENL3CoreService.from_subscription(subscription_id)
access_port = subscription.nren_l3_core_service.nren_ap_list[0]
input_form_data = [ input_form_data = [
{"subscription_id": subscription_id}, {"subscription_id": subscription_id},
{ {
...@@ -27,18 +34,27 @@ def test_modify_nren_l3_core_service_remove_edge_port_success( ...@@ -27,18 +34,27 @@ def test_modify_nren_l3_core_service_remove_edge_port_success(
{}, {},
] ]
result, _, _ = run_workflow("modify_nren_l3_core_service", input_form_data) result, _, _ = run_workflow("modify_l3_core_service", input_form_data)
state = extract_state(result) state = extract_state(result)
subscription = NRENL3CoreService.from_subscription(state["subscription_id"]) subscription = L3CoreService.from_subscription(state["subscription_id"])
assert len(subscription.nren_l3_core_service.nren_ap_list) == 1 assert len(subscription.l3_core_service.ap_list) == 1
assert subscription.nren_l3_core_service.nren_ap_list[0].ap_type == APType.LOAD_BALANCED assert subscription.l3_core_service.ap_list[0].ap_type == APType.LOAD_BALANCED
@pytest.mark.parametrize("l3_core_service_type", [NRENL3CoreServiceType.GEANT_IP, NRENL3CoreServiceType.IAS]) @pytest.mark.parametrize(
"l3_core_service_type",
[
L3CoreServiceType.GEANT_IP,
L3CoreServiceType.IAS,
L3CoreServiceType.GWS,
L3CoreServiceType.LHCONE,
L3CoreServiceType.COPERNICUS,
],
)
@pytest.mark.workflow() @pytest.mark.workflow()
def test_modify_nren_l3_core_service_add_new_edge_port_success( def test_modify_l3_core_service_add_new_edge_port_success(
nren_l3_core_service_subscription_factory, l3_core_service_subscription_factory,
edge_port_subscription_factory, edge_port_subscription_factory,
partner_factory, partner_factory,
faker, faker,
...@@ -46,10 +62,8 @@ def test_modify_nren_l3_core_service_add_new_edge_port_success( ...@@ -46,10 +62,8 @@ def test_modify_nren_l3_core_service_add_new_edge_port_success(
): ):
partner = partner_factory() partner = partner_factory()
new_edge_port = edge_port_subscription_factory(partner=partner) new_edge_port = edge_port_subscription_factory(partner=partner)
subscription_id = nren_l3_core_service_subscription_factory( subscription_id = l3_core_service_subscription_factory(partner=partner, l3_core_service_type=l3_core_service_type)
partner=partner, nren_l3_core_service_type=l3_core_service_type subscription = L3CoreService.from_subscription(subscription_id)
)
subscription = NRENL3CoreService.from_subscription(subscription_id)
input_form_data = [ input_form_data = [
{"subscription_id": subscription_id}, {"subscription_id": subscription_id},
{ {
...@@ -58,7 +72,7 @@ def test_modify_nren_l3_core_service_add_new_edge_port_success( ...@@ -58,7 +72,7 @@ def test_modify_nren_l3_core_service_add_new_edge_port_success(
"edge_port": str(port.sbp.edge_port.owner_subscription_id), "edge_port": str(port.sbp.edge_port.owner_subscription_id),
"ap_type": port.ap_type, "ap_type": port.ap_type,
} }
for port in subscription.nren_l3_core_service.nren_ap_list for port in subscription.l3_core_service.ap_list
] ]
+ [ + [
{ {
...@@ -89,11 +103,11 @@ def test_modify_nren_l3_core_service_add_new_edge_port_success( ...@@ -89,11 +103,11 @@ def test_modify_nren_l3_core_service_add_new_edge_port_success(
}, },
] ]
result, _, _ = run_workflow("modify_nren_l3_core_service", input_form_data) result, _, _ = run_workflow("modify_l3_core_service", input_form_data)
state = extract_state(result) state = extract_state(result)
subscription = NRENL3CoreService.from_subscription(state["subscription_id"]) subscription = L3CoreService.from_subscription(state["subscription_id"])
assert len(subscription.nren_l3_core_service.nren_ap_list) == 3 assert len(subscription.l3_core_service.ap_list) == 3
@pytest.fixture() @pytest.fixture()
...@@ -143,13 +157,22 @@ def sbp_input_form_data(faker): ...@@ -143,13 +157,22 @@ def sbp_input_form_data(faker):
return _generate_form_data return _generate_form_data
@pytest.mark.parametrize("l3_core_service_type", [NRENL3CoreServiceType.GEANT_IP, NRENL3CoreServiceType.IAS]) @pytest.mark.parametrize(
"l3_core_service_type",
[
L3CoreServiceType.GEANT_IP,
L3CoreServiceType.IAS,
L3CoreServiceType.GWS,
L3CoreServiceType.LHCONE,
L3CoreServiceType.COPERNICUS,
],
)
@pytest.mark.workflow() @pytest.mark.workflow()
def test_modify_nren_l3_core_service_modify_edge_port_success( def test_modify_l3_core_service_modify_edge_port_success(
faker, nren_l3_core_service_subscription_factory, l3_core_service_type, sbp_input_form_data faker, l3_core_service_subscription_factory, l3_core_service_type, sbp_input_form_data
): ):
subscription_id = nren_l3_core_service_subscription_factory(nren_l3_core_service_type=l3_core_service_type) subscription_id = l3_core_service_subscription_factory(l3_core_service_type=l3_core_service_type)
subscription = NRENL3CoreService.from_subscription(subscription_id) subscription = L3CoreService.from_subscription(subscription_id)
new_sbp_data = [sbp_input_form_data(), sbp_input_form_data()] new_sbp_data = [sbp_input_form_data(), sbp_input_form_data()]
input_form_data = [ input_form_data = [
{"subscription_id": subscription_id}, {"subscription_id": subscription_id},
...@@ -159,134 +182,124 @@ def test_modify_nren_l3_core_service_modify_edge_port_success( ...@@ -159,134 +182,124 @@ def test_modify_nren_l3_core_service_modify_edge_port_success(
"edge_port": str(port.sbp.edge_port.owner_subscription_id), "edge_port": str(port.sbp.edge_port.owner_subscription_id),
"ap_type": port.ap_type, "ap_type": port.ap_type,
} }
for port in subscription.nren_l3_core_service.nren_ap_list for port in subscription.l3_core_service.ap_list
] ]
}, },
{**new_sbp_data[0]}, {**new_sbp_data[0]},
{**new_sbp_data[1]}, {**new_sbp_data[1]},
] ]
result, _, _ = run_workflow("modify_nren_l3_core_service", input_form_data) result, _, _ = run_workflow("modify_l3_core_service", input_form_data)
state = extract_state(result) state = extract_state(result)
subscription = NRENL3CoreService.from_subscription(state["subscription_id"]) subscription = L3CoreService.from_subscription(state["subscription_id"])
assert len(subscription.nren_l3_core_service.nren_ap_list) == 2 assert len(subscription.l3_core_service.ap_list) == 2
for i in range(2): for i in range(2):
assert subscription.nren_l3_core_service.nren_ap_list[i].sbp.geant_sid == new_sbp_data[i]["geant_sid"] assert subscription.l3_core_service.ap_list[i].sbp.geant_sid == new_sbp_data[i]["geant_sid"]
assert subscription.nren_l3_core_service.nren_ap_list[i].sbp.is_tagged == new_sbp_data[i]["is_tagged"] assert subscription.l3_core_service.ap_list[i].sbp.is_tagged == new_sbp_data[i]["is_tagged"]
assert subscription.nren_l3_core_service.nren_ap_list[i].sbp.vlan_id == new_sbp_data[i]["vlan_id"] assert subscription.l3_core_service.ap_list[i].sbp.vlan_id == new_sbp_data[i]["vlan_id"]
assert ( assert str(subscription.l3_core_service.ap_list[i].sbp.ipv4_address) == new_sbp_data[i]["ipv4_address"]
str(subscription.nren_l3_core_service.nren_ap_list[i].sbp.ipv4_address) == new_sbp_data[i]["ipv4_address"] assert str(subscription.l3_core_service.ap_list[i].sbp.ipv6_address) == new_sbp_data[i]["ipv6_address"]
)
assert (
str(subscription.nren_l3_core_service.nren_ap_list[i].sbp.ipv6_address) == new_sbp_data[i]["ipv6_address"]
)
assert ( assert (
subscription.nren_l3_core_service.nren_ap_list[i].sbp.custom_firewall_filters subscription.l3_core_service.ap_list[i].sbp.custom_firewall_filters
== new_sbp_data[i]["custom_firewall_filters"] == new_sbp_data[i]["custom_firewall_filters"]
) )
assert ( assert (
subscription.nren_l3_core_service.nren_ap_list[i].sbp.bgp_session_list[0].bfd_enabled subscription.l3_core_service.ap_list[i].sbp.bgp_session_list[0].bfd_enabled
== new_sbp_data[i]["v4_bgp_peer"]["bfd_enabled"] == new_sbp_data[i]["v4_bgp_peer"]["bfd_enabled"]
) )
assert ( assert (
subscription.nren_l3_core_service.nren_ap_list[i].sbp.bgp_session_list[0].has_custom_policies subscription.l3_core_service.ap_list[i].sbp.bgp_session_list[0].has_custom_policies
== new_sbp_data[i]["v4_bgp_peer"]["has_custom_policies"] == new_sbp_data[i]["v4_bgp_peer"]["has_custom_policies"]
) )
assert ( assert (
subscription.nren_l3_core_service.nren_ap_list[i].sbp.bgp_session_list[0].authentication_key subscription.l3_core_service.ap_list[i].sbp.bgp_session_list[0].authentication_key
== new_sbp_data[i]["v4_bgp_peer"]["authentication_key"] == new_sbp_data[i]["v4_bgp_peer"]["authentication_key"]
) )
assert ( assert (
subscription.nren_l3_core_service.nren_ap_list[i].sbp.bgp_session_list[0].multipath_enabled subscription.l3_core_service.ap_list[i].sbp.bgp_session_list[0].multipath_enabled
== new_sbp_data[i]["v4_bgp_peer"]["multipath_enabled"] == new_sbp_data[i]["v4_bgp_peer"]["multipath_enabled"]
) )
assert ( assert (
subscription.nren_l3_core_service.nren_ap_list[i].sbp.bgp_session_list[0].send_default_route subscription.l3_core_service.ap_list[i].sbp.bgp_session_list[0].send_default_route
== new_sbp_data[i]["v4_bgp_peer"]["send_default_route"] == new_sbp_data[i]["v4_bgp_peer"]["send_default_route"]
) )
assert ( assert (
subscription.nren_l3_core_service.nren_ap_list[i].sbp.bgp_session_list[0].is_passive subscription.l3_core_service.ap_list[i].sbp.bgp_session_list[0].is_passive
== new_sbp_data[i]["v4_bgp_peer"]["is_passive"] == new_sbp_data[i]["v4_bgp_peer"]["is_passive"]
) )
assert ( assert (
str(subscription.nren_l3_core_service.nren_ap_list[i].sbp.bgp_session_list[0].peer_address) str(subscription.l3_core_service.ap_list[i].sbp.bgp_session_list[0].peer_address)
== new_sbp_data[i]["v4_bgp_peer"]["peer_address"] == new_sbp_data[i]["v4_bgp_peer"]["peer_address"]
) )
assert ( assert (
bool( bool(IPFamily.V4MULTICAST in subscription.l3_core_service.ap_list[i].sbp.bgp_session_list[0].families)
IPFamily.V4MULTICAST
in subscription.nren_l3_core_service.nren_ap_list[i].sbp.bgp_session_list[0].families
)
== new_sbp_data[i]["v4_bgp_peer"]["add_v4_multicast"] == new_sbp_data[i]["v4_bgp_peer"]["add_v4_multicast"]
) )
assert ( assert (
subscription.nren_l3_core_service.nren_ap_list[i].sbp.bgp_session_list[1].bfd_enabled subscription.l3_core_service.ap_list[i].sbp.bgp_session_list[1].bfd_enabled
== new_sbp_data[i]["v6_bgp_peer"]["bfd_enabled"] == new_sbp_data[i]["v6_bgp_peer"]["bfd_enabled"]
) )
assert ( assert (
subscription.nren_l3_core_service.nren_ap_list[i].sbp.bgp_session_list[1].has_custom_policies subscription.l3_core_service.ap_list[i].sbp.bgp_session_list[1].has_custom_policies
== new_sbp_data[i]["v6_bgp_peer"]["has_custom_policies"] == new_sbp_data[i]["v6_bgp_peer"]["has_custom_policies"]
) )
assert ( assert (
subscription.nren_l3_core_service.nren_ap_list[i].sbp.bgp_session_list[1].authentication_key subscription.l3_core_service.ap_list[i].sbp.bgp_session_list[1].authentication_key
== new_sbp_data[i]["v6_bgp_peer"]["authentication_key"] == new_sbp_data[i]["v6_bgp_peer"]["authentication_key"]
) )
assert ( assert (
subscription.nren_l3_core_service.nren_ap_list[i].sbp.bgp_session_list[1].multipath_enabled subscription.l3_core_service.ap_list[i].sbp.bgp_session_list[1].multipath_enabled
== new_sbp_data[i]["v6_bgp_peer"]["multipath_enabled"] == new_sbp_data[i]["v6_bgp_peer"]["multipath_enabled"]
) )
assert ( assert (
subscription.nren_l3_core_service.nren_ap_list[i].sbp.bgp_session_list[1].send_default_route subscription.l3_core_service.ap_list[i].sbp.bgp_session_list[1].send_default_route
== new_sbp_data[i]["v6_bgp_peer"]["send_default_route"] == new_sbp_data[i]["v6_bgp_peer"]["send_default_route"]
) )
assert ( assert (
subscription.nren_l3_core_service.nren_ap_list[i].sbp.bgp_session_list[1].is_passive subscription.l3_core_service.ap_list[i].sbp.bgp_session_list[1].is_passive
== new_sbp_data[i]["v6_bgp_peer"]["is_passive"] == new_sbp_data[i]["v6_bgp_peer"]["is_passive"]
) )
assert ( assert (
str(subscription.nren_l3_core_service.nren_ap_list[i].sbp.bgp_session_list[1].peer_address) str(subscription.l3_core_service.ap_list[i].sbp.bgp_session_list[1].peer_address)
== new_sbp_data[i]["v6_bgp_peer"]["peer_address"] == new_sbp_data[i]["v6_bgp_peer"]["peer_address"]
) )
assert ( assert (
bool( bool(IPFamily.V6MULTICAST in subscription.l3_core_service.ap_list[i].sbp.bgp_session_list[1].families)
IPFamily.V6MULTICAST
in subscription.nren_l3_core_service.nren_ap_list[i].sbp.bgp_session_list[1].families
)
== new_sbp_data[i]["v6_bgp_peer"]["add_v6_multicast"] == new_sbp_data[i]["v6_bgp_peer"]["add_v6_multicast"]
) )
assert ( assert (
subscription.nren_l3_core_service.nren_ap_list[i].sbp.v4_bfd_settings.bfd_enabled subscription.l3_core_service.ap_list[i].sbp.v4_bfd_settings.bfd_enabled
== new_sbp_data[i]["v4_bfd_settings"]["bfd_enabled"] == new_sbp_data[i]["v4_bfd_settings"]["bfd_enabled"]
) )
assert ( assert (
subscription.nren_l3_core_service.nren_ap_list[i].sbp.v4_bfd_settings.bfd_interval_rx subscription.l3_core_service.ap_list[i].sbp.v4_bfd_settings.bfd_interval_rx
== new_sbp_data[i]["v4_bfd_settings"]["bfd_interval_rx"] == new_sbp_data[i]["v4_bfd_settings"]["bfd_interval_rx"]
) )
assert ( assert (
subscription.nren_l3_core_service.nren_ap_list[i].sbp.v4_bfd_settings.bfd_interval_tx subscription.l3_core_service.ap_list[i].sbp.v4_bfd_settings.bfd_interval_tx
== new_sbp_data[i]["v4_bfd_settings"]["bfd_interval_tx"] == new_sbp_data[i]["v4_bfd_settings"]["bfd_interval_tx"]
) )
assert ( assert (
subscription.nren_l3_core_service.nren_ap_list[i].sbp.v4_bfd_settings.bfd_multiplier subscription.l3_core_service.ap_list[i].sbp.v4_bfd_settings.bfd_multiplier
== new_sbp_data[i]["v4_bfd_settings"]["bfd_multiplier"] == new_sbp_data[i]["v4_bfd_settings"]["bfd_multiplier"]
) )
assert ( assert (
subscription.nren_l3_core_service.nren_ap_list[i].sbp.v6_bfd_settings.bfd_enabled subscription.l3_core_service.ap_list[i].sbp.v6_bfd_settings.bfd_enabled
== new_sbp_data[i]["v6_bfd_settings"]["bfd_enabled"] == new_sbp_data[i]["v6_bfd_settings"]["bfd_enabled"]
) )
assert ( assert (
subscription.nren_l3_core_service.nren_ap_list[i].sbp.v6_bfd_settings.bfd_interval_rx subscription.l3_core_service.ap_list[i].sbp.v6_bfd_settings.bfd_interval_rx
== new_sbp_data[i]["v6_bfd_settings"]["bfd_interval_rx"] == new_sbp_data[i]["v6_bfd_settings"]["bfd_interval_rx"]
) )
assert ( assert (
subscription.nren_l3_core_service.nren_ap_list[i].sbp.v6_bfd_settings.bfd_interval_tx subscription.l3_core_service.ap_list[i].sbp.v6_bfd_settings.bfd_interval_tx
== new_sbp_data[i]["v6_bfd_settings"]["bfd_interval_tx"] == new_sbp_data[i]["v6_bfd_settings"]["bfd_interval_tx"]
) )
assert ( assert (
subscription.nren_l3_core_service.nren_ap_list[i].sbp.v6_bfd_settings.bfd_multiplier subscription.l3_core_service.ap_list[i].sbp.v6_bfd_settings.bfd_multiplier
== new_sbp_data[i]["v6_bfd_settings"]["bfd_multiplier"] == new_sbp_data[i]["v6_bfd_settings"]["bfd_multiplier"]
) )
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment