Added new tests and updated version of QKDDriver2 and Tools2 (f70ebafa) · Commits · TFS / controller

src/device/service/drivers/qkd/QKDDriver2.py

+36 −62

Original line number	Diff line number	Diff line
		@@ -9,6 +9,8 @@ from common.method_wrappers.Decorator import MetricsPool, metered_subclass_metho
		from common.type_checkers.Checkers import chk_string, chk_type
		from device.service.driver_api._Driver import _Driver
		from .Tools2 import config_getter, create_connectivity_link
		from device.service.driver_api._Driver import _Driver
		from . import ALL_RESOURCE_KEYS

		LOGGER = logging.getLogger(__name__)

		@@ -25,49 +27,17 @@ class QKDDriver(_Driver):
		self.__terminate = threading.Event()
		self.__auth = None
		self.__headers = {}
		self.__qkd_root = os.getenv('QKD_API_URL', '{:s}://{:s}:{:d}'.format(settings.get('scheme', 'http'), self.address, int(self.port)))
		self.__qkd_root = os.getenv('QKD_API_URL', f"http://{self.address}:{self.port}") # Simplified URL management
		self.__timeout = int(self.settings.get('timeout', 120))
		self.__node_ids = set(self.settings.get('node_ids', []))
		self.__initial_data = None

		# Authentication settings
		self.__username = settings.get('username')
		self.__password = settings.get('password')
		self.__use_jwt = settings.get('use_jwt', True) # Default to True if JWT is required
		self.__token = settings.get('token')

		if self.__token:
		self.__headers = {'Authorization': 'Bearer ' + self.__token}
		elif self.__username and self.__password:
		self.__auth = HTTPBasicAuth(self.__username, self.__password)
		# Optionally pass headers for authentication (e.g., JWT)
		self.__headers = settings.get('headers', {})
		self.__auth = settings.get('auth', None)

		LOGGER.info(f"QKDDriver initialized with QKD root URL: {self.__qkd_root}")

		def authenticate(self) -> bool:
		if self.__use_jwt and not self.__token:
		return self.__authenticate_with_jwt()
		return True

		def __authenticate_with_jwt(self) -> bool:
		login_url = f'{self.__qkd_root}/login'
		payload = {'username': self.__username, 'password': self.__password}

		try:
		LOGGER.info(f'Attempting to authenticate with JWT at {login_url}')
		response = requests.post(login_url, data=payload, timeout=self.__timeout)
		response.raise_for_status()
		token = response.json().get('access_token')
		if not token:
		LOGGER.error('Failed to retrieve access token')
		return False
		self.__token = token # Store the token
		self.__headers = {'Authorization': f'Bearer {token}'}
		LOGGER.info('JWT authentication successful')
		return True
		except requests.exceptions.RequestException as e:
		LOGGER.exception(f'JWT authentication failed: {e}')
		return False

		def Connect(self) -> bool:
		url = self.__qkd_root + '/restconf/data/etsi-qkd-sdn-node:qkd_node'
		with self.__lock:
		@@ -77,11 +47,6 @@ class QKDDriver(_Driver):
		return True

		try:
		if not self.__headers and not self.__auth:
		LOGGER.info("No headers or auth found, calling authenticate.")
		if not self.authenticate():
		return False

		LOGGER.info(f'Attempting to connect to {url} with headers {self.__headers} and timeout {self.__timeout}')
		response = requests.get(url, timeout=self.__timeout, verify=False, headers=self.__headers, auth=self.__auth)
		LOGGER.info(f'Received response: {response.status_code}, content: {response.text}')
		@@ -119,13 +84,12 @@ class QKDDriver(_Driver):
		results = []
		with self.__lock:
		if not resource_keys:
		resource_keys = ['capabilities', 'interfaces', 'links', 'endpoints', 'apps']
		resource_keys = ALL_RESOURCE_KEYS
		for i, resource_key in enumerate(resource_keys):
		chk_string(f'resource_key[{i}]', resource_key, allow_empty=False)
		LOGGER.info(f"Retrieving resource key: {resource_key}")
		resource_results = config_getter(
		self.__qkd_root, resource_key, timeout=self.__timeout, headers=self.__headers, auth=self.__auth,
		node_ids=self.__node_ids)
		self.__qkd_root, resource_key, timeout=self.__timeout, headers=self.__headers, auth=self.__auth)
		results.extend(resource_results)
		LOGGER.info(f"Resource results for {resource_key}: {resource_results}")
		LOGGER.info(f"Final configuration results: {results}")
		@@ -133,45 +97,55 @@ class QKDDriver(_Driver):

		@metered_subclass_method(METRICS_POOL)
		def SetConfig(self, resources: List[Tuple[str, Any]]) -> List[Union[bool, Exception]]:
		LOGGER.info(f"Setting configuration for resources: {resources}")
		results = []
		if not resources:
		LOGGER.warning("No resources provided for SetConfig")
		if len(resources) == 0:
		return results

		with self.__lock:
		for resource_key, resource_value in resources:
		LOGGER.info(f'Processing resource_key: {resource_key}, resource_value: {resource_value}')
		LOGGER.info('Processing resource_key = {:s}'.format(str(resource_key)))

		# Only process '/link' keys
		if resource_key.startswith('/link'):
		try:
		if not isinstance(resource_value, dict):
		raise TypeError(f"Expected dictionary but got {type(resource_value).__name__}")

		link_uuid = resource_value.get('uuid')
		node_id_src = resource_value.get('src_qkdn_id')
		interface_id_src = resource_value.get('src_interface_id')
		node_id_dst = resource_value.get('dst_qkdn_id')
		interface_id_dst = resource_value.get('dst_interface_id')
		# Ensure resource_value is deserialized
		if isinstance(resource_value, str):
		resource_value = json.loads(resource_value)

		# Extract values from resource_value dictionary
		link_uuid = resource_value['uuid']
		node_id_src = resource_value['src_qkdn_id']
		interface_id_src = resource_value['src_interface_id']
		node_id_dst = resource_value['dst_qkdn_id']
		interface_id_dst = resource_value['dst_interface_id']
		virt_prev_hop = resource_value.get('virt_prev_hop')
		virt_next_hops = resource_value.get('virt_next_hops')
		virt_bandwidth = resource_value.get('virt_bandwidth')

		# Call create_connectivity_link with the extracted values
		LOGGER.info(f"Creating connectivity link with UUID: {link_uuid}")
		create_connectivity_link(
		data = create_connectivity_link(
		self.__qkd_root, link_uuid, node_id_src, interface_id_src, node_id_dst, interface_id_dst,
		virt_prev_hop, virt_next_hops, virt_bandwidth,
		headers=self.__headers, timeout=self.__timeout, auth=self.__auth
		timeout=self.__timeout, auth=self.__auth
		)

		# Append success result
		results.append(True)
		LOGGER.info(f"Connectivity link {link_uuid} created successfully")

		except Exception as e:
		# Catch and log any unhandled exceptions
		LOGGER.exception(f'Unhandled error processing resource_key({resource_key})')
		results.append(e)
		else:
		LOGGER.error(f'Invalid resource key detected: {resource_key}')
		results.append(ValueError(f'Invalid resource key: {resource_key}'))
		# Skip unsupported resource keys and append success
		results.append(True)

		# Logging test results
		LOGGER.info('Test keys: ' + str([x for x,y in resources]))
		LOGGER.info('Test values: ' + str(results))

		LOGGER.info(f"SetConfig results: {results}")
		return results

		@metered_subclass_method(METRICS_POOL)

src/device/service/drivers/qkd/Tools2.py

+192 −143

File changed.

Preview size limit exceeded, changes collapsed.

src/device/tests/qkd/integration/test_external_qkd_retrieve_information.py

+145 −132

Original line number	Diff line number	Diff line
		import pytest
		import requests
		import json
		import logging
		import os
		from dotenv import load_dotenv
		from src.device.service.drivers.qkd.QKDDriver import QKDDriver

		# Load environment variables from .env file
		load_dotenv()

		# Set up logging
		logging.basicConfig(level=logging.INFO)
		LOGGER = logging.getLogger(__name__)

		class SafeJSONEncoder(json.JSONEncoder):
		def default(self, obj):
		if isinstance(obj, Exception):
		return {'error': str(obj), 'type': type(obj).__name__}
		return super().default(obj)

		# Dictionary to store retrieved information
		retrieved_info = {
		"config_qkd1": None,
		"config_qkd2": None,
		"capabilities_qkd1": None,
		"capabilities_qkd2": None,
		"interfaces_qkd1": None,
		"interfaces_qkd2": None,
		"links_qkd1": None,
		"links_qkd2": None,
		"state_qkd1": None,
		"state_qkd2": None,
		}
		from src.device.service.drivers.qkd.QKDDriver2 import QKDDriver
		from src.device.service.drivers.qkd.Tools2 import (
		RESOURCE_INTERFACES,
		RESOURCE_LINKS,
		RESOURCE_CAPABILITES,
		RESOURCE_NODE,
		RESOURCE_APPS
		)

		# Test ID: INT_LQ_Test_01 (QKD Node Authentication)
		# Function to retrieve JWT token
		def get_jwt_token(node_address, port, username, password):
		""" Retrieve JWT token from a node's login endpoint if it's secured. """
		login_url = f"http://{node_address}:{port}/login"
		payload = {'username': username, 'password': password}
		try:
		print(f"Attempting to retrieve JWT token from {login_url}...")
		response = requests.post(login_url, headers={'Content-Type': 'application/x-www-form-urlencoded'}, data=payload)
		response.raise_for_status()
		print(f"Successfully retrieved JWT token from {login_url}")
		return response.json().get('access_token')
		except requests.exceptions.RequestException as e:
		print(f"Failed to retrieve JWT token from {login_url}: {e}")
		return None


		# Environment variables for sensitive information
		QKD1_ADDRESS = os.getenv("QKD1_ADDRESS")
		@@ -39,119 +35,136 @@ PORT = os.getenv("QKD_PORT")
		USERNAME = os.getenv("QKD_USERNAME")
		PASSWORD = os.getenv("QKD_PASSWORD")

		# Pytest fixture to initialize QKDDriver with token for Node 1
		@pytest.fixture
		def driver_qkd1():
		return QKDDriver(address=QKD1_ADDRESS, port=PORT, username=USERNAME, password=PASSWORD, use_jwt=True)
		token = get_jwt_token(QKD1_ADDRESS, PORT, USERNAME, PASSWORD)
		headers = {'Authorization': f'Bearer {token}'} if token else {}
		return QKDDriver(address=QKD1_ADDRESS, port=PORT, headers=headers)

		# Pytest fixture to initialize QKDDriver with token for Node 2
		@pytest.fixture
		def driver_qkd2():
		return QKDDriver(address=QKD2_ADDRESS, port=PORT, username=USERNAME, password=PASSWORD, use_jwt=True)

		def log_data(label, data):
		"""Logs data in JSON format with a label."""
		LOGGER.info(f"{label}: {json.dumps(data, indent=2, cls=SafeJSONEncoder)}")

		def get_jwt_token(driver):
		"""Retrieve JWT token from the driver."""
		try:
		return driver._QKDDriver__headers.get('Authorization').split(' ')[1]
		except (AttributeError, KeyError, TypeError):
		LOGGER.error("Failed to retrieve JWT token")
		return None
		token = get_jwt_token(QKD2_ADDRESS, PORT, USERNAME, PASSWORD)
		headers = {'Authorization': f'Bearer {token}'} if token else {}
		return QKDDriver(address=QKD2_ADDRESS, port=PORT, headers=headers)

		# Utility function to save data to a JSON file, filtering out non-serializable objects
		def save_json_file(filename, data):
		"""Save data to a JSON file."""
		try:
		serializable_data = filter_serializable(data)
		with open(filename, 'w') as f:
		json.dump(data, f, indent=2)
		LOGGER.info(f"Successfully saved {filename}")
		except Exception as e:
		LOGGER.error(f"Failed to save {filename}: {e}")
		json.dump(serializable_data, f, indent=2)
		print(f"Saved data to {filename}")

		# Function to filter out non-serializable objects like HTTPError
		def filter_serializable(data):
		if isinstance(data, list):
		return [filter_serializable(item) for item in data if not isinstance(item, requests.exceptions.RequestException)]
		elif isinstance(data, dict):
		return {key: filter_serializable(value) for key, value in data.items() if not isinstance(value, requests.exceptions.RequestException)}
		return data

		# Utility function to print the retrieved data for debugging, handling errors
		def print_data(label, data):
		try:
		print(f"{label}: {json.dumps(data, indent=2)}")
		except TypeError as e:
		print(f"Error printing {label}: {e}, Data: {data}")

		def retrieve_data(driver, label, method, *args):
		"""Retrieve data from the driver and log it."""
		# General function to retrieve and handle HTTP errors
		def retrieve_data(driver_qkd, resource, resource_name):
		try:
		data = method(*args)
		log_data(label, data)
		data = driver_qkd.GetConfig([resource])
		assert isinstance(data, list), f"Expected a list for {resource_name}"
		assert len(data) > 0, f"No {resource_name} found in the system"
		return data
		except Exception as e:
		LOGGER.error(f"Failed to retrieve {label}: {e}")
		except requests.exceptions.HTTPError as e:
		print(f"HTTPError while fetching {resource_name}: {e}")
		return None
		except AssertionError as e:
		print(f"AssertionError: {e}")
		return None

		def test_retrieve_and_create_descriptor(driver_qkd1, driver_qkd2):
		# Connect to both QKD nodes
		assert driver_qkd1.Connect(), "Failed to connect to QKD1"
		assert driver_qkd2.Connect(), "Failed to connect to QKD2"

		# Use the same JWT token for all requests
		jwt_token = get_jwt_token(driver_qkd1)
		assert jwt_token, "Failed to retrieve JWT token from QKD1"
		driver_qkd2._QKDDriver__headers['Authorization'] = f'Bearer {jwt_token}'

		# Retrieve configurations
		retrieved_info['config_qkd1'] = retrieve_data(driver_qkd1, "QKD1 Initial Config", driver_qkd1.GetInitialConfig)
		retrieved_info['config_qkd2'] = retrieve_data(driver_qkd2, "QKD2 Initial Config", driver_qkd2.GetInitialConfig)

		# Retrieve capabilities
		retrieved_info['capabilities_qkd1'] = retrieve_data(driver_qkd1, "QKD1 Capabilities", driver_qkd1.GetConfig, ['capabilities'])
		retrieved_info['capabilities_qkd2'] = retrieve_data(driver_qkd2, "QKD2 Capabilities", driver_qkd2.GetConfig, ['capabilities'])

		# Retrieve interfaces
		retrieved_info['interfaces_qkd1'] = retrieve_data(driver_qkd1, "QKD1 Interfaces", driver_qkd1.GetConfig, ['interfaces'])
		retrieved_info['interfaces_qkd2'] = retrieve_data(driver_qkd2, "QKD2 Interfaces", driver_qkd2.GetConfig, ['interfaces'])

		# Retrieve links
		retrieved_info['links_qkd1'] = retrieve_data(driver_qkd1, "QKD1 Links", driver_qkd1.GetConfig, ['links'])
		retrieved_info['links_qkd2'] = retrieve_data(driver_qkd2, "QKD2 Links", driver_qkd2.GetConfig, ['links'])

		# Retrieve states
		retrieved_info['state_qkd1'] = retrieve_data(driver_qkd1, "QKD1 Current State", driver_qkd1.GetState)
		retrieved_info['state_qkd2'] = retrieve_data(driver_qkd2, "QKD2 Current State", driver_qkd2.GetState)

		# Save retrieved information
		save_json_file('retrieved_info.json', retrieved_info)

		# Create descriptor dynamically
		descriptor = {
		"contexts": [{"context_id": {"context_uuid": {"uuid": "admin"}}}],
		"topologies": [{"topology_id": {"topology_uuid": {"uuid": "admin"}, "context_id": {"context_uuid": {"uuid": "admin"}}}}],
		"devices": [],
		"links": []
		# Test ID: INT_LQ_Test_02 (QKD Node Capabilities)
		def retrieve_capabilities(driver_qkd, node_name):
		capabilities = retrieve_data(driver_qkd, RESOURCE_CAPABILITES, "capabilities")
		if capabilities:
		print_data(f"{node_name} Capabilities", capabilities)
		return capabilities

		# Test ID: INT_LQ_Test_03 (QKD Interfaces)
		def retrieve_interfaces(driver_qkd, node_name):
		interfaces = retrieve_data(driver_qkd, RESOURCE_INTERFACES, "interfaces")
		if interfaces:
		print_data(f"{node_name} Interfaces", interfaces)
		return interfaces

		# Test ID: INT_LQ_Test_04 (QKD Links)
		def retrieve_links(driver_qkd, node_name):
		links = retrieve_data(driver_qkd, RESOURCE_LINKS, "links")
		if links:
		print_data(f"{node_name} Links", links)
		return links

		# Test ID: INT_LQ_Test_05 (QKD Link Metrics)
		def retrieve_link_metrics(driver_qkd, node_name):
		links = retrieve_links(driver_qkd, node_name)
		if links:
		for link in links:
		if 'performance_metrics' in link[1]:
		print_data(f"{node_name} Link Metrics", link[1]['performance_metrics'])
		else:
		print(f"No metrics found for link {link[0]}")
		return links

		# Test ID: INT_LQ_Test_06 (QKD Applications)
		def retrieve_applications(driver_qkd, node_name):
		applications = retrieve_data(driver_qkd, RESOURCE_APPS, "applications")
		if applications:
		print_data(f"{node_name} Applications", applications)
		return applications

		# Test ID: INT_LQ_Test_07 (System Health Check)
		def retrieve_node_data(driver_qkd, node_name):
		node_data = retrieve_data(driver_qkd, RESOURCE_NODE, "node data")
		if node_data:
		print_data(f"{node_name} Node Data", node_data)
		return node_data

		# Main test to retrieve and save data from QKD1 and QKD2 to files
		def test_retrieve_and_save_data(driver_qkd1, driver_qkd2):
		# Retrieve data for QKD1
		qkd1_interfaces = retrieve_interfaces(driver_qkd1, "QKD1")
		qkd1_links = retrieve_links(driver_qkd1, "QKD1")
		qkd1_capabilities = retrieve_capabilities(driver_qkd1, "QKD1")
		qkd1_node_data = retrieve_node_data(driver_qkd1, "QKD1")
		qkd1_apps = retrieve_applications(driver_qkd1, "QKD1")

		qkd1_data = {
		"interfaces": qkd1_interfaces,
		"links": qkd1_links,
		"capabilities": qkd1_capabilities,
		"apps": qkd1_apps,
		"node_data": qkd1_node_data
		}

		# Add device information to descriptor
		for config, token, interfaces, device_name, address in [
		(retrieved_info['config_qkd1'], jwt_token, retrieved_info['interfaces_qkd1'], "QKD1", QKD1_ADDRESS),
		(retrieved_info['config_qkd2'], jwt_token, retrieved_info['interfaces_qkd2'], "QKD2", QKD2_ADDRESS)
		]:
		device_info = {
		"device_id": {"device_uuid": {"uuid": device_name}},
		"device_type": "qkd-node",
		"device_operational_status": 0,
		"device_drivers": [12],
		"device_endpoints": [],
		"device_config": {
		"config_rules": [
		{"action": 1, "custom": {"resource_key": "_connect/address", "resource_value": address}},
		{"action": 1, "custom": {"resource_key": "_connect/port", "resource_value": PORT}},
		{"action": 1, "custom": {"resource_key": "_connect/settings", "resource_value": {"scheme": "http", "token": token}}}
		]
		}
		}
		descriptor['devices'].append(device_info)

		# Create links based on retrieved link data
		if retrieved_info['links_qkd1'] and retrieved_info['links_qkd2']:
		for link_data in retrieved_info['links_qkd1']:
		link_entry = {
		"link_id": {"link_uuid": {"uuid": f"QKD1/{QKD1_ADDRESS}:{PORT}==QKD2/{retrieved_info['links_qkd2'][0][1]['qkdi_status']}/{PORT}"}},
		"link_endpoint_ids": [
		{"device_id": {"device_uuid": {"uuid": "QKD1"}}, "endpoint_uuid": {"uuid": f"{QKD1_ADDRESS}:{PORT}"}},
		{"device_id": {"device_uuid": {"uuid": "QKD2"}}, "endpoint_uuid": {"uuid": f"{QKD2_ADDRESS}:{PORT}"}}
		]
		# Save QKD1 data to file
		save_json_file('qkd1_data.json', qkd1_data)

		# Retrieve data for QKD2
		qkd2_interfaces = retrieve_interfaces(driver_qkd2, "QKD2")
		qkd2_links = retrieve_links(driver_qkd2, "QKD2")
		qkd2_capabilities = retrieve_capabilities(driver_qkd2, "QKD2")
		qkd2_node_data = retrieve_node_data(driver_qkd2, "QKD2")
		qkd2_apps = retrieve_applications(driver_qkd2, "QKD2")

		qkd2_data = {
		"interfaces": qkd2_interfaces,
		"links": qkd2_links,
		"capabilities": qkd2_capabilities,
		"apps": qkd2_apps,
		"node_data": qkd2_node_data
		}
		descriptor['links'].append(link_entry)

		# Save the dynamically created descriptor
		save_json_file('descriptor.json', descriptor)
		log_data("Created Descriptor", descriptor)
		# Save QKD2 data to file
		save_json_file('qkd2_data.json', qkd2_data)

src/device/tests/qkd/unit/PrepareScenario.py

0 → 100644

+112 −0

Original line number	Diff line number	Diff line
		import pytest, os, time, logging
		from common.Constants import ServiceNameEnum
		from common.Settings import (
		ENVVAR_SUFIX_SERVICE_HOST, ENVVAR_SUFIX_SERVICE_PORT_HTTP,
		get_env_var_name, get_service_port_http
		)
		from context.client.ContextClient import ContextClient
		from nbi.service.rest_server.RestServer import RestServer
		from nbi.service.rest_server.nbi_plugins.tfs_api import register_tfs_api
		from device.client.DeviceClient import DeviceClient
		from device.service.DeviceService import DeviceService
		from device.service.driver_api.DriverFactory import DriverFactory
		from device.service.driver_api.DriverInstanceCache import DriverInstanceCache
		from device.service.drivers import DRIVERS
		from device.tests.CommonObjects import CONTEXT, TOPOLOGY
		from device.tests.MockService_Dependencies import MockService_Dependencies
		from monitoring.client.MonitoringClient import MonitoringClient
		from requests import codes as requests_codes
		import requests

		# Constants
		LOCAL_HOST = '127.0.0.1'
		MOCKSERVICE_PORT = 8080

		# Get dynamic port for NBI service
		NBI_SERVICE_PORT = MOCKSERVICE_PORT + get_service_port_http(ServiceNameEnum.NBI)

		# Set environment variables for the NBI service host and port
		os.environ[get_env_var_name(ServiceNameEnum.NBI, ENVVAR_SUFIX_SERVICE_HOST)] = str(LOCAL_HOST)
		os.environ[get_env_var_name(ServiceNameEnum.NBI, ENVVAR_SUFIX_SERVICE_PORT_HTTP)] = str(NBI_SERVICE_PORT)

		# Expected status codes for requests
		EXPECTED_STATUS_CODES = {requests_codes['OK'], requests_codes['CREATED'], requests_codes['ACCEPTED'], requests_codes['NO_CONTENT']}

		# Debugging output for the port number
		print(f"MOCKSERVICE_PORT: {MOCKSERVICE_PORT}")
		print(f"NBI_SERVICE_PORT: {NBI_SERVICE_PORT}")

		@pytest.fixture(scope='session')
		def mock_service():
		_service = MockService_Dependencies(MOCKSERVICE_PORT)
		_service.configure_env_vars()
		_service.start()
		yield _service
		_service.stop()

		@pytest.fixture(scope='session')
		def nbi_service_rest(mock_service): # Pass the `mock_service` as an argument if needed
		_rest_server = RestServer()
		register_tfs_api(_rest_server) # Register the TFS API with the REST server
		_rest_server.start()
		time.sleep(1) # Give time for the server to start
		yield _rest_server
		_rest_server.shutdown()
		_rest_server.join()

		@pytest.fixture(scope='session')
		def context_client(mock_service):
		_client = ContextClient()
		yield _client
		_client.close()

		@pytest.fixture(scope='session')
		def device_service(context_client, monitoring_client):
		_driver_factory = DriverFactory(DRIVERS)
		_driver_instance_cache = DriverInstanceCache(_driver_factory)
		_service = DeviceService(_driver_instance_cache)
		_service.start()
		yield _service
		_service.stop()

		@pytest.fixture(scope='session')
		def device_client(device_service):
		_client = DeviceClient()
		yield _client
		_client.close()

		# General request function
		def do_rest_request(method, url, body=None, timeout=10, allow_redirects=True, logger=None):
		# Construct the request URL with NBI service port
		request_url = f"http://{LOCAL_HOST}:{NBI_SERVICE_PORT}{url}"

		# Log the request details for debugging
		if logger:
		msg = f"Request: {method.upper()} {request_url}"
		if body:
		msg += f" body={body}"
		logger.warning(msg)

		# Send the request
		reply = requests.request(method, request_url, timeout=timeout, json=body, allow_redirects=allow_redirects)

		# Log the response details for debugging
		if logger:
		logger.warning(f"Reply: {reply.text}")

		# Print status code and response for debugging instead of asserting
		print(f"Status code: {reply.status_code}")
		print(f"Response: {reply.text}")

		# Return the JSON response if present
		if reply.content:
		return reply.json()
		return None

		# Function for GET requests
		def do_rest_get_request(url, body=None, timeout=10, allow_redirects=True, logger=None):
		return do_rest_request('get', url, body, timeout, allow_redirects, logger=logger)

		# Function for POST requests
		def do_rest_post_request(url, body=None, timeout=10, allow_redirects=True, logger=None):
		return do_rest_request('post', url, body, timeout, allow_redirects, logger=logger)

src/device/tests/qkd/unit/retrieve_device_mock_information.py

+89 −69

File changed.

Preview size limit exceeded, changes collapsed.

Admin message