End-to-end test - L2 VPN gNMI OpenConfig: (0328f8c7) · Commits · TFS / controller

src/tests/l2_vpn_gnmi_oc/data/ietf-l2vpn-service.json

0 → 100644

+70 −0

Original line number	Diff line number	Diff line
		{
		"ietf-l2vpn-svc:l2vpn-svc": {
		"vpn-services": {"vpn-service": [
		{
		"vpn-id": "ietf-l2vpn-svc",
		"vpn-svc-type": "vpws",
		"svc-topo": "any-to-any",
		"customer-name": "somebody"
		}
		]},
		"sites": {
		"site": [
		{
		"site-id": "site_DC1",
		"management": {"type": "ietf-l2vpn-svc:provider-managed"},
		"locations": {"location": [{"location-id": "DC1"}]},
		"devices": {"device": [{"device-id": "dc1", "location": "DC1"}]},
		"site-network-accesses": {
		"site-network-access": [
		{
		"network-access-id": "eth1",
		"type": "ietf-l3vpn-svc:multipoint",
		"device-reference": "dc1",
		"vpn-attachment": {
		"vpn-id": "ietf-l2vpn-svc",
		"site-role": "ietf-l2vpn-svc:any-to-any-role"
		},
		"bearer": {"bearer-reference": "r1:Ethernet10"},
		"connection": {
		"encapsulation-type": "vlan",
		"tagged-interface": {
		"type": "ietf-l2vpn-svc:dot1q",
		"dot1q-vlan-tagged": {"cvlan-id": 125}
		}
		}
		}
		]
		}
		},
		{
		"site-id": "site_DC2",
		"management": {"type": "ietf-l2vpn-svc:provider-managed"},
		"locations": {"location": [{"location-id": "DC2"}]},
		"devices": {"device": [{"device-id": "dc2", "location": "DC2"}]},
		"site-network-accesses": {
		"site-network-access": [
		{
		"network-access-id": "eth1",
		"type": "ietf-l3vpn-svc:multipoint",
		"device-reference": "dc2",
		"vpn-attachment": {
		"vpn-id": "ietf-l2vpn-svc",
		"site-role": "ietf-l2vpn-svc:any-to-any-role"
		},
		"bearer": {"bearer-reference": "r3:Ethernet10"},
		"connection": {
		"encapsulation-type": "vlan",
		"tagged-interface": {
		"type": "ietf-l2vpn-svc:dot1q",
		"dot1q-vlan-tagged": {"cvlan-id": 125}
		}
		}
		}
		]
		}
		}
		]
		}
		}
		}

src/tests/l2_vpn_gnmi_oc/tests/Fixtures.py

+0 −15

Original line number	Diff line number	Diff line
		@@ -13,24 +13,9 @@
		# limitations under the License.

		import pytest
		from common.Constants import ServiceNameEnum
		from common.Settings import get_service_host, get_service_port_http
		from context.client.ContextClient import ContextClient
		from device.client.DeviceClient import DeviceClient
		from service.client.ServiceClient import ServiceClient
		from .MockOSM import MockOSM
		from .OSM_Constants import WIM_MAPPING

		NBI_ADDRESS = get_service_host(ServiceNameEnum.NBI)
		NBI_PORT = get_service_port_http(ServiceNameEnum.NBI)
		NBI_USERNAME = 'admin'
		NBI_PASSWORD = 'admin'
		NBI_BASE_URL = ''

		@pytest.fixture(scope='session')
		def osm_wim() -> MockOSM:
		wim_url = 'http://{:s}:{:d}'.format(NBI_ADDRESS, NBI_PORT)
		return MockOSM(wim_url, WIM_MAPPING, NBI_USERNAME, NBI_PASSWORD)

		@pytest.fixture(scope='session')
		def context_client() -> ContextClient:

src/tests/l2_vpn_gnmi_oc/tests/MockOSM.py

deleted100644 → 0

+0 −62

