src/device/service/drivers/openconfig/templates/Interfaces.py

@@ -31,10 +31,6 @@ def parse(xml_data : ET.Element) -> List[Tuple[str, Dict[str, Any]]]:
 
         interface = {}
 
-        interface_name = xml_interface.find('oci:name', namespaces=NAMESPACES)
-        if interface_name is None or interface_name.text is None: continue
-        add_value_from_tag(interface, 'name', interface_name)
-
         #interface_type = xml_interface.find('oci:config/oci:type', namespaces=NAMESPACES)
         #add_value_from_tag(interface, 'type', interface_type)
 
@@ -42,8 +38,11 @@ def parse(xml_data : ET.Element) -> List[Tuple[str, Dict[str, Any]]]:
             interface_type = xml_interface.find('oci:config/oci:type', namespaces=NAMESPACES)
         elif xml_interface.find('oci:state/oci:type', namespaces=NAMESPACES) is not None:
             interface_type = xml_interface.find('oci:state/oci:type', namespaces=NAMESPACES)
-        else:
-            interface_type = ''
+        else: continue
+            
+        interface_name = xml_interface.find('oci:name', namespaces=NAMESPACES)
+        if interface_name is None or interface_name.text is None: continue
+        add_value_from_tag(interface, 'name', interface_name)
             
         # Get the type of interface according to the vendor's type
         if 'ianaift:' in interface_type.text:

src/device/service/drivers/openconfig/templates/Inventory.py

@@ -54,7 +54,6 @@ XPATH_PORTS = "//ocp:components/ocp:component"
 
 def parse(xml_data : ET.Element) -> List[Tuple[str, Dict[str, Any]]]:
     response = []
-    LOGGER.debug("InventoryPrueba")
     parent_types = {}
     for xml_component in xml_data.xpath(XPATH_PORTS, namespaces=NAMESPACES):
         LOGGER.info('xml_component inventario = {:s}'.format(str(ET.tostring(xml_component))))
@@ -78,9 +77,9 @@ def parse(xml_data : ET.Element) -> List[Tuple[str, Dict[str, Any]]]:
             add_value_from_tag(inventory['attributes'], 'location', component_location)
         
         component_type = xml_component.find('ocp:state/ocp:type', namespaces=NAMESPACES)
-        component_type.text = component_type.text.replace('oc-platform-types:','')
-        if component_type is None: continue
-        add_value_from_tag(inventory, 'class', component_type)
+        if component_type is not None:
+            component_type.text = component_type.text.replace('oc-platform-types:','')
+            add_value_from_tag(inventory, 'class', component_type)
         
         if inventory['class'] == 'CPU' or inventory['class'] == 'STORAGE': continue
 

src/device/service/drivers/openconfig/templates/RoutingPolicy.py

@@ -35,7 +35,7 @@ def parse(xml_data : ET.Element) -> List[Tuple[str, Dict[str, Any]]]:
         #LOGGER.info('xml_policy_definition = {:s}'.format(str(ET.tostring(xml_policy_definition))))
 
         policy_definition = {}
-
+        statement_name = ''
         policy_name = xml_policy_definition.find('ocrp:name', namespaces=NAMESPACES)
         if policy_name is None or policy_name.text is None: continue
         add_value_from_tag(policy_definition, 'policy_name', policy_name)

src/device/tests/test_unitary_ietf_actn.py

