Merge branch 'feat/191-cttc-implement-telemetry-backend-collector-emulated'... (7dbbfc03) · Commits · TFS / controller

src/telemetry/backend/drivers/core/_Driver.py→src/telemetry/backend/driver_api/_Driver.py

+0 −0

File moved.

View file

src/telemetry/backend/drivers/core/init.py→src/telemetry/backend/driver_api/init.py

+0 −0

File moved.

View file

src/telemetry/backend/drivers/emulated/EmulatedDriver.py

+257 −368

File changed.

Preview size limit exceeded, changes collapsed.

src/telemetry/backend/drivers/emulated/EmulatedHelper.py

0 → 100644

+166 −0

Original line number	Diff line number	Diff line
		# Copyright 2022-2024 ETSI OSG/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 anytree import Node
		import json
		from typing import Any, List


		class EmulatedDriverHelper:
		"""
		Helper class for the emulated driver.
		"""
		def __init__(self):
		pass

		def validate_resource_key(self, key: str) -> str:
		"""
		Splits the input string into two parts:
		- The first part is '_connect/settings/endpoints/'.
		- The second part is the remaining string after the first part, with '/' replaced by '_'.

		Args:
		key (str): The input string to process.

		Returns:
		str: A single string with the processed result.
		"""
		prefix = '_connect/settings/endpoints/'
		if not key.startswith(prefix):
		raise ValueError(f"The input path '{key}' does not start with the expected prefix: {prefix}")
		second_part = key[len(prefix):]
		second_part_processed = second_part.replace('/', '_')
		validated_key = prefix + second_part_processed
		return validated_key

		#--------------------------------------------------------------------------------------
		# ------- Below function is kept for debugging purposes (test-cases) only -------------
		#--------------------------------------------------------------------------------------

		# This below methods can be commented but are called by the SetConfig method in EmulatedDriver.py

		def _find_or_create_node(self, name: str, parent: Node) -> Node:
		"""
		Finds or creates a node with the given name under the specified parent.

		Args:
		name (str): The name of the node to find or create.
		parent (Node): The parent node.

		Returns:
		Node: The found or created node.
		"""
		node = next((child for child in parent.children if child.name == name), None)
		if not node:
		node = Node(name, parent=parent)
		return node


		def _create_or_update_node(self, name: str, parent: Node, value: Any):
		"""
		Creates or updates a node with the given name and value under the specified parent.

		Args:
		name (str): The name of the node.
		parent (Node): The parent node.
		value (Any): The value to set on the node.
		"""
		node = next((child for child in parent.children if child.name == name), None)
		if node:
		node.value = json.dumps(value)
		else:
		Node(name, parent=parent, value=json.dumps(value))


		def _parse_resource_key(self, resource_key: str) -> List[str]:
		"""
		Parses the resource key into parts, correctly handling brackets.

		Args:
		resource_key (str): The resource key to parse.

		Returns:
		List[str]: A list of parts from the resource key.
		"""
		resource_path = []
		current_part = ""
		in_brackets = False

		if not resource_key.startswith('/interface'):
		for char in resource_key.strip('/'):
		if char == '[':
		in_brackets = True
		current_part += char
		elif char == ']':
		in_brackets = False
		current_part += char
		elif char == '/' and not in_brackets:
		resource_path.append(current_part)
		current_part = ""
		else:
		current_part += char
		if current_part:
		resource_path.append(current_part)
		return resource_path
		else:
		resource_path = resource_key.strip('/').split('/', 1)
		if resource_path[1] == 'settings':
		return resource_path
		else:
		resource_path = [resource_key.strip('/').split('[')[0].strip('/'), resource_key.strip('/').split('[')[1].split(']')[0].replace('/', '_')]
		return resource_path


		#-----------------------------------
		# ------- EXTRA Methods ------------
		#-----------------------------------

		# def _generate_subtree(self, node: Node) -> dict:
		# """
		# Generates a subtree of the configuration tree starting from the specified node.

		# Args:
		# node (Node): The node from which to generate the subtree.

		# Returns:
		# dict: The subtree as a dictionary.
		# """
		# subtree = {}
		# for child in node.children:
		# if child.children:
		# subtree[child.name] = self._generate_subtree(child)
		# else:
		# value = getattr(child, "value", None)
		# subtree[child.name] = json.loads(value) if value else None
		# return subtree


		# def _find_or_raise_node(self, name: str, parent: Node) -> Node:
		# """
		# Finds a node with the given name under the specified parent or raises an exception if not found.

		# Args:
		# name (str): The name of the node to find.
		# parent (Node): The parent node.

		# Returns:
		# Node: The found node.

		# Raises:
		# ValueError: If the node is not found.
		# """
		# node = next((child for child in parent.children if child.name == name), None)
		# if not node:
		# raise ValueError(f"Node '{name}' not found under parent '{parent.name}'.")
		# return node

