blob: 3b2deba2b003eb723e4f0cb9a5b4569de4ef72db [file] [log] [blame]
#!/usr/bin/env python
#
# Copyright 2017 the original author or authors.
#
# Code adapted from https://github.com/choppsv1/netconf
#
# 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 structlog
from lxml import etree
import netconf.nc_common.error as ncerror
log = structlog.get_logger()
class RpcResponse():
def __init__(self, capabilities):
self.is_error = False
# if there is an error then the reply_node will contains an Error
# object
self.reply_node = None
self.close_session = False
self.capabilities = capabilities
self.custom_rpc = False
def build_xml_response(self, request, voltha_response, custom_rpc=False):
if request is None:
return
voltha_xml_string = etree.tostring(voltha_response)
# Remove the leading and trailing <yang> tags
if voltha_xml_string.startswith('<yang>'):
voltha_xml_string = voltha_xml_string[len('<yang>'):]
if voltha_xml_string.endswith('</yang>'):
voltha_xml_string = voltha_xml_string[:-len('</yang>')]
# Empty response
elif voltha_xml_string.startswith('<yang/>'):
voltha_xml_string = ''
# Replace any True/False data to true/false
voltha_xml_string = voltha_xml_string.replace('>False<', '>false<')
voltha_xml_string = voltha_xml_string.replace('>True<', '>true<')
if not custom_rpc:
# Create the xml body as
if request.has_key('subclass'):
body = ''.join([
'<data>',
'<',
request['class'],
' xmlns="',
request['namespace'],
'">',
'<',
request['subclass'],
'>',
voltha_xml_string,
'</',
request['subclass'],
'>',
'</',
request['class'],
'>',
'</data>'
])
else:
body = ''.join([
'<data>',
'<',
request['class'],
' xmlns="urn:opencord:params:xml:ns:voltha:ietf-voltha">',
voltha_xml_string,
'</',
request['class'],
'>',
'</data>'
])
else: # custom_rpc
body = ''.join([
'<rpc-reply',
' xmlns="urn:opencord:params:xml:ns:voltha:ietf-voltha">',
voltha_xml_string,
'</rpc-reply>',
])
return etree.fromstring(body)
def add_node(self, new_node, tree):
if new_node.tag == 'ignore':
# We want only sub-elements
for elem in list(new_node):
tree.append(elem)
else:
tree.append(new_node)
def copy_basic_element(self, elm):
new_elem = etree.Element(elm.tag)
new_elem.text = elm.text
return new_elem
def process_inline_option(self, elem):
inline_option = False
inline_node_name = None
for elm in list(elem):
if elm.tag == 'yang_field_option':
inline_option = True
if elm.tag == 'name':
inline_node_name = elm.text
if not inline_option:
new_elem = etree.Element(elem.tag)
return new_elem, elem
# look for the node with the inline_node_name
for elm in list(elem):
if elm.tag == inline_node_name:
new_elem = etree.Element('ignore')
return new_elem, elm
def process_element(self, elem):
attrib = elem.get('type')
if (attrib == 'list'):
if list(elem) is None:
return self.copy_basic_element(elem)
# if elem.tag == 'items':
# new_elem = etree.Element('items')
new_elem = etree.Element('ignore')
# else:
# new_elem = etree.Element('ignore')
for elm in list(elem):
elm.tag = elem.tag
if elm.get('type') in ['list', 'dict']:
self.add_node(self.process_element(elm), new_elem)
else:
new_elem.append(self.copy_basic_element(elm))
return new_elem
elif (attrib == 'dict'):
# Empty case
if list(elem) is None:
return self.copy_basic_element(elem)
# Process field option.
new_elem, elem = self.process_inline_option(elem)
for elm in list(elem):
if elm.get('type') in ['list', 'dict']:
self.add_node(self.process_element(elm), new_elem)
else:
new_elem.append(self.copy_basic_element(elm))
return new_elem
else:
return self.copy_basic_element(elem)
def to_yang_xml(self, from_xml, request, yang_options=None,
custom_rpc=False):
# Parse from_xml as follows:
# 1. Any element having a list attribute shoud have each item move 1 level
# up and retag using the parent tag
# 2. Any element having a dict attribute and has a <yang_field_option>
# sub-element should have all it's items move to teh parent level
top = etree.Element('yang')
elms = list(from_xml)
xml_tag = yang_options[0]
list_items_name = yang_options[1]
# special case the xml contain a list type
if len(elms) == 1:
item = elms[0]
if item.get('type') == 'list':
if list_items_name == 'items':
# Create a new parent element
new_elem = etree.Element(xml_tag)
self.add_node(self.process_element(item), new_elem)
top.append(new_elem)
else:
if xml_tag and custom_rpc:
item.tag = xml_tag
elif request.has_key('subclass'):
item.tag = request['subclass']
# remove the subclass element in request to avoid duplicate tag
del request['subclass']
else:
item.tag = 'ignore'
self.add_node(self.process_element(item), top)
return top
# Process normally for all other cases
for elm in elms:
self.add_node(self.process_element(elm), top)
return top
# Helper method to sort the xml message based on the xml tags
def sort_xml_response(self, xml):
for parent in xml.xpath('//*[./*]'): # Search for parent elements
parent[:] = sorted(parent, key=lambda x: x.tag)
return xml
# custom_rpc refers to custom RPCs different from Netconf default RPCs
# like get, get-config, edit-config, etc
def build_yang_response(self, root, request, yang_options=None,
custom_rpc=False):
try:
self.custom_rpc = custom_rpc
yang_xml = self.to_yang_xml(root, request, yang_options,
custom_rpc)
log.info('yang-xml', yang_xml=etree.tounicode(yang_xml,
pretty_print=True))
return self.build_xml_response(request, yang_xml, custom_rpc)
except Exception as e:
log.exception('error-building-yang-response', request=request,
xml=etree.tostring(root))
self.rpc_response.is_error = True
self.rpc_response.node = ncerror.BadMsg(request)
return