diff --git a/src/tests/tools/mock_mw_sdn_ctrl/MockMWSdnCtrl.py b/src/tests/tools/mock_mw_sdn_ctrl/MockMWSdnCtrl.py
index d4aabbe5b0325e476fc524fc26dec00a98758c46..d3d9de4a297e8967604e6479314b6a70ff19930b 100644
--- a/src/tests/tools/mock_mw_sdn_ctrl/MockMWSdnCtrl.py
+++ b/src/tests/tools/mock_mw_sdn_ctrl/MockMWSdnCtrl.py
@@ -25,8 +25,7 @@
# Ref: https://blog.miguelgrinberg.com/post/designing-a-restful-api-using-flask-restful
import functools, logging, sys, time
-from flask import Flask, abort, request
-from flask.json import jsonify
+from flask import Flask, abort, jsonify, make_response, request
from flask_restful import Api, Resource
BIND_ADDRESS = '0.0.0.0'
@@ -77,55 +76,37 @@ def log_request(logger : logging.Logger, response):
return response
class Health(Resource):
- def get(self): return jsonify({})
+ def get(self):
+ return make_response(jsonify({}), 200)
class Network(Resource):
def get(self, network_uuid : str):
if network_uuid != 'SIAE-ETH-TOPOLOGY': abort(400)
network = {'node': NETWORK_NODES, 'ietf-network-topology:link': NETWORK_LINKS}
- return jsonify({'ietf-network:network': network})
+ return make_response(jsonify({'ietf-network:network': network}), 200)
class Services(Resource):
def get(self):
services = [service for service in NETWORK_SERVICES.values()]
- return jsonify({'ietf-eth-tran-service:etht-svc': {'etht-svc-instances': services}})
+ return make_response(jsonify({'ietf-eth-tran-service:etht-svc': {'etht-svc-instances': services}}), 200)
def post(self):
- LOGGER.info('[post] begin')
- try:
- json_request = request.get_json()
- LOGGER.info('[post] json_request={:s}'.format(str(json_request)))
- if not json_request:
- LOGGER.info('[post] abort 1')
- abort(400)
- if not isinstance(json_request, dict):
- LOGGER.info('[post] abort 2')
- abort(400)
- if 'etht-svc-instances' not in json_request:
- LOGGER.info('[post] abort 3')
- abort(400)
- json_services = json_request['etht-svc-instances']
- LOGGER.info('[post] json_services={:s}'.format(str(json_services)))
- if not isinstance(json_services, list):
- LOGGER.info('[post] abort 4')
- abort(400)
- if len(json_services) != 1:
- LOGGER.info('[post] abort 5')
- abort(400)
- svc_data = json_services[0]
- LOGGER.info('[post] svc_data={:s}'.format(str(svc_data)))
- etht_svc_name = svc_data['etht-svc-name']
- LOGGER.info('[post] etht_svc_name={:s}'.format(str(etht_svc_name)))
- NETWORK_SERVICES[etht_svc_name] = svc_data
- LOGGER.info('[post] done')
- return jsonify({}), 201
- except:
- LOGGER.exception('Exception in POST')
+ json_request = request.get_json()
+ if not json_request: abort(400)
+ if not isinstance(json_request, dict): abort(400)
+ if 'etht-svc-instances' not in json_request: abort(400)
+ json_services = json_request['etht-svc-instances']
+ if not isinstance(json_services, list): abort(400)
+ if len(json_services) != 1: abort(400)
+ svc_data = json_services[0]
+ etht_svc_name = svc_data['etht-svc-name']
+ NETWORK_SERVICES[etht_svc_name] = svc_data
+ return make_response(jsonify({}), 201)
class DelServices(Resource):
def delete(self, service_uuid : str):
NETWORK_SERVICES.pop(service_uuid, None)
- return jsonify({}), 204
+ return make_response(jsonify({}), 204)
def main():
LOGGER.info('Starting...')
diff --git a/src/tests/tools/mock_mw_sdn_ctrl/test_mw.py b/src/tests/tools/mock_mw_sdn_ctrl/test_mw.py
new file mode 100644
index 0000000000000000000000000000000000000000..0329d30ad234398200c0fe29aac46f72f5a2e924
--- /dev/null
+++ b/src/tests/tools/mock_mw_sdn_ctrl/test_mw.py
@@ -0,0 +1,84 @@
+import json, logging, requests
+from requests.auth import HTTPBasicAuth
+from typing import Optional
+
+LOGGER = logging.getLogger(__name__)
+
+HTTP_OK_CODES = {
+ 200, # OK
+ 201, # Created
+ 202, # Accepted
+ 204, # No Content
+}
+
+def create_connectivity_service(
+ root_url, uuid, node_id_src, tp_id_src, node_id_dst, tp_id_dst, vlan_id,
+ auth : Optional[HTTPBasicAuth] = None, timeout : Optional[int] = None
+):
+
+ url = '{:s}/nmswebs/restconf/data/ietf-eth-tran-service:etht-svc'.format(root_url)
+ headers = {'content-type': 'application/json'}
+ data = {
+ 'etht-svc-instances': [
+ {
+ 'etht-svc-name': uuid,
+ 'etht-svc-type': 'ietf-eth-tran-types:p2p-svc',
+ 'etht-svc-end-points': [
+ {
+ 'etht-svc-access-points': [
+ {'access-node-id': node_id_src, 'access-ltp-id': tp_id_src, 'access-point-id': '1'}
+ ],
+ 'outer-tag': {'vlan-value': vlan_id, 'tag-type': 'ietf-eth-tran-types:classify-c-vlan'},
+ 'etht-svc-end-point-name': '{:s}:{:s}'.format(str(node_id_src), str(tp_id_src)),
+ 'service-classification-type': 'ietf-eth-tran-types:vlan-classification'
+ },
+ {
+ 'etht-svc-access-points': [
+ {'access-node-id': node_id_dst, 'access-ltp-id': tp_id_dst, 'access-point-id': '2'}
+ ],
+ 'outer-tag': {'vlan-value': vlan_id, 'tag-type': 'ietf-eth-tran-types:classify-c-vlan'},
+ 'etht-svc-end-point-name': '{:s}:{:s}'.format(str(node_id_dst), str(tp_id_dst)),
+ 'service-classification-type': 'ietf-eth-tran-types:vlan-classification'
+ }
+ ]
+ }
+ ]
+ }
+ results = []
+ try:
+ LOGGER.info('Connectivity service {:s}: {:s}'.format(str(uuid), str(data)))
+ response = requests.post(
+ url=url, data=json.dumps(data), timeout=timeout, headers=headers, verify=False, auth=auth)
+ LOGGER.info('Microwave Driver response: {:s}'.format(str(response)))
+ except Exception as e: # pylint: disable=broad-except
+ LOGGER.exception('Exception creating ConnectivityService(uuid={:s}, data={:s})'.format(str(uuid), str(data)))
+ results.append(e)
+ else:
+ if response.status_code not in HTTP_OK_CODES:
+ msg = 'Could not create ConnectivityService(uuid={:s}, data={:s}). status_code={:s} reply={:s}'
+ LOGGER.error(msg.format(str(uuid), str(data), str(response.status_code), str(response)))
+ results.append(response.status_code in HTTP_OK_CODES)
+ return results
+
+def delete_connectivity_service(root_url, uuid, auth : Optional[HTTPBasicAuth] = None, timeout : Optional[int] = None):
+ url = '{:s}/nmswebs/restconf/data/ietf-eth-tran-service:etht-svc/etht-svc-instances={:s}'
+ url = url.format(root_url, uuid)
+ results = []
+ try:
+ response = requests.delete(url=url, timeout=timeout, verify=False, auth=auth)
+ except Exception as e: # pylint: disable=broad-except
+ LOGGER.exception('Exception deleting ConnectivityService(uuid={:s})'.format(str(uuid)))
+ results.append(e)
+ else:
+ if response.status_code not in HTTP_OK_CODES:
+ msg = 'Could not delete ConnectivityService(uuid={:s}). status_code={:s} reply={:s}'
+ LOGGER.error(msg.format(str(uuid), str(response.status_code), str(response)))
+ results.append(response.status_code in HTTP_OK_CODES)
+ return results
+
+if __name__ == '__main__':
+ ROOT_URL = 'https://127.0.0.1:8443'
+ SERVICE_UUID = 'my-service'
+
+ create_connectivity_service(ROOT_URL, SERVICE_UUID, '172.18.0.1', '1', '172.18.0.2', '2', 300)
+ delete_connectivity_service(ROOT_URL, SERVICE_UUID)