Original line number	Diff line number	Diff line
		# Copyright 2022-2025 ETSI 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.

		import logging
		from .WimconnectorIETFL2VPN import WimconnectorIETFL2VPN

		LOGGER = logging.getLogger(__name__)

		class MockOSM:
		def __init__(self, url, mapping, username, password):
		wim = {'wim_url': url}
		wim_account = {'user': username, 'password': password}
		config = {'mapping_not_needed': False, 'service_endpoint_mapping': mapping}
		self.wim = WimconnectorIETFL2VPN(wim, wim_account, config=config)
		self.conn_info = {} # internal database emulating OSM storage provided to WIM Connectors

		def create_connectivity_service(self, service_type, connection_points):
		LOGGER.info('[create_connectivity_service] service_type={:s}'.format(str(service_type)))
		LOGGER.info('[create_connectivity_service] connection_points={:s}'.format(str(connection_points)))
		self.wim.check_credentials()
		result = self.wim.create_connectivity_service(service_type, connection_points)
		LOGGER.info('[create_connectivity_service] result={:s}'.format(str(result)))
		service_uuid, conn_info = result
		self.conn_info[service_uuid] = conn_info
		return service_uuid

		def get_connectivity_service_status(self, service_uuid):
		LOGGER.info('[get_connectivity_service] service_uuid={:s}'.format(str(service_uuid)))
		conn_info = self.conn_info.get(service_uuid)
		if conn_info is None: raise Exception('ServiceId({:s}) not found'.format(str(service_uuid)))
		LOGGER.info('[get_connectivity_service] conn_info={:s}'.format(str(conn_info)))
		self.wim.check_credentials()
		result = self.wim.get_connectivity_service_status(service_uuid, conn_info=conn_info)
		LOGGER.info('[get_connectivity_service] result={:s}'.format(str(result)))
		return result

		def edit_connectivity_service(self, service_uuid, connection_points):
		LOGGER.info('[edit_connectivity_service] service_uuid={:s}'.format(str(service_uuid)))
		LOGGER.info('[edit_connectivity_service] connection_points={:s}'.format(str(connection_points)))
		conn_info = self.conn_info.get(service_uuid)
		if conn_info is None: raise Exception('ServiceId({:s}) not found'.format(str(service_uuid)))
		LOGGER.info('[edit_connectivity_service] conn_info={:s}'.format(str(conn_info)))
		self.wim.edit_connectivity_service(service_uuid, conn_info=conn_info, connection_points=connection_points)

		def delete_connectivity_service(self, service_uuid):
		LOGGER.info('[delete_connectivity_service] service_uuid={:s}'.format(str(service_uuid)))
		conn_info = self.conn_info.get(service_uuid)
		if conn_info is None: raise Exception('ServiceId({:s}) not found'.format(str(service_uuid)))
		LOGGER.info('[delete_connectivity_service] conn_info={:s}'.format(str(conn_info)))
		self.wim.check_credentials()
		self.wim.delete_connectivity_service(service_uuid, conn_info=conn_info)

src/tests/l2_vpn_gnmi_oc/tests/OSM_Constants.py

deleted100644 → 0

+0 −53

Original line number	Diff line number	Diff line
		# Copyright 2022-2025 ETSI 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.


		# Ref: https://osm.etsi.org/wikipub/index.php/WIM
		WIM_MAPPING = [
		{
		'device-id' : 'dc1', # pop_switch_dpid
		#'device_interface_id' : ??, # pop_switch_port
		'service_endpoint_id' : 'ep-1', # wan_service_endpoint_id
		'service_mapping_info': { # wan_service_mapping_info, other extra info
		'bearer': {'bearer-reference': 'r1:Ethernet10'},
		'site-id': '1',
		},
		#'switch_dpid' : ??, # wan_switch_dpid
		#'switch_port' : ??, # wan_switch_port
		#'datacenter_id' : ??, # vim_account
		},
		{
		'device-id' : 'dc2', # pop_switch_dpid
		#'device_interface_id' : ??, # pop_switch_port
		'service_endpoint_id' : 'ep-2', # wan_service_endpoint_id
		'service_mapping_info': { # wan_service_mapping_info, other extra info
		'bearer': {'bearer-reference': 'r3:Ethernet10'},
		'site-id': '2',
		},
		#'switch_dpid' : ??, # wan_switch_dpid
		#'switch_port' : ??, # wan_switch_port
		#'datacenter_id' : ??, # vim_account
		},
		]

		SERVICE_TYPE = 'ELINE'

		SERVICE_CONNECTION_POINTS = [
		{'service_endpoint_id': 'ep-1',
		'service_endpoint_encapsulation_type': 'dot1q',
		'service_endpoint_encapsulation_info': {'vlan': 125}},
		{'service_endpoint_id': 'ep-2',
		'service_endpoint_encapsulation_type': 'dot1q',
		'service_endpoint_encapsulation_info': {'vlan': 125}},
		]

src/tests/l2_vpn_gnmi_oc/tests/Tools.py

0 → 100644

+109 −0

