diff --git a/src/device/service/drivers/ietf_l3vpn/Constants.py b/src/device/service/drivers/ietf_l3vpn/Constants.py
new file mode 100644
index 0000000000000000000000000000000000000000..df66eb16b3d78c1b388a086011ed6f6b75b8099f
--- /dev/null
+++ b/src/device/service/drivers/ietf_l3vpn/Constants.py
@@ -0,0 +1,25 @@
+# Copyright 2022-2024 ETSI OSG/SDG TeraFlowSDN (TFS) (https://tfs.etsi.org/)
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+from device.service.driver_api._Driver import (
+ RESOURCE_ENDPOINTS,
+ RESOURCE_INTERFACES,
+ RESOURCE_NETWORK_INSTANCES,
+)
+
+SPECIAL_RESOURCE_MAPPINGS = {
+ RESOURCE_ENDPOINTS: "/endpoints",
+ RESOURCE_INTERFACES: "/interfaces",
+ RESOURCE_NETWORK_INSTANCES: "/net-instances",
+}
diff --git a/src/device/service/drivers/ietf_l3vpn/TfsApiClient.py b/src/device/service/drivers/ietf_l3vpn/TfsApiClient.py
index 86ee2b1bed13c09f2a024f5bb8790590f09e1d8a..661907a7384cc4abb6aafc106f01216ee5b0652c 100644
--- a/src/device/service/drivers/ietf_l3vpn/TfsApiClient.py
+++ b/src/device/service/drivers/ietf_l3vpn/TfsApiClient.py
@@ -69,6 +69,7 @@ class TfsApiClient:
) -> None:
self._devices_url = GET_DEVICES_URL.format(scheme, address, port)
self._links_url = GET_LINKS_URL.format(scheme, address, port)
+ self._l3vph_url = L3VPN_URL.format(scheme, address, port)
self._auth = (
HTTPBasicAuth(username, password)
if username is not None and password is not None
@@ -162,12 +163,12 @@ class TfsApiClient:
def create_connectivity_service(self, l3vpn_data: dict) -> None:
try:
- requests.post(L3VPN_URL, json=l3vpn_data, auth=self._auth)
+ requests.post(self._l3vph_url, json=l3vpn_data, auth=self._auth)
except requests.exceptions.ConnectionError:
raise Exception("faild to send post request to TFS L3VPN NBI")
def delete_connectivity_service(self, service_uuid: str) -> None:
- url = L3VPN_URL + f"/vpn-service={service_uuid}"
+ url = self._l3vph_url + f"/vpn-service={service_uuid}"
try:
requests.delete(url, auth=self._auth)
except requests.exceptions.ConnectionError:
diff --git a/src/device/service/drivers/ietf_l3vpn/Tools.py b/src/device/service/drivers/ietf_l3vpn/Tools.py
index 8930cde00ad08f6d05126c39759bc04b62dd25a8..bf9c40732d00f5a7e37a37ca8c56d565ca8e29e9 100644
--- a/src/device/service/drivers/ietf_l3vpn/Tools.py
+++ b/src/device/service/drivers/ietf_l3vpn/Tools.py
@@ -12,10 +12,16 @@
# See the License for the specific language governing permissions and
# limitations under the License.
import logging
-from typing import TypedDict
+from typing import Any, Dict, Optional, Tuple, TypedDict
import requests
+from common.proto.kpi_sample_types_pb2 import KpiSampleType
+from common.type_checkers.Checkers import chk_attribute, chk_string, chk_type
+from device.service.driver_api._Driver import RESOURCE_ENDPOINTS
+
+from .Constants import SPECIAL_RESOURCE_MAPPINGS
+
class LANPrefixesDict(TypedDict):
lan: str
@@ -79,54 +85,66 @@ def create_l3vpn_datamodel(service_uuid, resource_value: dict) -> dict:
)
src_site_id: str = resource_value.get("src_site_id", f"site_{src_site_location}")
src_management_type: str = resource_value.get(
- "src_management_type", "ietf-l3vpn-svc:provider-managed"
+ "src_management_type", "ietf-l3vpn-svc:customer-managed"
)
- if src_management_type != "ietf-l3vpn-svc:provider-managed":
+ if src_management_type != "ietf-l3vpn-svc:customer-managed":
raise Exception("management type %s not supported", src_management_type)
src_role: str = "ietf-l3vpn-svc:hub-role"
src_ce_address: str = resource_value["src_ce_address"]
src_pe_address: str = resource_value["src_pe_address"]
src_ce_pe_network_prefix: int = resource_value["src_ce_pe_network_prefix"]
src_mtu: int = resource_value["src_mtu"]
- src_input_bw: int = resource_value["src_input_bw"]
- src_output_bw: int = resource_value["src_output_bw"]
- src_qos_profile_id: str = resource_value["src_qos_profile_id"]
- src_qos_profile_direction: str = resource_value["src_qos_profile_direction"]
- src_qos_profile_latency: int = resource_value["src_qos_profile_latency"]
- src_qos_profile_bw_guarantee: int = resource_value["src_qos_profile_bw_guarantee"]
+ src_input_bw: int = resource_value.get("src_input_bw", 1000000000)
+ src_output_bw: int = resource_value.get("src_input_bw", 1000000000)
+ src_qos_profile_id: str = resource_value.get(
+ "src_qos_profile_id", "src_qos_profile"
+ )
+ src_qos_profile_direction: str = (
+ resource_value.get("src_qos_profile_direction", "ietf-l3vpn-svc:both"),
+ )
+ src_qos_profile_latency: int = resource_value.get("src_qos_profile_latency", 10)
+ src_qos_profile_bw_guarantee: int = resource_value.get(
+ "src_qos_profile_bw_guarantee", 100
+ )
dst_device_uuid = resource_value["dst_device_name"]
dst_endpoint_uuid = resource_value["dst_endpoint_name"]
dst_site_location: str = resource_value["dst_site_location"]
- dst_ipv4_lan_prefixes: list[LANPrefixesDict] = resource_value.get(
+ dst_ipv4_lan_prefixes: list[LANPrefixesDict] = resource_value[
"dst_ipv4_lan_prefixes"
- )
+ ]
dst_site_id: str = resource_value.get("dst_site_id", f"site_{dst_site_location}")
dst_management_type: str = resource_value.get(
- "dst_management_type", "ietf-l3vpn-svc:provider-managed"
+ "dst_management_type", "ietf-l3vpn-svc:customer-managed"
)
- if dst_management_type != "ietf-l3vpn-svc:provider-managed":
+ if dst_management_type != "ietf-l3vpn-svc:customer-managed":
raise Exception("management type %s not supported", dst_management_type)
dst_role: str = "ietf-l3vpn-svc:spoke-role"
dst_ce_address: str = resource_value["dst_ce_address"]
dst_pe_address: str = resource_value["dst_pe_address"]
dst_ce_pe_network_prefix: int = resource_value["dst_ce_pe_network_prefix"]
dst_mtu: int = resource_value["dst_mtu"]
- dst_input_bw: int = resource_value["dst_input_bw"]
- dst_output_bw: int = resource_value["dst_output_bw"]
- dst_qos_profile_id: str = resource_value["dst_qos_profile_id"]
- dst_qos_profile_direction: str = resource_value["dst_qos_profile_direction"]
- dst_qos_profile_latency: int = resource_value["dst_qos_profile_latency"]
- dst_qos_profile_bw_guarantee: int = resource_value["dst_qos_profile_bw_guarantee"]
+ dst_input_bw: int = resource_value.get("dst_input_bw", 1000000000)
+ dst_output_bw: int = resource_value.get("dst_output_bw", 1000000000)
+ dst_qos_profile_id: str = resource_value.get(
+ "dst_qos_profile_id", "dst_qos_profile"
+ )
+ dst_qos_profile_direction: str = (
+ resource_value.get("dst_qos_profile_direction", "ietf-l3vpn-svc:both"),
+ )
+ dst_qos_profile_latency: int = resource_value.get("dst_qos_profile_latency", 10)
+ dst_qos_profile_bw_guarantee: int = resource_value.get(
+ "dst_qos_profile_bw_guarantee", 100
+ )
# Create source site information
- src_management = {"type": "ietf-l3vpn-svc:provider-managed"}
+ src_management = {"type": src_management_type}
src_locations = {"location": [{"location-id": src_site_location}]}
src_devices = {
"device": [{"device-id": src_device_uuid, "location": src_site_location}]
}
src_site_lan_prefixes = [
- {"lan": lp["lan"], "lan-tag": lp["lan_tag"], "next-hop": src_ce_address}
+ {"lan": lp["lan"], "lan-tag": lp["lan_tag"], "next-hop": src_pe_address}
for lp in src_ipv4_lan_prefixes
]
src_site_routing_protocols = {
@@ -186,13 +204,13 @@ def create_l3vpn_datamodel(service_uuid, resource_value: dict) -> dict:
}
# Create destination site information
- dst_management = {"type": "ietf-l3vpn-svc:provider-managed"}
+ dst_management = {"type": src_management_type}
dst_locations = {"location": [{"location-id": dst_site_location}]}
dst_devices = {
"device": [{"device-id": dst_device_uuid, "location": dst_site_location}]
}
dst_site_lan_prefixes = [
- {"lan": lp["lan"], "lan-tag": lp["lan_tag"], "next-hop": dst_ce_address}
+ {"lan": lp["lan"], "lan-tag": lp["lan_tag"], "next-hop": dst_pe_address}
for lp in dst_ipv4_lan_prefixes
]
dst_site_routing_protocols = {
@@ -262,14 +280,12 @@ def create_l3vpn_datamodel(service_uuid, resource_value: dict) -> dict:
"site-network-accesses": src_site_network_accesses,
},
{
- {
- "site-id": dst_site_id,
- "management": dst_management,
- "locations": dst_locations,
- "devices": dst_devices,
- "routing-protocols": dst_site_routing_protocols,
- "site-network-accesses": dst_site_network_accesses,
- },
+ "site-id": dst_site_id,
+ "management": dst_management,
+ "locations": dst_locations,
+ "devices": dst_devices,
+ "routing-protocols": dst_site_routing_protocols,
+ "site-network-accesses": dst_site_network_accesses,
},
]
}
@@ -282,3 +298,111 @@ def create_l3vpn_datamodel(service_uuid, resource_value: dict) -> dict:
}
return l3_vpn_data_model
+
+
+def process_optional_string_field(
+ endpoint_data: Dict[str, Any],
+ field_name: str,
+ endpoint_resource_value: Dict[str, Any],
+) -> None:
+ field_value = chk_attribute(
+ field_name, endpoint_data, "endpoint_data", default=None
+ )
+ if field_value is None:
+ return
+ chk_string("endpoint_data.{:s}".format(field_name), field_value)
+ if len(field_value) > 0:
+ endpoint_resource_value[field_name] = field_value
+
+
+def compose_resource_endpoint(
+ endpoint_data: Dict[str, Any],
+) -> Optional[Tuple[str, Dict]]:
+ try:
+ # Check type of endpoint_data
+ chk_type("endpoint_data", endpoint_data, dict)
+
+ # Check endpoint UUID (mandatory)
+ endpoint_uuid = chk_attribute("uuid", endpoint_data, "endpoint_data")
+ chk_string("endpoint_data.uuid", endpoint_uuid, min_length=1)
+ endpoint_resource_path = SPECIAL_RESOURCE_MAPPINGS.get(RESOURCE_ENDPOINTS)
+ endpoint_resource_key = "{:s}/endpoint[{:s}]".format(
+ endpoint_resource_path, endpoint_uuid
+ )
+ endpoint_resource_value = {"uuid": endpoint_uuid}
+
+ # Check endpoint optional string fields
+ process_optional_string_field(endpoint_data, "name", endpoint_resource_value)
+ process_optional_string_field(endpoint_data, "type", endpoint_resource_value)
+ process_optional_string_field(
+ endpoint_data, "context_uuid", endpoint_resource_value
+ )
+ process_optional_string_field(
+ endpoint_data, "topology_uuid", endpoint_resource_value
+ )
+
+ # Check endpoint sample types (optional)
+ endpoint_sample_types = chk_attribute(
+ "sample_types", endpoint_data, "endpoint_data", default=[]
+ )
+ chk_type("endpoint_data.sample_types", endpoint_sample_types, list)
+ sample_types = {}
+ sample_type_errors = []
+ for i, endpoint_sample_type in enumerate(endpoint_sample_types):
+ field_name = "endpoint_data.sample_types[{:d}]".format(i)
+ try:
+ chk_type(field_name, endpoint_sample_type, (int, str))
+ if isinstance(endpoint_sample_type, int):
+ metric_name = KpiSampleType.Name(endpoint_sample_type)
+ metric_id = endpoint_sample_type
+ elif isinstance(endpoint_sample_type, str):
+ metric_id = KpiSampleType.Value(endpoint_sample_type)
+ metric_name = endpoint_sample_type
+ else:
+ str_type = str(type(endpoint_sample_type))
+ raise Exception("Bad format: {:s}".format(str_type)) # pylint: disable=broad-exception-raised
+ except Exception as e: # pylint: disable=broad-exception-caught
+ MSG = "Unsupported {:s}({:s}) : {:s}"
+ sample_type_errors.append(
+ MSG.format(field_name, str(endpoint_sample_type), str(e))
+ )
+
+ metric_name = metric_name.lower().replace("kpisampletype_", "")
+ monitoring_resource_key = "{:s}/state/{:s}".format(
+ endpoint_resource_key, metric_name
+ )
+ sample_types[metric_id] = monitoring_resource_key
+
+ if len(sample_type_errors) > 0:
+ # pylint: disable=broad-exception-raised
+ raise Exception(
+ "Malformed Sample Types:\n{:s}".format("\n".join(sample_type_errors))
+ )
+
+ if len(sample_types) > 0:
+ endpoint_resource_value["sample_types"] = sample_types
+
+ if "location" in endpoint_data:
+ endpoint_resource_value["location"] = endpoint_data["location"]
+
+ if "ce-ip" in endpoint_data:
+ endpoint_resource_value["ce-ip"] = endpoint_data["ce-ip"]
+
+ if "address_ip" in endpoint_data:
+ endpoint_resource_value["address_ip"] = endpoint_data["address_ip"]
+
+ if "address_prefix" in endpoint_data:
+ endpoint_resource_value["address_prefix"] = endpoint_data["address_prefix"]
+
+ if "mtu" in endpoint_data:
+ endpoint_resource_value["mtu"] = endpoint_data["mtu"]
+
+ if "ipv4_lan_prefixes" in endpoint_data:
+ endpoint_resource_value["ipv4_lan_prefixes"] = endpoint_data[
+ "ipv4_lan_prefixes"
+ ]
+
+ return endpoint_resource_key, endpoint_resource_value
+ except: # pylint: disable=bare-except
+ LOGGER.exception("Problem composing endpoint({:s})".format(str(endpoint_data)))
+ return None
diff --git a/src/device/service/drivers/ietf_l3vpn/driver.py b/src/device/service/drivers/ietf_l3vpn/driver.py
index 3d3fa2299cb7517912206fae6476aa2ef448afb2..e81736d43875e4344efc804e2709103d867cc6cd 100644
--- a/src/device/service/drivers/ietf_l3vpn/driver.py
+++ b/src/device/service/drivers/ietf_l3vpn/driver.py
@@ -14,29 +14,37 @@
import json
import logging
+import re
import threading
from typing import Any, Iterator, List, Optional, Tuple, Union
+import anytree
import requests
from requests.auth import HTTPBasicAuth
from common.method_wrappers.Decorator import MetricsPool, metered_subclass_method
-from common.type_checkers.Checkers import chk_string, chk_type
+from common.type_checkers.Checkers import chk_length, chk_string, chk_type
from device.service.driver_api._Driver import (
RESOURCE_ENDPOINTS,
RESOURCE_SERVICES,
_Driver,
)
+from device.service.driver_api.AnyTreeTools import (
+ TreeNode,
+ dump_subtree,
+ get_subnode,
+ set_subnode_value,
+)
from device.service.driver_api.ImportTopologyEnum import (
ImportTopologyEnum,
get_import_topology,
)
+from .Constants import SPECIAL_RESOURCE_MAPPINGS
from .TfsApiClient import TfsApiClient
from .Tools import (
+ compose_resource_endpoint,
create_l3vpn_datamodel,
- get_all_active_connectivity_services,
- get_connectivity_service,
service_exists,
)
@@ -48,6 +56,8 @@ ALL_RESOURCE_KEYS = [
RESOURCE_SERVICES,
]
+RE_GET_ENDPOINT_FROM_INTERFACE = re.compile(r"^\/interface\[([^\]]+)\].*")
+
DRIVER_NAME = "ietf_l3vpn"
METRICS_POOL = MetricsPool("Device", "Driver", labels={"driver": DRIVER_NAME})
@@ -58,12 +68,13 @@ class IetfL3VpnDriver(_Driver):
self.__lock = threading.Lock()
self.__started = threading.Event()
self.__terminate = threading.Event()
+ self.__running = TreeNode(".")
scheme = self.settings.get("scheme", "http")
username = self.settings.get("username")
password = self.settings.get("password")
self.tac = TfsApiClient(
self.address,
- int(self.port),
+ self.port,
scheme=scheme,
username=username,
password=password,
@@ -80,6 +91,52 @@ class IetfL3VpnDriver(_Driver):
self.__import_topology = get_import_topology(
self.settings, default=ImportTopologyEnum.DEVICES
)
+ endpoints = self.settings.get("endpoints", [])
+ endpoint_resources = []
+ for endpoint in endpoints:
+ endpoint_resource = compose_resource_endpoint(endpoint)
+ if endpoint_resource is None:
+ continue
+ endpoint_resources.append(endpoint_resource)
+ self._set_initial_config(endpoint_resources)
+
+ def _set_initial_config(
+ self, resources: List[Tuple[str, Any]]
+ ) -> List[Union[bool, Exception]]:
+ chk_type("resources", resources, list)
+ if len(resources) == 0:
+ return []
+ results = []
+ resolver = anytree.Resolver(pathattr="name")
+ with self.__lock:
+ for i, resource in enumerate(resources):
+ str_resource_name = "resources[#{:d}]".format(i)
+ try:
+ chk_type(str_resource_name, resource, (list, tuple))
+ chk_length(str_resource_name, resource, min_length=2, max_length=2)
+ resource_key, resource_value = resource
+ chk_string(str_resource_name, resource_key, allow_empty=False)
+ resource_path = resource_key.split("/")
+ except Exception as e: # pylint: disable=broad-except
+ LOGGER.exception(
+ "Exception validating {:s}: {:s}".format(
+ str_resource_name, str(resource_key)
+ )
+ )
+ results.append(e) # if validation fails, store the exception
+ continue
+
+ try:
+ resource_value = json.loads(resource_value)
+ except: # pylint: disable=bare-except
+ pass
+
+ set_subnode_value(
+ resolver, self.__running, resource_path, resource_value
+ )
+
+ results.append(True)
+ return results
def Connect(self) -> bool:
url = (
@@ -121,38 +178,38 @@ class IetfL3VpnDriver(_Driver):
self, resource_keys: List[str] = []
) -> List[Tuple[str, Union[Any, None, Exception]]]:
chk_type("resources", resource_keys, list)
- results = []
with self.__lock:
if len(resource_keys) == 0:
- resource_keys = ALL_RESOURCE_KEYS
+ return dump_subtree(self.__running)
+ results = []
+ resolver = anytree.Resolver(pathattr="name")
for i, resource_key in enumerate(resource_keys):
str_resource_name = "resource_key[#{:d}]".format(i)
try:
chk_string(str_resource_name, resource_key, allow_empty=False)
- if resource_key == RESOURCE_ENDPOINTS:
- # return endpoints through TFS NBI API and list-devices method
- results.extend(
- self.tac.get_devices_endpoints(self.__import_topology)
- )
- elif resource_key == RESOURCE_SERVICES:
- # return all services through
- reply = get_all_active_connectivity_services(
- wim_url=self.__tfs_nbi_root, auth=self.__auth
- )
- results.extend(reply.json())
- else:
- # assume single-service retrieval
- reply = get_connectivity_service(
- self.__tfs_nbi_root, resource_key
- )
- results.append(reply.json())
+ resource_key = SPECIAL_RESOURCE_MAPPINGS.get(
+ resource_key, resource_key
+ )
+ resource_path = resource_key.split("/")
except Exception as e: # pylint: disable=broad-except
LOGGER.exception(
- "Unhandled error processing resource_key({:s})".format(
- str(resource_key)
+ "Exception validating {:s}: {:s}".format(
+ str_resource_name, str(resource_key)
)
)
- results.append((resource_key, e))
+ results.append(
+ (resource_key, e)
+ ) # if validation fails, store the exception
+ continue
+
+ resource_node = get_subnode(
+ resolver, self.__running, resource_path, default=None
+ )
+ # if not found, resource_node is None
+ if resource_node is None:
+ continue
+ results.extend(dump_subtree(resource_node))
+ return results
return results
@metered_subclass_method(METRICS_POOL)
@@ -169,7 +226,7 @@ class IetfL3VpnDriver(_Driver):
try:
resource_value = json.loads(resource_value)
- service_uuid = resource_value["uuid"] #! fix based on resources
+ service_uuid = resource_value["uuid"]
if service_exists(self.__tfs_nbi_root, self.__auth, service_uuid):
exc = NotImplementedError(
diff --git a/src/device/tests/test_unitary_ietf_l3vpn.py b/src/device/tests/test_unitary_ietf_l3vpn.py
new file mode 100644
index 0000000000000000000000000000000000000000..8683c345451a9bf1504fae1799b4160186c47c6a
--- /dev/null
+++ b/src/device/tests/test_unitary_ietf_l3vpn.py
@@ -0,0 +1,345 @@
+import json
+from json import dumps
+
+import requests
+
+from device.service.drivers.ietf_l3vpn.driver import IetfL3VpnDriver
+from device.service.Tools import RESOURCE_ENDPOINTS
+
+settings = {
+ "endpoints": [
+ {
+ "uuid": "access-pe",
+ "name": "access-pe",
+ "type": "copper",
+ "ce-ip": "1.1.1.1",
+ "address_ip": "3.3.2.1",
+ "address_prefix": 24,
+ "location": "access",
+ "mtu": 1500,
+ "ipv4_lan_prefixes": [
+ {"lan": "128.32.10.0/24", "lan_tag": 10},
+ {"lan": "128.32.20.0/24", "lan_tag": 20},
+ ],
+ },
+ {
+ "uuid": "cloud-pe",
+ "name": "cloud-pe",
+ "type": "copper",
+ "ce-ip": "1.1.1.1",
+ "address_ip": "3.3.2.1",
+ "address_prefix": 24,
+ "location": "cloud",
+ "mtu": 1500,
+ "ipv4_lan_prefixes": [{"lan": "172.1.101.0/24", "lan_tag": 101}],
+ },
+ ],
+ "scheme": "http",
+ "username": "admin",
+ "password": "admin",
+ "base_url": "/restconf/v2/data",
+ "timeout": 120,
+ "verify": False,
+}
+
+post_request_data = []
+get_request_data = []
+
+
+def mock_post(*args, **kwargs):
+ post_request_data.append((args, kwargs))
+
+
+def mock_get(*args, **kwargs):
+ get_request_data.append((args, kwargs))
+
+
+driver = IetfL3VpnDriver(address="1.2.3.4", port=0, **settings)
+
+
+def test_connect(monkeypatch):
+ global post_request_data
+ global get_request_data
+ post_request_data = []
+ get_request_data = []
+ monkeypatch.setattr(requests, "post", mock_post)
+ monkeypatch.setattr(requests, "get", mock_get)
+
+ driver.Connect()
+ assert not post_request_data
+ assert len(get_request_data) == 1
+ assert get_request_data[0][0] == (
+ "http://1.2.3.4:0/restconf/data/ietf-l3vpn-svc:l3vpn-svc/vpn-services",
+ )
+ assert list(get_request_data[0][1].keys()) == ["timeout", "verify", "auth"]
+
+
+def test_GetConfig(monkeypatch):
+ global post_request_data
+ global get_request_data
+ post_request_data = []
+ get_request_data = []
+ monkeypatch.setattr(requests, "post", mock_post)
+ monkeypatch.setattr(requests, "get", mock_get)
+
+ resources_to_get = [RESOURCE_ENDPOINTS]
+ result_GetConfig = driver.GetConfig(resources_to_get)
+ assert result_GetConfig == [
+ (
+ "/endpoints/endpoint[access-pe]",
+ {
+ "uuid": "access-pe",
+ "name": "access-pe",
+ "type": "copper",
+ "location": "access",
+ "ce-ip": "1.1.1.1",
+ "address_ip": "3.3.2.1",
+ "address_prefix": 24,
+ "mtu": 1500,
+ "ipv4_lan_prefixes": [
+ {"lan": "128.32.10.0/24", "lan_tag": 10},
+ {"lan": "128.32.20.0/24", "lan_tag": 20},
+ ],
+ },
+ ),
+ (
+ "/endpoints/endpoint[cloud-pe]",
+ {
+ "uuid": "cloud-pe",
+ "name": "cloud-pe",
+ "type": "copper",
+ "location": "cloud",
+ "ce-ip": "1.1.1.1",
+ "address_ip": "3.3.2.1",
+ "address_prefix": 24,
+ "mtu": 1500,
+ "ipv4_lan_prefixes": [{"lan": "172.1.101.0/24", "lan_tag": 101}],
+ },
+ ),
+ ]
+
+
+def test_SetConfig(monkeypatch):
+ global post_request_data
+ global get_request_data
+ post_request_data = []
+ get_request_data = []
+ monkeypatch.setattr(requests, "post", mock_post)
+ monkeypatch.setattr(requests, "get", mock_get)
+
+ resources = [
+ (
+ "/services/service[vpn_A]",
+ json.dumps(
+ {
+ "uuid": "vpn_A",
+ "src_device_name": "ip-net-controller",
+ "src_endpoint_name": settings["endpoints"][0]["name"],
+ "src_site_location": settings["endpoints"][0]["location"],
+ "src_ipv4_lan_prefixes": settings["endpoints"][0][
+ "ipv4_lan_prefixes"
+ ],
+ "src_ce_address": settings["endpoints"][0]["ce-ip"],
+ "src_pe_address": settings["endpoints"][0]["address_ip"],
+ "src_ce_pe_network_prefix": settings["endpoints"][0][
+ "address_prefix"
+ ],
+ "src_mtu": settings["endpoints"][0]["mtu"],
+ "dst_device_name": "ip-net-controller",
+ "dst_endpoint_name": settings["endpoints"][1]["name"],
+ "dst_site_location": settings["endpoints"][1]["location"],
+ "dst_ipv4_lan_prefixes": settings["endpoints"][1][
+ "ipv4_lan_prefixes"
+ ],
+ "dst_ce_address": settings["endpoints"][1]["ce-ip"],
+ "dst_pe_address": settings["endpoints"][1]["address_ip"],
+ "dst_ce_pe_network_prefix": settings["endpoints"][1][
+ "address_prefix"
+ ],
+ "dst_mtu": settings["endpoints"][1]["mtu"],
+ }
+ ),
+ )
+ ]
+ result_SetConfig = driver.SetConfig(resources)
+ assert result_SetConfig == [("/services/service[vpn_A]", True)]
+ assert len(get_request_data) == 1
+ assert get_request_data[0][0] == (
+ "http://1.2.3.4:0/restconf/data/ietf-l3vpn-svc:l3vpn-svc/vpn-service=vpn_A",
+ )
+ assert len(post_request_data) == 1
+ assert post_request_data[0][0] == (
+ "http://1.2.3.4:0/restconf/data/ietf-l3vpn-svc:l3vpn-svc/vpn-services",
+ )
+ assert post_request_data[0][1]["json"] == {
+ "ietf-l3vpn-svc:l3vpn-svc": {
+ "vpn-services": {"vpn-service": [{"vpn-id": "vpn_A"}]},
+ "sites": {
+ "site": [
+ {
+ "site-id": "site_access",
+ "management": {"type": "ietf-l3vpn-svc:customer-managed"},
+ "locations": {"location": [{"location-id": "access"}]},
+ "devices": {
+ "device": [
+ {
+ "device-id": "ip-net-controller",
+ "location": "access",
+ }
+ ]
+ },
+ "routing-protocols": {
+ "routing-protocol": [
+ {
+ "type": "ietf-l3vpn-svc:static",
+ "static": {
+ "cascaded-lan-prefixes": {
+ "ipv4-lan-prefixes": [
+ {
+ "lan": "128.32.10.0/24",
+ "lan-tag": 10,
+ "next-hop": "3.3.2.1",
+ },
+ {
+ "lan": "128.32.20.0/24",
+ "lan-tag": 20,
+ "next-hop": "3.3.2.1",
+ },
+ ]
+ }
+ },
+ }
+ ]
+ },
+ "site-network-accesses": {
+ "site-network-access": [
+ {
+ "site-network-access-id": "access-pe",
+ "site-network-access-type": "ietf-l3vpn-svc:multipoint",
+ "device-reference": "ip-net-controller",
+ "vpn-attachment": {
+ "vpn-id": "vpn_A",
+ "site-role": "ietf-l3vpn-svc:hub-role",
+ },
+ "ip-connection": {
+ "ipv4": {
+ "address-allocation-type": "ietf-l3vpn-svc:static-address",
+ "addresses": {
+ "provider-address": "3.3.2.1",
+ "customer-address": "1.1.1.1",
+ "prefix-length": 24,
+ },
+ }
+ },
+ "service": {
+ "svc-mtu": 1500,
+ "svc-input-bandwidth": 1000000000,
+ "svc-output-bandwidth": 1000000000,
+ "qos": {
+ "qos-profile": {
+ "classes": {
+ "class": [
+ {
+ "class-id": "src_qos_profile",
+ "direction": (
+ "ietf-l3vpn-svc:both",
+ ),
+ "latency": {
+ "latency-boundary": 10
+ },
+ "bandwidth": {
+ "guaranteed-bw-percent": 100
+ },
+ }
+ ]
+ }
+ }
+ },
+ },
+ }
+ ]
+ },
+ },
+ {
+ "site-id": "site_cloud",
+ "management": {"type": "ietf-l3vpn-svc:customer-managed"},
+ "locations": {"location": [{"location-id": "cloud"}]},
+ "devices": {
+ "device": [
+ {
+ "device-id": "ip-net-controller",
+ "location": "cloud",
+ }
+ ]
+ },
+ "routing-protocols": {
+ "routing-protocol": [
+ {
+ "type": "ietf-l3vpn-svc:static",
+ "static": {
+ "cascaded-lan-prefixes": {
+ "ipv4-lan-prefixes": [
+ {
+ "lan": "172.1.101.0/24",
+ "lan-tag": 101,
+ "next-hop": "3.3.2.1",
+ }
+ ]
+ }
+ },
+ }
+ ]
+ },
+ "site-network-accesses": {
+ "site-network-access": [
+ {
+ "site-network-access-id": "cloud-pe",
+ "site-network-access-type": "ietf-l3vpn-svc:multipoint",
+ "device-reference": "ip-net-controller",
+ "vpn-attachment": {
+ "vpn-id": "vpn_A",
+ "site-role": "ietf-l3vpn-svc:spoke-role",
+ },
+ "ip-connection": {
+ "ipv4": {
+ "address-allocation-type": "ietf-l3vpn-svc:static-address",
+ "addresses": {
+ "provider-address": "3.3.2.1",
+ "customer-address": "1.1.1.1",
+ "prefix-length": 24,
+ },
+ }
+ },
+ "service": {
+ "svc-mtu": 1500,
+ "svc-input-bandwidth": 1000000000,
+ "svc-output-bandwidth": 1000000000,
+ "qos": {
+ "qos-profile": {
+ "classes": {
+ "class": [
+ {
+ "class-id": "dst_qos_profile",
+ "direction": (
+ "ietf-l3vpn-svc:both",
+ ),
+ "latency": {
+ "latency-boundary": 10
+ },
+ "bandwidth": {
+ "guaranteed-bw-percent": 100
+ },
+ }
+ ]
+ }
+ }
+ },
+ },
+ }
+ ]
+ },
+ },
+ ]
+ },
+ }
+ }