Tests - Tools - Mock NCE-T Controller (74384a73) · Commits · TFS / controller

src/tests/tools/mock_nce_t_ctrl/nce_t_ctrl/YangHandler.py

+107 −16

Original line number	Diff line number	Diff line
		@@ -14,23 +14,22 @@


		import json, libyang, logging
		from typing import Dict, List, Optional, Type
		from typing import Dict, List, Optional, Set


		LOGGER = logging.getLogger(__name__)


		def walk_schema(node : libyang.SNode, path : str = '') -> Dict[str, Type]:
		schema_paths : Dict[str, Type] = dict()
		def walk_schema(node : libyang.SNode, path : str = '') -> Set[str]:
		current_path = f'{path}/{node.name()}'
		schema_paths[current_path] = type(node)
		schema_paths : Set[str] = {current_path}
		for child in node.children():
		if isinstance(child, (libyang.SLeaf, libyang.SLeafList)): continue
		schema_paths.update(walk_schema(child, current_path))
		return schema_paths

		def extract_schema_paths(yang_module : libyang.Module) -> Dict[str, Type]:
		schema_paths : Dict[str, Type] = dict()
		def extract_schema_paths(yang_module : libyang.Module) -> Set[str]:
		schema_paths : Set[str] = set()
		for node in yang_module.children():
		schema_paths.update(walk_schema(node))
		return schema_paths
		@@ -41,14 +40,14 @@ class YangHandler:
		yang_startup_data : Dict
		) -> None:
		self._yang_context = libyang.Context(yang_search_path)
		self._loaded_modules = set()
		self._yang_module_paths : Dict[str, Type] = dict()
		self._loaded_modules : Set[str] = set()
		self._schema_paths : Set[str] = set()
		for yang_module_name in yang_module_names:
		LOGGER.info('Loading module: {:s}'.format(str(yang_module_name)))
		yang_module = self._yang_context.load_module(yang_module_name)
		yang_module.feature_enable_all()
		self._loaded_modules.add(yang_module_name)
		self._yang_module_paths.update(extract_schema_paths(yang_module))
		self._schema_paths.update(extract_schema_paths(yang_module))

		self._datastore = self._yang_context.parse_data_mem(
		json.dumps(yang_startup_data), fmt='json'
		@@ -57,11 +56,11 @@ class YangHandler:
		def destroy(self) -> None:
		self._yang_context.destroy()

		def get_module_paths(self) -> Dict[str, Type]:
		return self._yang_module_paths
		def get_schema_paths(self) -> Set[str]:
		return self._schema_paths

		def get(self, path : str) -> Optional[str]:
		if not path.startswith('/'): path = '/' + path
		path = self._normalize_path(path)
		data = self._datastore.find_path(path)
		if data is None: return None
		json_data = data.print_mem(
		@@ -71,7 +70,7 @@ class YangHandler:
		return json_data

		def get_xpath(self, xpath : str) -> List[str]:
		if not path.startswith('/'): path = '/' + path
		if not xpath.startswith('/'): xpath = '/' + xpath
		nodes = self._datastore.find_all(xpath)
		result = list()
		for node in nodes:
		@@ -82,7 +81,7 @@ class YangHandler:
		return result

		def create(self, path : str, payload : Dict) -> str:
		if not path.startswith('/'): path = '/' + path
		path = self._normalize_path(path)
		# TODO: client should not provide identifier of element to be created, add it to subpath
		dnode_parsed : Optional[libyang.DNode] = self._yang_context.parse_data_mem(
		json.dumps(payload), 'json', strict=True, parse_only=False,
		@@ -103,7 +102,7 @@ class YangHandler:
		return json_data

		def update(self, path : str, payload : Dict) -> str:
		if not path.startswith('/'): path = '/' + path
		path = self._normalize_path(path)
		# NOTE: client should provide identifier of element to be updated
		dnode_parsed : Optional[libyang.DNode] = self._yang_context.parse_data_mem(
		json.dumps(payload), 'json', strict=True, parse_only=False,
		@@ -124,7 +123,7 @@ class YangHandler:
		return json_data

		def delete(self, path : str) -> Optional[str]:
		if not path.startswith('/'): path = '/' + path
		path = self._normalize_path(path)

		# NOTE: client should provide identifier of element to be deleted

		@@ -142,3 +141,95 @@ class YangHandler:
		node.free()

		return json_data

		def _normalize_path(self, path : str) -> str:
		"""
		Normalize RESTCONF path segments using the standard `list=<keys>`
		syntax into the libyang bracketed predicate form expected by
		the datastore (e.g. `network="admin"` -> `network[network-id="admin"]`).

		This implementation looks up the schema node for the list and
		uses its key leaf names to build the proper predicates. If the
		schema information is unavailable, it falls back to using the
		list name as the key name.
		"""

		parts = [p for p in path.strip('/').split('/') if p != '']
		schema_path = ''
		out_parts: List[str] = []

		for part in parts:
		if '=' in part:
		# split into name and value (value may contain commas/quotes)
		name, val = part.split('=', 1)
		# keep original name (may include prefix) for output, but
		# use local name (without module prefix) to lookup schema
		local_name = name.split(':', 1)[1] if ':' in name else name
		schema_path = schema_path + '/' + local_name if schema_path else '/' + local_name
		LOGGER.info('[_normalize_path] schema_path={:s}'.format(str(schema_path)))
		schema_nodes = list(self._yang_context.find_path(schema_path))
		if len(schema_nodes) != 1:
		MSG = 'No/Multiple SchemaNodes({:s}) for SchemaPath({:s})'
		raise Exception(MSG.format(
		str([repr(sn) for sn in schema_nodes]), schema_path
		))
		schema_node = schema_nodes[0]
		LOGGER.info('[_normalize_path] schema_node={:s}'.format(str(repr(schema_node))))

		# parse values splitting on commas outside quotes
		values = []
		cur = ''
		in_quotes = False
		for ch in val:
		if ch == '"':
		in_quotes = not in_quotes
		cur += ch
		elif ch == ',' and not in_quotes:
		values.append(cur)
		cur = ''
		else:
		cur += ch
		if cur != '':
		values.append(cur)

		# determine key names from schema_node if possible
		key_names = None
		if isinstance(schema_node, libyang.SList):
		key_names = [k.name() for k in schema_node.keys()]
		LOGGER.info('[_normalize_path] [SList] key_names={:s}'.format(str(key_names)))
		#if isinstance(keys, (list, tuple)):
		# key_names = keys
		# LOGGER.info('[_normalize_path] key_names={:s}'.format(str(key_names)))
		#elif isinstance(keys, str):
		# key_names = [kn for kn in k.split() if kn]
		# LOGGER.info('[_normalize_path] 1 key_names={:s}'.format(str(key_names)))
		#else:
		# MSG = 'Unsupported keys format: {:s} / {:s}'
		# raise Exception(MSG.format(str(type(keys)), str(keys)))
		#elif hasattr(schema_node, 'key'):
		# LOGGER.info('[_normalize_path] has key')
		# k = schema_node.key()
		# LOGGER.info('[_normalize_path] k={:s}'.format(str(k)))
		# if isinstance(k, str):
		# key_names = [kn for kn in k.split() if kn]
		# LOGGER.info('[_normalize_path] 3 key_names={:s}'.format(str(key_names)))

		if not key_names:
		# fallback: use the local list name as the single key
		key_names = [local_name]

		LOGGER.info('[_normalize_path] 5 key_names={:s}'.format(str(key_names)))

		# build predicate(s)
		preds = []
		for idx, kn in enumerate(key_names):
		kv = values[idx] if idx < len(values) else values[0]
		preds.append(f'[{kn}="{kv}"]')

		out_parts.append(name + ''.join(preds))
		else:
		local_part = part.split(':', 1)[1] if ':' in part else part
		schema_path = schema_path + '/' + local_part if schema_path else '/' + local_part
		out_parts.append(part)

		return '/' + '/'.join(out_parts)

src/tests/tools/mock_nce_t_ctrl/nce_t_ctrl/app.py

+2 −2

Original line number	Diff line number	Diff line
		@@ -46,7 +46,7 @@ with open(STARTUP_FILE, mode='r', encoding='UTF-8') as fp:
		yang_handler = YangHandler(
		YANG_SEARCH_PATH, YANG_MODULE_NAMES, YANG_STARTUP_DATA
		)
		restconf_paths = yang_handler.get_module_paths()
		restconf_paths = yang_handler.get_schema_paths()

		app = Flask(__name__)
		app.config['SECRET_KEY'] = SECRET_KEY
		@@ -66,5 +66,5 @@ api.add_resource(
		)

		LOGGER.info('Available RESTCONF paths:')
		for restconf_path in restconf_paths:
		for restconf_path in sorted(restconf_paths):
		LOGGER.info('- {:s}'.format(str(restconf_path)))