src/telemetry/backend/tests/test_emulated.py

+24 −23

Original line number	Diff line number	Diff line
		@@ -52,38 +52,39 @@ def test_disconnect(setup_driver):
		assert driver.Disconnect() is True
		assert driver.connected is False

		def test_set_config(setup_driver):
		logger.info(">>> test_set_config <<<")
		driver = setup_driver
		driver.Connect()
		# def test_set_config(setup_driver):
		# logger.info(">>> test_set_config <<<")
		# driver = setup_driver
		# driver.Connect()

		config = create_test_configuration()
		# config = create_test_configuration()

		results = driver.SetConfig(config)
		assert all(result is True for result in results)
		# results = driver.SetConfig(config)
		# assert all(result is True for result in results)

		def test_get_config(connected_configured_driver):
		logger.info(">>> test_get_config <<<")
		resource_keys = create_specific_config_keys()
		results = connected_configured_driver.GetConfig(resource_keys)
		# def test_get_config(connected_configured_driver):
		# logger.info(">>> test_get_config <<<")
		# resource_keys = create_specific_config_keys()
		# results = connected_configured_driver.GetConfig(resource_keys)

		for key, value in results:
		assert key in create_specific_config_keys()
		assert value is not None
		# for key, value in results:
		# assert key in create_specific_config_keys()
		# assert value is not None

		def test_delete_config(connected_configured_driver):
		logger.info(">>> test_delete_config <<<")
		resource_keys = create_config_for_delete()
		# def test_delete_config(connected_configured_driver):
		# logger.info(">>> test_delete_config <<<")
		# resource_keys = create_config_for_delete()

		results = connected_configured_driver.DeleteConfig(resource_keys)
		assert all(result is True for result in results)
		# results = connected_configured_driver.DeleteConfig(resource_keys)
		# assert all(result is True for result in results)

		def test_subscribe_state(connected_configured_driver):
		logger.info(">>> test_subscribe_state <<<")
		subscriptions = create_test_subscriptions()

		results = connected_configured_driver.SubscribeState(subscriptions)
		assert all(result is True for result in results)
		# logger.info(f"Subscribed result: {results}.")
		assert results == [False, True, True] # all(result is True for result in results)

		def test_unsubscribe_state(connected_configured_driver):
		logger.info(">>> test_unsubscribe_state <<<")
		@@ -91,15 +92,15 @@ def test_unsubscribe_state(connected_configured_driver):

		connected_configured_driver.SubscribeState(subscriptions)
		results = connected_configured_driver.UnsubscribeState(subscriptions)
		assert all(result is True for result in results)
		assert results == [False, True, True] # all(result is True for result in results)

		def test_get_state(connected_configured_driver):
		logger.info(">>> test_get_state <<<")
		subscriptions = create_test_subscriptions()

		connected_configured_driver.SubscribeState(subscriptions)
		logger.info(f"Subscribed to state: {subscriptions}. waiting for 3 seconds ...")
		time.sleep(3)
		logger.info(f"Subscribed to state: {subscriptions}. waiting for 12 seconds ...")
		time.sleep(12)

		state_iterator = connected_configured_driver.GetState(blocking=False)
		states = list(state_iterator)