Skip to content
Snippets Groups Projects
Commit 69b96fbc authored by Lluis Gifre Renom's avatar Lluis Gifre Renom
Browse files

Device - IETF ACTN Driver:

- Added test script
- Improved unitary test
- Minor code fixings
- Removed unneeded files
- Added related objects factory methods
parent f1d61a8c
No related branches found
No related tags found
2 merge requests!235Release TeraFlowSDN 3.0,!188Resolve "(CTTC) Implement SBI Driver ACTN"
#!/bin/bash
# Copyright 2022-2023 ETSI TeraFlowSDN - TFS OSG (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.
PROJECTDIR=`pwd`
cd $PROJECTDIR/src
RCFILE=$PROJECTDIR/coverage/.coveragerc
# Run unitary tests and analyze coverage of code at same time
coverage run --rcfile=$RCFILE --append -m pytest --log-level=INFO --verbose \
device/tests/test_unitary_ietf_actn.py
...@@ -46,6 +46,10 @@ DEVICE_P4_DRIVERS = [DeviceDriverEnum.DEVICEDRIVER_P4] ...@@ -46,6 +46,10 @@ DEVICE_P4_DRIVERS = [DeviceDriverEnum.DEVICEDRIVER_P4]
DEVICE_TFS_TYPE = DeviceTypeEnum.TERAFLOWSDN_CONTROLLER.value DEVICE_TFS_TYPE = DeviceTypeEnum.TERAFLOWSDN_CONTROLLER.value
DEVICE_TFS_DRIVERS = [DeviceDriverEnum.DEVICEDRIVER_IETF_L2VPN] DEVICE_TFS_DRIVERS = [DeviceDriverEnum.DEVICEDRIVER_IETF_L2VPN]
DEVICE_IETF_ACTN_TYPE = DeviceTypeEnum.OPEN_LINE_SYSTEM.value
DEVICE_IETF_ACTN_DRIVERS = [DeviceDriverEnum.DEVICEDRIVER_IETF_ACTN]
def json_device_id(device_uuid : str): def json_device_id(device_uuid : str):
return {'device_uuid': {'uuid': device_uuid}} return {'device_uuid': {'uuid': device_uuid}}
...@@ -136,6 +140,14 @@ def json_device_tfs_disabled( ...@@ -136,6 +140,14 @@ def json_device_tfs_disabled(
device_uuid, DEVICE_TFS_TYPE, DEVICE_DISABLED, name=name, endpoints=endpoints, config_rules=config_rules, device_uuid, DEVICE_TFS_TYPE, DEVICE_DISABLED, name=name, endpoints=endpoints, config_rules=config_rules,
drivers=drivers) drivers=drivers)
def json_device_ietf_actn_disabled(
device_uuid : str, name : Optional[str] = None, endpoints : List[Dict] = [], config_rules : List[Dict] = [],
drivers : List[Dict] = DEVICE_IETF_ACTN_DRIVERS
):
return json_device(
device_uuid, DEVICE_IETF_ACTN_TYPE, DEVICE_DISABLED, name=name, endpoints=endpoints, config_rules=config_rules,
drivers=drivers)
def json_device_connect_rules(address : str, port : int, settings : Dict = {}) -> List[Dict]: def json_device_connect_rules(address : str, port : int, settings : Dict = {}) -> List[Dict]:
return [ return [
json_config_rule_set('_connect/address', address), json_config_rule_set('_connect/address', address),
......
...@@ -84,6 +84,15 @@ DRIVERS.append( ...@@ -84,6 +84,15 @@ DRIVERS.append(
} }
])) ]))
from .ietf_actn.IetfActnDriver import IetfActnDriver # pylint: disable=wrong-import-position
DRIVERS.append(
(IetfActnDriver, [
{
FilterFieldEnum.DEVICE_TYPE: DeviceTypeEnum.OPEN_LINE_SYSTEM,
FilterFieldEnum.DRIVER: DeviceDriverEnum.DEVICEDRIVER_IETF_ACTN,
}
]))
if LOAD_ALL_DEVICE_DRIVERS: if LOAD_ALL_DEVICE_DRIVERS:
from .openconfig.OpenConfigDriver import OpenConfigDriver # pylint: disable=wrong-import-position from .openconfig.OpenConfigDriver import OpenConfigDriver # pylint: disable=wrong-import-position
DRIVERS.append( DRIVERS.append(
......
osu_tunnel_1:
delay = 20
te_odu_number = 40
src_ttp_channel_name = 'och:1-odu2:1-oduflex:1-osuflex:2'
dst_ttp_channel_name = 'och:1-odu2:1-oduflex:3-osuflex:1'
etht_service_1:
etht_svc_type = 'op-mp2mp-svc'
src_endpoint_static_route_list:
dst='128.32.10.5', mask=24 => next_hop='128.32.33.5'
dst='128.32.20.5', mask=24 => next_hop='128.32.33.5'
dst_endpoint_static_route_list:
dst='172.1.101.22', mask=24 => next_hop='172.10.33.5'
...@@ -176,10 +176,10 @@ class EthtServiceHandler: ...@@ -176,10 +176,10 @@ class EthtServiceHandler:
'dst_node_id' : dst_endpoint['node-id'], 'dst_node_id' : dst_endpoint['node-id'],
'dst_tp_id' : dst_endpoint['tp-id'], 'dst_tp_id' : dst_endpoint['tp-id'],
'dst_vlan_tag' : src_endpoint['outer-tag']['vlan-value'], 'dst_vlan_tag' : dst_endpoint['outer-tag']['vlan-value'],
'dst_static_routes': [ 'dst_static_routes': [
(static_route['destination'], static_route['destination-mask'], static_route['next-hop']) (static_route['destination'], static_route['destination-mask'], static_route['next-hop'])
for static_route in src_endpoint.get('static-route-list', list()) for static_route in dst_endpoint.get('static-route-list', list())
], ],
} }
etht_services.append(etht_service) etht_services.append(etht_service)
......
...@@ -127,11 +127,11 @@ class OsuTunnelHandler: ...@@ -127,11 +127,11 @@ class OsuTunnelHandler:
osu_tunnel = { osu_tunnel = {
'name' : item['name'], 'name' : item['name'],
'src_node_id' : src_endpoint['node-id'], 'src_node_id' : src_endpoint['node-id'],
'src_tp_id' : src_endpoint['node-id'], 'src_tp_id' : src_endpoint['tp-id'],
'src_ttp_channel_name': src_endpoint['ttp-channel-name'], 'src_ttp_channel_name': src_endpoint['ttp-channel-name'],
'dst_node_id' : dst_endpoint['node-id'], 'dst_node_id' : dst_endpoint['node-id'],
'dst_tp_id' : dst_endpoint['node-id'], 'dst_tp_id' : dst_endpoint['tp-id'],
'dst_ttp_channel_name': src_endpoint['ttp-channel-name'], 'dst_ttp_channel_name': dst_endpoint['ttp-channel-name'],
'odu_type' : item['te-bandwidth']['odu-type'], 'odu_type' : item['te-bandwidth']['odu-type'],
'osuflex_number' : item['te-bandwidth']['number'], 'osuflex_number' : item['te-bandwidth']['number'],
'delay' : item['delay'], 'delay' : item['delay'],
......
...@@ -19,7 +19,7 @@ from typing import Any, Dict, List, Set, Tuple, Union ...@@ -19,7 +19,7 @@ from typing import Any, Dict, List, Set, Tuple, Union
LOGGER = logging.getLogger(__name__) LOGGER = logging.getLogger(__name__)
DEFAULT_BASE_URL = '/restconf/v2/data' DEFAULT_BASE_URL = '/restconf/v2/data'
DEFAULT_SCHEMA = 'https' DEFAULT_SCHEME = 'https'
DEFAULT_TIMEOUT = 120 DEFAULT_TIMEOUT = 120
DEFAULT_VERIFY = False DEFAULT_VERIFY = False
...@@ -41,12 +41,12 @@ class RestApiClient: ...@@ -41,12 +41,12 @@ class RestApiClient:
password = settings.get('password') password = settings.get('password')
self._auth = HTTPBasicAuth(username, password) if username is not None and password is not None else None self._auth = HTTPBasicAuth(username, password) if username is not None and password is not None else None
scheme = settings.get('scheme', DEFAULT_SCHEMA ) scheme = settings.get('scheme', DEFAULT_SCHEME )
base_url = settings.get('base_url', DEFAULT_BASE_URL) base_url = settings.get('base_url', DEFAULT_BASE_URL)
self._base_url = '{:s}://{:s}:{:d}{:s}'.format(scheme, address, int(port), base_url) self._base_url = '{:s}://{:s}:{:d}{:s}'.format(scheme, address, int(port), base_url)
self._timeout = int(settings.get('timeout', DEFAULT_TIMEOUT)) self._timeout = int(settings.get('timeout', DEFAULT_TIMEOUT))
self._verify = int(settings.get('verify', DEFAULT_VERIFY )) self._verify = bool(settings.get('verify', DEFAULT_VERIFY))
def get( def get(
self, object_name : str, url : str, filters : List[Tuple[str, str]], self, object_name : str, url : str, filters : List[Tuple[str, str]],
......
# Copyright 2022-2023 ETSI TeraFlowSDN - TFS OSG (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 copy, logging, os, pytest, time
from flask import Flask
from common.proto.context_pb2 import ConfigActionEnum, Device, DeviceId
from common.tools.grpc.Tools import grpc_message_to_json_string
from common.tools.object_factory.ConfigRule import json_config_rule_delete, json_config_rule_set
from common.tools.object_factory.Device import (
json_device_connect_rules, json_device_id, json_device_ietf_actn_disabled
)
from common.tools.service.GenericRestServer import GenericRestServer
from context.client.ContextClient import ContextClient
from device.client.DeviceClient import DeviceClient
from device.service.DeviceService import DeviceService
from device.service.driver_api._Driver import _Driver
from tests.tools.mock_ietf_actn_sdn_ctrl.ResourceEthServices import EthService, EthServices
from tests.tools.mock_ietf_actn_sdn_ctrl.ResourceOsuTunnels import OsuTunnel, OsuTunnels
os.environ['DEVICE_EMULATED_ONLY'] = 'TRUE'
from .PrepareTestScenario import ( # pylint: disable=unused-import
# be careful, order of symbols is important here!
mock_service, device_service, context_client, device_client, test_prepare_environment
)
DEVICE_UUID = 'DEVICE-IETF-ACTN'
DEVICE_ADDRESS = '127.0.0.1'
DEVICE_PORT = 8080
DEVICE_USERNAME = 'admin'
DEVICE_PASSWORD = 'admin'
DEVICE_SCHEME = 'http'
DEVICE_BASE_URL = '/restconf/v2/data'
DEVICE_TIMEOUT = 120
DEVICE_VERIFY = False
DEVICE_ID = json_device_id(DEVICE_UUID)
DEVICE = json_device_ietf_actn_disabled(DEVICE_UUID)
DEVICE_CONNECT_RULES = json_device_connect_rules(DEVICE_ADDRESS, DEVICE_PORT, {
'scheme' : DEVICE_SCHEME,
'username': DEVICE_USERNAME,
'password': DEVICE_PASSWORD,
'base_url': DEVICE_BASE_URL,
'timeout' : DEVICE_TIMEOUT,
'verify' : DEVICE_VERIFY,
})
DEVICE_CONFIG_RULES = [
json_config_rule_set('/osu_tunnels/osu_tunnel[osu_tunnel_1]', {
'name' : 'osu_tunnel_1',
'src_node_id' : '10.0.10.1',
'src_tp_id' : '200',
'src_ttp_channel_name': 'och:1-odu2:1-oduflex:1-osuflex:2',
'dst_node_id' : '10.0.30.1',
'dst_tp_id' : '200',
'dst_ttp_channel_name': 'och:1-odu2:1-oduflex:3-osuflex:1',
'odu_type' : 'osuflex',
'osuflex_number' : 40,
'delay' : 20,
'bidirectional' : True,
}),
json_config_rule_set('/etht_services/etht_service[etht_service_1]', {
'name' : 'etht_service_1',
'service_type' : 'op-mp2mp-svc',
'osu_tunnel_name' : 'osu_tunnel_1',
'src_node_id' : '10.0.10.1',
'src_tp_id' : '200',
'src_vlan_tag' : 21,
'src_static_routes': [('128.32.10.5', 24, '128.32.33.5'), ('128.32.20.5', 24, '128.32.33.5')],
'dst_node_id' : '10.0.30.1',
'dst_tp_id' : '200',
'dst_vlan_tag' : 101,
'dst_static_routes': [('172.1.101.22', 24, '172.10.33.5')],
}),
]
DEVICE_DECONFIG_RULES = [
json_config_rule_delete('/osu_tunnels/osu_tunnel[osu_tunnel_1]', {'name': 'osu_tunnel_1'}),
json_config_rule_delete('/etht_services/etht_service[etht_service_1]', {'name': 'etht_service_1'}),
]
LOGGER = logging.getLogger(__name__)
LOGGER.setLevel(logging.DEBUG)
@pytest.fixture(scope='session')
def ietf_actn_sdn_ctrl(
device_service: DeviceService, # pylint: disable=redefined-outer-name
) -> Flask:
_rest_server = GenericRestServer(DEVICE_PORT, DEVICE_BASE_URL, bind_address=DEVICE_ADDRESS)
_rest_server.app.debug = True
_rest_server.app.env = 'development'
_rest_server.app.testing = True
_rest_server.add_resource(OsuTunnels, '/ietf-te:tunnel')
_rest_server.add_resource(OsuTunnel, '/ietf-te:tunnel[name=<string:name>]')
_rest_server.add_resource(EthServices, '/ietf-eth-tran-service:etht-svc')
_rest_server.add_resource(EthService, '/ietf-eth-tran-service:etht-svc[etht-svc-name=<string:etht_svc_name>]')
_rest_server.start()
time.sleep(1) # bring time for the server to start
yield _rest_server
_rest_server.shutdown()
_rest_server.join()
def test_device_ietf_actn_add_correct(
device_client: DeviceClient, # pylint: disable=redefined-outer-name
device_service: DeviceService, # pylint: disable=redefined-outer-name
) -> None:
DEVICE_WITH_CONNECT_RULES = copy.deepcopy(DEVICE)
DEVICE_WITH_CONNECT_RULES['device_config']['config_rules'].extend(DEVICE_CONNECT_RULES)
device_client.AddDevice(Device(**DEVICE_WITH_CONNECT_RULES))
driver_instance_cache = device_service.device_servicer.driver_instance_cache
driver: _Driver = driver_instance_cache.get(DEVICE_UUID)
assert driver is not None
def test_device_ietf_actn_get(
context_client: ContextClient, # pylint: disable=redefined-outer-name
device_client: DeviceClient, # pylint: disable=redefined-outer-name
) -> None:
initial_config = device_client.GetInitialConfig(DeviceId(**DEVICE_ID))
LOGGER.info('initial_config = {:s}'.format(grpc_message_to_json_string(initial_config)))
device_data = context_client.GetDevice(DeviceId(**DEVICE_ID))
LOGGER.info('device_data = {:s}'.format(grpc_message_to_json_string(device_data)))
def test_device_ietf_actn_configure(
context_client: ContextClient, # pylint: disable=redefined-outer-name
device_client: DeviceClient, # pylint: disable=redefined-outer-name
device_service: DeviceService, # pylint: disable=redefined-outer-name
) -> None:
driver_instance_cache = device_service.device_servicer.driver_instance_cache
driver : _Driver = driver_instance_cache.get(DEVICE_UUID)
assert driver is not None
# Requires to retrieve data from device; might be slow. Uncomment only when needed and test does not pass directly.
#driver_config = sorted(driver.GetConfig(), key=operator.itemgetter(0))
#LOGGER.info('driver_config = {:s}'.format(str(driver_config)))
DEVICE_WITH_CONFIG_RULES = copy.deepcopy(DEVICE)
DEVICE_WITH_CONFIG_RULES['device_config']['config_rules'].extend(DEVICE_CONFIG_RULES)
device_client.ConfigureDevice(Device(**DEVICE_WITH_CONFIG_RULES))
# Requires to retrieve data from device; might be slow. Uncomment only when needed and test does not pass directly.
#driver_config = sorted(driver.GetConfig(), key=operator.itemgetter(0))
#LOGGER.info('driver_config = {:s}'.format(str(driver_config)))
device_data = context_client.GetDevice(DeviceId(**DEVICE_ID))
config_rules = [
(ConfigActionEnum.Name(config_rule.action), config_rule.custom.resource_key, config_rule.custom.resource_value)
for config_rule in device_data.device_config.config_rules
if config_rule.WhichOneof('config_rule') == 'custom'
]
LOGGER.info('device_data.device_config.config_rules = \n{:s}'.format(
'\n'.join(['{:s} {:s} = {:s}'.format(*config_rule) for config_rule in config_rules])))
for config_rule in DEVICE_CONFIG_RULES:
assert 'custom' in config_rule
config_rule = (
ConfigActionEnum.Name(config_rule['action']), config_rule['custom']['resource_key'],
config_rule['custom']['resource_value'])
assert config_rule in config_rules
def test_device_ietf_actn_deconfigure(
context_client: ContextClient, # pylint: disable=redefined-outer-name
device_client: DeviceClient, # pylint: disable=redefined-outer-name
device_service: DeviceService, # pylint: disable=redefined-outer-name
) -> None:
driver_instance_cache = device_service.device_servicer.driver_instance_cache
driver: _Driver = driver_instance_cache.get(DEVICE_UUID)
assert driver is not None
# Requires to retrieve data from device; might be slow. Uncomment only when needed and test does not pass directly.
#driver_config = sorted(driver.GetConfig(), key=operator.itemgetter(0))
#LOGGER.info('driver_config = {:s}'.format(str(driver_config)))
DEVICE_WITH_DECONFIG_RULES = copy.deepcopy(DEVICE)
DEVICE_WITH_DECONFIG_RULES['device_config']['config_rules'].extend(DEVICE_DECONFIG_RULES)
device_client.ConfigureDevice(Device(**DEVICE_WITH_DECONFIG_RULES))
# Requires to retrieve data from device; might be slow. Uncomment only when needed and test does not pass directly.
#driver_config = sorted(driver.GetConfig(), key=operator.itemgetter(0))
#LOGGER.info('driver_config = {:s}'.format(str(driver_config)))
device_data = context_client.GetDevice(DeviceId(**DEVICE_ID))
config_rules = [
(ConfigActionEnum.Name(config_rule.action), config_rule.custom.resource_key, config_rule.custom.resource_value)
for config_rule in device_data.device_config.config_rules
if config_rule.WhichOneof('config_rule') == 'custom'
]
LOGGER.info('device_data.device_config.config_rules = \n{:s}'.format(
'\n'.join(['{:s} {:s} = {:s}'.format(*config_rule) for config_rule in config_rules])))
for config_rule in DEVICE_DECONFIG_RULES:
assert 'custom' in config_rule
action_set = ConfigActionEnum.Name(ConfigActionEnum.CONFIGACTION_SET)
config_rule = (action_set, config_rule['custom']['resource_key'], config_rule['custom']['resource_value'])
assert config_rule not in config_rules
def test_device_ietf_actn_delete(
device_client : DeviceClient, # pylint: disable=redefined-outer-name
device_service : DeviceService, # pylint: disable=redefined-outer-name
) -> None:
device_client.DeleteDevice(DeviceId(**DEVICE_ID))
driver_instance_cache = device_service.device_servicer.driver_instance_cache
driver : _Driver = driver_instance_cache.get(DEVICE_UUID, {})
assert driver is None
0% Loading or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment