Newer
Older
# Copyright 2022-2024 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.
from flask import Flask, request
from YangValidator import YangValidator
app = Flask(__name__)
# Retrieve the IP address of the current machine
current_ip = socket.gethostbyname(socket.gethostname())
yang_validator = YangValidator('etsi-qkd-sdn-node', ['etsi-qkd-node-types'])
# Update IP address with the current machine's IP
'qkdn_id': '00000001-0000-0000-0000-000000000000',
},
'qkdn_capabilities': {
},
'qkd_applications': {
'qkd_app': [
{
'app_id': '00000001-0001-0000-0000-000000000000',
'client_app_id': [],
'app_statistics': {
'statistics': []
},
'app_qos': {
},
'backing_qkdl_id': []
}
]
},
'qkd_interfaces': {
'qkd_interface': [
{
'qkdi_id': '100',
'qkdi_att_point': {
},
'qkdi_capabilities': {
}
},
{
'qkdi_id': '101',
'qkdi_att_point': {
},
'qkdi_capabilities': {
}
}
]
},
'qkd_links': {
'qkdn_id': '00000002-0000-0000-0000-000000000000',
},
'qkdn_capabilities': {
},
'qkd_applications': {
'qkd_app': [
{
'app_id': '00000002-0001-0000-0000-000000000000',
'client_app_id': [],
'app_statistics': {
'statistics': []
},
'app_qos': {
},
'backing_qkdl_id': []
}
]
},
'qkd_interfaces': {
'qkd_interface': [
{
'qkdi_id': '200',
'qkdi_att_point': {
},
'qkdi_capabilities': {
}
},
{
'qkdi_id': '201',
'qkdi_att_point': {
},
'qkdi_capabilities': {
}
},
{
'qkdi_id': '202',
'qkdi_att_point': {
},
'qkdi_capabilities': {
}
}
]
},
'qkd_links': {
'qkdn_id': '00000003-0000-0000-0000-000000000000',
},
'qkdn_capabilities': {
},
'qkd_applications': {
'qkd_app': [
{
'app_id': '00000003-0001-0000-0000-000000000000',
'client_app_id': [],
'app_statistics': {
'statistics': []
},
'app_qos': {
},
'backing_qkdl_id': []
}
]
},
'qkd_interfaces': {
'qkd_interface': [
{
'qkdi_id': '300',
'qkdi_att_point': {
},
'qkdi_capabilities': {
}
},
{
'qkdi_id': '301',
'qkdi_att_point': {
},
'qkdi_capabilities': {
}
}
]
},
'qkd_links': {
}
}
}
def get_side_effect(url):
steps = url.lstrip('https://').lstrip('http://').rstrip('/')
parts = steps.split('/')
# Ensure there are enough parts to unpack
if len(parts) < 4:
raise ValueError(f"Expected at least 4 parts in the URL, got {len(parts)}: {steps}")
ip_port, _, _, header, *steps = parts
header_splitted = header.split(':')
module = header_splitted[0]
assert(module == 'etsi-qkd-sdn-node')
if ip_port.startswith('127.0.0.1'):
ip_port = ip_port.replace('127.0.0.1', current_ip)
tree = {'qkd_node': nodes[ip_port]['node'].copy()}
if len(header_splitted) == 1 or not header_splitted[1]:
value = nodes[ip_port].copy()
value.pop('node')
tree['qkd_node'].update(value)
return tree, tree
root = header_splitted[1]
assert(root == 'qkd_node')
if not steps:
return tree, tree
endpoint, *steps = steps
value = nodes[ip_port][endpoint]
if not steps:
tree['qkd_node'].update(return_value)
return return_value, tree
raise Exception('Url too long')
def edit(from_dict, to_dict, create):
for key, value in from_dict.items():
if isinstance(value, dict):
if key not in to_dict and create:
to_dict[key] = {}
edit(from_dict[key], to_dict[key], create)
elif isinstance(value, list):
to_dict[key].extend(value)
else:
to_dict[key] = value
def edit_side_effect(url, json, create):
steps = url.lstrip('https://').lstrip('http://').rstrip('/')
parts = steps.split('/')
# Ensure there are enough parts to unpack
if len(parts) < 4:
raise ValueError(f"Expected at least 4 parts in the URL, got {len(parts)}: {steps}")
ip_port, _, _, header, *steps = parts
module, root = header.split(':')
assert(module == 'etsi-qkd-sdn-node')
assert(root == 'qkd_node')
if not steps:
edit(json, nodes[ip_port]['node'], create)
return
endpoint, *steps = steps
if not steps:
edit(json[endpoint], nodes[ip_port][endpoint], create)
return
raise Exception('Url too long')
@app.get('/', defaults={'path': ''})
@app.get("/<string:path>")
@app.get('/<path:path>')
def get(path):
try:
msg, msg_validate = get_side_effect(request.base_url)
print(msg_validate)
yang_validator.parse_to_dict(msg_validate)
return msg
except ValueError as e:
return {'error': str(e)}, 400
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
@app.post('/', defaults={'path': ''})
@app.post("/<string:path>")
@app.post('/<path:path>')
def post(path):
success = True
reason = ''
try:
edit_side_effect(request.base_url, request.json, True)
except Exception as e:
reason = str(e)
success = False
return {'success': success, 'reason': reason}
@app.route('/', defaults={'path': ''}, methods=['PUT', 'PATCH'])
@app.route("/<string:path>", methods=['PUT', 'PATCH'])
@app.route('/<path:path>', methods=['PUT', 'PATCH'])
def patch(path):
success = True
reason = ''
try:
edit_side_effect(request.base_url, request.json, False)
except Exception as e:
reason = str(e)
success = False
return {'success': success, 'reason': reason}
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
# import json
# from mock import requests
# import pyangbind.lib.pybindJSON as enc
# from pyangbind.lib.serialise import pybindJSONDecoder as dec
# from yang.sbi.qkd.templates.etsi_qkd_sdn_node import etsi_qkd_sdn_node
# module = etsi_qkd_sdn_node()
# url = 'https://1.1.1.1/restconf/data/etsi-qkd-sdn-node:'
# # Get node all info
# z = requests.get(url).json()
# var = dec.load_json(z, None, None, obj=module)
# print(enc.dumps(var))
# Reset module variable because it is already filled
# module = etsi_qkd_sdn_node()
# # Get node basic info
# node = module.qkd_node
# z = requests.get(url + 'qkd_node').json()
# var = dec.load_json(z, None, None, obj=node)
# print(enc.dumps(var))
# # Get all apps
# apps = node.qkd_applications
# z = requests.get(url + 'qkd_node/qkd_applications').json()
# var = dec.load_json(z, None, None, obj=apps)
# print(enc.dumps(var))
# # Edit app 0
# app = apps.qkd_app['00000000-0001-0000-0000-000000000000']
# app.client_app_id = 'id_0'
# requests.put(url + 'qkd_node/qkd_applications/qkd_app=00000000-0001-0000-0000-000000000000', json=json.loads(enc.dumps(app)))
# # Create app 1
# app = apps.qkd_app.add('00000000-0001-0000-0000-000000000001')
# requests.post(url + 'qkd_node/qkd_applications/qkd_app=00000000-0001-0000-0000-000000000001', json=json.loads(enc.dumps(app)))
# # Get all apps
# apps = node.qkd_applications
# z = requests.get(url + 'qkd_node/qkd_applications').json()
# var = dec.load_json(z, None, None, obj=apps)
# print(enc.dumps(var))