Test Tools - MW Mock SDN Ctrl: (a7acb1bc) · Commits · TFS / controller

src/tests/tools/mock_mw_sdn_ctrl/MockMWSdnCtrl.py

+17 −36

Original line number	Diff line number	Diff line
		@@ -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)
		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']
		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)
		if not isinstance(json_services, list): abort(400)
		if len(json_services) != 1: 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')
		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...')

src/tests/tools/mock_mw_sdn_ctrl/test_mw.py

0 → 100644

+84 −0

Original line number	Diff line number	Diff line
		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)