@@ -151,6 +151,11 @@ def test_device_ietf_actn_configure(
     retrieved_driver_config_rules = sorted(driver.GetConfig(), key=operator.itemgetter(0))
     LOGGER.info('driver_config = {:s}'.format(str(retrieved_driver_config_rules)))
     assert isinstance(retrieved_driver_config_rules, list)
+    retrieved_driver_config_rules = [
+        (resource_key, resource_value)
+        for resource_key, resource_value in retrieved_driver_config_rules
+        if resource_key != '/endpoints/endpoint[mgmt]'
+    ]
     if len(retrieved_driver_config_rules) > 0:
         LOGGER.error('PRE DRIVER CONFIG RULES - Differences:\n{:s}'.format(str(retrieved_driver_config_rules)))
     assert len(retrieved_driver_config_rules) == 0
@@ -186,6 +191,7 @@ def test_device_ietf_actn_configure(
     retrieved_driver_config_rules = [
         {'action': 1, 'custom': {'resource_key': resource_key, 'resource_value': resource_value}}
         for resource_key, resource_value in retrieved_driver_config_rules
+        if resource_key != '/endpoints/endpoint[mgmt]'
     ]
     with open(DATA_FILE_CONFIG_RULES, 'r', encoding='UTF-8') as f:
         expected_driver_config_rules = sorted(json.load(f), key=lambda cr: cr['custom']['resource_key'])
@@ -231,6 +237,7 @@ def test_device_ietf_actn_deconfigure(
     retrieved_driver_config_rules = [
         {'action': 1, 'custom': {'resource_key': resource_key, 'resource_value': resource_value}}
         for resource_key, resource_value in retrieved_driver_config_rules
+        if resource_key != '/endpoints/endpoint[mgmt]'
     ]
     with open(DATA_FILE_CONFIG_RULES, 'r', encoding='UTF-8') as f:
         expected_driver_config_rules = sorted(json.load(f), key=lambda cr: cr['custom']['resource_key'])
@@ -266,6 +273,11 @@ def test_device_ietf_actn_deconfigure(
     retrieved_driver_config_rules = sorted(driver.GetConfig(), key=operator.itemgetter(0))
     LOGGER.info('retrieved_driver_config_rules = {:s}'.format(str(retrieved_driver_config_rules)))
     assert isinstance(retrieved_driver_config_rules, list)
+    retrieved_driver_config_rules = [
+        (resource_key, resource_value)
+        for resource_key, resource_value in retrieved_driver_config_rules
+        if resource_key != '/endpoints/endpoint[mgmt]'
+    ]
     if len(retrieved_driver_config_rules) > 0:
         LOGGER.error('POST DRIVER CONFIG RULES - Differences:\n{:s}'.format(str(retrieved_driver_config_rules)))
     assert len(retrieved_driver_config_rules) == 0

src/nbi/service/rest_server/nbi_plugins/ietf_l3vpn/L3VPN_Services.py

@@ -13,7 +13,7 @@
 # limitations under the License.
 
 import logging
-from typing import Dict
+from typing import Dict, List
 from flask import request
 from flask.json import jsonify
 from flask_restful import Resource
@@ -36,11 +36,40 @@ class L3VPN_Services(Resource):
         request_data : Dict = request.json
         LOGGER.debug('Request: {:s}'.format(str(request_data)))
 
+        errors = list()
+        if 'ietf-l3vpn-svc:l3vpn-services' in request_data:
+            # processing multiple L3VPN service requests formatted as:
+            #{
+            #  "ietf-l3vpn-svc:l3vpn-services": {
+            #    "l3vpn-svc": [
+            #      {
+            #        "service-id": "vpn1",
+            #        "vpn-services": {
+            #          "vpn-service": [
+            for l3vpn_svc in request_data['ietf-l3vpn-svc:l3vpn-services']['l3vpn-svc']:
+                l3vpn_svc.pop('service-id', None)
+                l3vpn_svc_request_data = {'ietf-l3vpn-svc:l3vpn-svc': l3vpn_svc}
+                errors.extend(self._process_l3vpn(l3vpn_svc_request_data))
+        elif 'ietf-l3vpn-svc:l3vpn-svc' in request_data:
+            # processing single (standard) L3VPN service request formatted as:
+            #{
+            #  "ietf-l3vpn-svc:l3vpn-svc": {
+            #    "vpn-services": {
+            #      "vpn-service": [
+            errors.extend(self._process_l3vpn(request_data))
+        else:
+            errors.append('unexpected request: {:s}'.format(str(request_data)))
+
+        response = jsonify(errors)
+        response.status_code = HTTP_CREATED if len(errors) == 0 else HTTP_SERVERERROR
+        return response
+
+    def _process_l3vpn(self, request_data : Dict) -> List[Dict]:
         yang_validator = YangValidator('ietf-l3vpn-svc')
         request_data = yang_validator.parse_to_dict(request_data)
         yang_validator.destroy()
 
-        errors = []
+        errors = list()
 
         for vpn_service in request_data['l3vpn-svc']['vpn-services']['vpn-service']:
             process_vpn_service(vpn_service, errors)
@@ -48,6 +77,4 @@ class L3VPN_Services(Resource):
         for site in request_data['l3vpn-svc']['sites']['site']:
             process_site(site, errors)
 
-        response = jsonify(errors)
-        response.status_code = HTTP_CREATED if len(errors) == 0 else HTTP_SERVERERROR
-        return response
+        return errors

src/tests/tools/mock_ietf_actn_sdn_ctrl/README.md

@@ -7,22 +7,17 @@ This REST server implements very basic support for the following YANG data model
   - Ref: https://datatracker.ietf.org/doc/draft-ietf-teas-yang-te/
 
 The aim of this server is to enable testing the IetfActnDeviceDriver and the IetfActnServiceHandler.
-Follow the steps below to perform the test:
 
-## 1. Deploy TeraFlowSDN controller and the scenario
-Deploy the test scenario "ietf_actn_deploy.sh":
-```bash
-source src/tests/tools/mock_ietf_actn_sdn_ctrl/scenario/ietf_actn_deploy.sh
-./deploy/all.sh
-```
 
-## 2. Install requirements and run the Mock IETF ACTN SDN controller
-__NOTE__: if you run the Mock IETF ACTN SDN controller from the PyEnv used for developping on the TeraFlowSDN framework,
+## 1. Install requirements for the Mock IETF ACTN SDN controller
+__NOTE__: if you run the Mock IETF ACTN SDN controller from the PyEnv used for developing on the TeraFlowSDN
+framework and you followed the official steps in
+[Development Guide > Configure Environment > Python](https://labs.etsi.org/rep/tfs/controller/-/wikis/2.-Development-Guide/2.1.-Configure-Environment/2.1.1.-Python),
 all the requirements are already in place. Install them only if you execute it in a separate/standalone environment.
 
 Install the required dependencies as follows:
 ```bash
-pip install Flask==2.1.3 Flask-RESTful==0.3.9
+pip install -r src/tests/tools/mock_ietf_actn_sdn_ctrl/requirements.in
 ```
 
 Run the Mock IETF ACTN SDN Controller as follows:
@@ -30,24 +25,9 @@ Run the Mock IETF ACTN SDN Controller as follows:
 python src/tests/tools/mock_ietf_actn_sdn_ctrl/MockIetfActnSdnCtrl.py
 ```
 
-## 3. Deploy the test descriptors
-Edit the descriptors to meet your environment specifications.
-Edit "network_descriptors.json" and change IP address and port of the IETF ACTN SDN controller of the "ACTN" device.
-- Set value of config rule "_connect/address" to the address of the host where the Mock IETF ACTN SDN controller is
-  running (default="192.168.1.1").
-- Set value of config rule "_connect/port" to the port where your Mock IETF ACTN SDN controller is listening on
-  (default="8443").
-
-Upload the "network_descriptors.json" through the TeraFlowSDN WebUI.
-- If not already selected, select Context(admin)/Topology(admin).
-- Check that a network topology with 4 routers + 1 IETF ACTN radio system are loaded. They should form 2 rings.
-
-Upload the "service_descriptor.json" through the TeraFlowSDN WebUI.
-- Check that 2 services have been created.
-- The "actn-svc" should have a connection and be supported by a sub-service.
-- The sub-service should also have a connection.
-- The R1, R3, and MW devices should have configuration rules established.
-
-# 4. Delete the IETF ACTN service
-Find the "mw-svc" on the WebUI, navigate to its details, and delete the service pressing the "Delete Service" button.
-The service, sub-service, and device configuration rules should be removed.
+
+## 2. Run the Mock IETF ACTN SDN controller
+Run the Mock IETF ACTN SDN Controller as follows:
+```bash
+python src/tests/tools/mock_ietf_actn_sdn_ctrl/MockIetfActnSdnCtrl.py
+```

src/tests/tools/mock_ietf_actn_sdn_ctrl/build.sh

@@ -13,6 +13,9 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+# Make folder containing the script the root folder for its execution
+cd $(dirname $0)
+
 docker build -t mock-ietf-actn-sdn-ctrl:test -f Dockerfile .
 docker tag mock-ietf-actn-sdn-ctrl:test localhost:32000/tfs/mock-ietf-actn-sdn-ctrl:test
 docker push localhost:32000/tfs/mock-ietf-actn-sdn-ctrl:test

src/tests/tools/mock_ietf_actn_sdn_ctrl/run.sh

@@ -13,4 +13,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+# Make folder containing the script the root folder for its execution
+cd $(dirname $0)
+
 python MockIetfActnSdnCtrl.py