Original line number	Diff line number	Diff line
		# Copyright 2022-2025 ETSI 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.

		import enum, logging, requests
		from typing import Any, Dict, List, Optional, Set, Union
		from common.Constants import ServiceNameEnum
		from common.Settings import get_service_host, get_service_port_http

		NBI_ADDRESS = get_service_host(ServiceNameEnum.NBI)
		NBI_PORT = get_service_port_http(ServiceNameEnum.NBI)
		NBI_USERNAME = 'admin'
		NBI_PASSWORD = 'admin'
		NBI_BASE_URL = ''

		class RestRequestMethod(enum.Enum):
		GET = 'get'
		POST = 'post'
		PUT = 'put'
		PATCH = 'patch'
		DELETE = 'delete'

		EXPECTED_STATUS_CODES : Set[int] = {
		requests.codes['OK' ],
		requests.codes['CREATED' ],
		requests.codes['ACCEPTED' ],
		requests.codes['NO_CONTENT'],
		}

		def do_rest_request(
		method : RestRequestMethod, url : str, body : Optional[Any] = None, timeout : int = 10,
		allow_redirects : bool = True, expected_status_codes : Set[int] = EXPECTED_STATUS_CODES,
		logger : Optional[logging.Logger] = None
		) -> Optional[Union[Dict, List]]:
		request_url = 'http://{:s}:{:s}@{:s}:{:d}{:s}{:s}'.format(
		NBI_USERNAME, NBI_PASSWORD, NBI_ADDRESS, NBI_PORT, str(NBI_BASE_URL), url
		)

		if logger is not None:
		msg = 'Request: {:s} {:s}'.format(str(method.value).upper(), str(request_url))
		if body is not None: msg += ' body={:s}'.format(str(body))
		logger.warning(msg)
		reply = requests.request(method.value, request_url, timeout=timeout, json=body, allow_redirects=allow_redirects)
		if logger is not None:
		logger.warning('Reply: {:s}'.format(str(reply.text)))
		assert reply.status_code in expected_status_codes, 'Reply failed with status code {:d}'.format(reply.status_code)

		if reply.content and len(reply.content) > 0: return reply.json()
		return None

		def do_rest_get_request(
		url : str, body : Optional[Any] = None, timeout : int = 10,
		allow_redirects : bool = True, expected_status_codes : Set[int] = EXPECTED_STATUS_CODES,
		logger : Optional[logging.Logger] = None
		) -> Optional[Union[Dict, List]]:
		return do_rest_request(
		RestRequestMethod.GET, url, body=body, timeout=timeout, allow_redirects=allow_redirects,
		expected_status_codes=expected_status_codes, logger=logger
		)

		def do_rest_post_request(
		url : str, body : Optional[Any] = None, timeout : int = 10,
		allow_redirects : bool = True, expected_status_codes : Set[int] = EXPECTED_STATUS_CODES,
		logger : Optional[logging.Logger] = None
		) -> Optional[Union[Dict, List]]:
		return do_rest_request(
		RestRequestMethod.POST, url, body=body, timeout=timeout, allow_redirects=allow_redirects,
		expected_status_codes=expected_status_codes, logger=logger
		)

		def do_rest_put_request(
		url : str, body : Optional[Any] = None, timeout : int = 10,
		allow_redirects : bool = True, expected_status_codes : Set[int] = EXPECTED_STATUS_CODES,
		logger : Optional[logging.Logger] = None
		) -> Optional[Union[Dict, List]]:
		return do_rest_request(
		RestRequestMethod.PUT, url, body=body, timeout=timeout, allow_redirects=allow_redirects,
		expected_status_codes=expected_status_codes, logger=logger
		)

		def do_rest_patch_request(
		url : str, body : Optional[Any] = None, timeout : int = 10,
		allow_redirects : bool = True, expected_status_codes : Set[int] = EXPECTED_STATUS_CODES,
		logger : Optional[logging.Logger] = None
		) -> Optional[Union[Dict, List]]:
		return do_rest_request(
		RestRequestMethod.PATCH, url, body=body, timeout=timeout, allow_redirects=allow_redirects,
		expected_status_codes=expected_status_codes, logger=logger
		)

		def do_rest_delete_request(
		url : str, body : Optional[Any] = None, timeout : int = 10,
		allow_redirects : bool = True, expected_status_codes : Set[int] = EXPECTED_STATUS_CODES,
		logger : Optional[logging.Logger] = None
		) -> Optional[Union[Dict, List]]:
		return do_rest_request(
		RestRequestMethod.DELETE, url, body=body, timeout=timeout, allow_redirects=allow_redirects,
		expected_status_codes=expected_status_codes, logger=logger
		)