test_qkd_configuration.py

# 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.

import pytest
import json
from requests.exceptions import HTTPError
from device.service.drivers.qkd.QKDDriver2 import QKDDriver
import requests
from device.service.drivers.qkd.Tools2 import (
    RESOURCE_INTERFACES, 
    RESOURCE_LINKS, 
    RESOURCE_ENDPOINTS, 
    RESOURCE_APPS, 
    RESOURCE_CAPABILITES, 
    RESOURCE_NODE
)

MOCK_QKD_ADDRRESS = '127.0.0.1'
MOCK_PORT = 11111

@pytest.fixture
def qkd_driver():
    # Initialize the QKD driver with the appropriate settings, ensure correct JWT headers are included
    token = "YOUR_JWT_TOKEN"  # Replace with your actual JWT token
    if not token:
        pytest.fail("JWT token is missing. Make sure to generate a valid JWT token.")
    headers = {"Authorization": f"Bearer {token}"}
    return QKDDriver(address=MOCK_QKD_ADDRRESS, port=MOCK_PORT, headers=headers)

# Utility function to print the retrieved data for debugging
def print_data(label, data):
    print(f"{label}: {json.dumps(data, indent=2)}")

# Test ID: SBI_Test_03 (Initial Config Retrieval)
def test_initial_config_retrieval(qkd_driver):
    qkd_driver.Connect()
    
    # Retrieve and validate the initial configuration
    config = qkd_driver.GetInitialConfig()
    
    # Since GetInitialConfig returns a list, adjust the assertions accordingly
    assert isinstance(config, list), "Expected a list for initial config"
    assert len(config) > 0, "Initial config should not be empty"
    
    # Output for debugging
    print_data("Initial Config", config)

# Test ID: INT_LQ_Test_05 (QKD Devices Retrieval)
def test_retrieve_devices(qkd_driver):
    qkd_driver.Connect()
    
    # Retrieve and validate device information
    devices = qkd_driver.GetConfig([RESOURCE_NODE])
    assert isinstance(devices, list), "Expected a list of devices"
    
    if not devices:
        pytest.skip("No devices found in the system. Skipping device test.")
    
    for device in devices:
        assert isinstance(device, tuple), "Each device entry must be a tuple"
        assert isinstance(device[1], dict), "Device data must be a dictionary"
        if isinstance(device[1], Exception):
            pytest.fail(f"Error retrieving devices: {device[1]}")
    
    # Output for debugging
    print_data("Devices", devices)

# Test ID: INT_LQ_Test_04 (QKD Links Retrieval)
def test_retrieve_links(qkd_driver):
    qkd_driver.Connect()

    try:
        # Fetch the links using the correct resource key
        links = qkd_driver.GetConfig([RESOURCE_LINKS])
        assert isinstance(links, list), "Expected a list of tuples (resource key, data)."

        if len(links) == 0:
            pytest.skip("No links found in the system, skipping link validation.")

        for link in links:
            assert isinstance(link, tuple), "Each link entry must be a tuple"
            resource_key, link_data = link  # Unpack the tuple

            # Handle HTTPError or exception in the response
            if isinstance(link_data, requests.exceptions.HTTPError):
                pytest.fail(f"Failed to retrieve links due to HTTP error: {link_data}")
            
            if isinstance(link_data, dict):
                # For real QKD data (links as dictionaries)
                assert 'qkdl_id' in link_data, "Missing 'qkdl_id' in link data"
                assert 'qkdl_local' in link_data, "Missing 'qkdl_local' in link data"
                assert 'qkdl_remote' in link_data, "Missing 'qkdl_remote' in link data"
                assert 'qkdl_type' in link_data, "Missing 'qkdl_type' in link data"

                # Check 'virt_prev_hop' only for virtual links (VIRT)
                if link_data['qkdl_type'] == 'etsi-qkd-node-types:VIRT':
                    virt_prev_hop = link_data.get('virt_prev_hop')
                    assert virt_prev_hop is None or re.match(r'[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12}', str(virt_prev_hop)), \
                        f"Invalid 'virt_prev_hop': {virt_prev_hop}"

                # Print out the link details for debugging
                print(f"Link ID: {link_data['qkdl_id']}")
                print(f"Link Type: {link_data['qkdl_type']}")
                print(f"Local QKD: {json.dumps(link_data['qkdl_local'], indent=2)}")
                print(f"Remote QKD: {json.dumps(link_data['qkdl_remote'], indent=2)}")

            elif isinstance(link_data, list):
                # For mocked QKD data (links as lists of dictionaries)
                for mock_link in link_data:
                    assert 'uuid' in mock_link, "Missing 'uuid' in mocked link data"
                    assert 'src_qkdn_id' in mock_link, "Missing 'src_qkdn_id' in mocked link data"
                    assert 'dst_qkdn_id' in mock_link, "Missing 'dst_qkdn_id' in mocked link data"

                    # Print out the mocked link details for debugging
                    print(f"Mock Link ID: {mock_link['uuid']}")
                    print(f"Source QKD ID: {mock_link['src_qkdn_id']}")
                    print(f"Destination QKD ID: {mock_link['dst_qkdn_id']}")

            else:
                pytest.fail(f"Unexpected link data format: {type(link_data)}")

    except HTTPError as e:
        pytest.fail(f"HTTP error occurred while retrieving links: {e}")
    except Exception as e:
        pytest.fail(f"An unexpected error occurred: {e}")

# Test for QKD Services
def test_retrieve_services(qkd_driver):
    qkd_driver.Connect()
    services = qkd_driver.GetConfig([RESOURCE_ENDPOINTS])
    assert isinstance(services, list), "Expected a list of services"
    
    if not services:
        pytest.skip("No services found in the system. Skipping service test.")
    
    for service in services:
        assert isinstance(service, tuple), "Each service entry must be a tuple"
        assert isinstance(service[1], dict), "Service data must be a dictionary"
        if isinstance(service[1], Exception):
            pytest.fail(f"Error retrieving services: {service[1]}")
    
    print("Services:", json.dumps(services, indent=2))

# Test ID: INT_LQ_Test_07 (QKD Applications Retrieval)
def test_retrieve_applications(qkd_driver):
    qkd_driver.Connect()
    
    # Retrieve and validate applications information
    applications = qkd_driver.GetConfig([RESOURCE_APPS])  # Adjust to fetch applications using the correct key
    assert isinstance(applications, list), "Expected a list of applications"
    
    if not applications:
        pytest.skip("No applications found in the system. Skipping applications test.")
    
    for app in applications:
        assert isinstance(app, tuple), "Each application entry must be a tuple"
        assert isinstance(app[1], dict), "Application data must be a dictionary"
        if isinstance(app[1], Exception):
            pytest.fail(f"Error retrieving applications: {app[1]}")
    
    # Output for debugging
    print_data("Applications", applications)

# Test ID: INT_LQ_Test_03 (QKD Interfaces Retrieval)
def test_retrieve_interfaces(qkd_driver):
    qkd_driver.Connect()
    
    # Retrieve and validate interface information
    interfaces = qkd_driver.GetConfig([RESOURCE_INTERFACES])
    
    assert isinstance(interfaces, list), "Expected a list of interfaces"
    assert len(interfaces) > 0, "No interfaces found in the system"
    
    for interface in interfaces:
        assert isinstance(interface, tuple), "Each interface entry must be a tuple"
        assert isinstance(interface[1], dict), "Interface data must be a dictionary"
        if isinstance(interface[1], Exception):
            pytest.fail(f"Error retrieving interfaces: {interface[1]}")
    
    # Output for debugging
    print_data("Interfaces", interfaces)

# Test ID: INT_LQ_Test_02 (QKD Capabilities Retrieval)
def test_retrieve_capabilities(qkd_driver):
    qkd_driver.Connect()
    
    # Retrieve and validate capabilities information
    capabilities = qkd_driver.GetConfig([RESOURCE_CAPABILITES])
    
    assert isinstance(capabilities, list), "Expected a list of capabilities"
    assert len(capabilities) > 0, "No capabilities found in the system"
    
    for capability in capabilities:
        assert isinstance(capability, tuple), "Each capability entry must be a tuple"
        assert isinstance(capability[1], dict), "Capability data must be a dictionary"
        if isinstance(capability[1], Exception):
            pytest.fail(f"Error retrieving capabilities: {capability[1]}")
    
    # Output for debugging
    print_data("Capabilities", capabilities)

# Test ID: INT_LQ_Test_03 (QKD Endpoints Retrieval)
def test_retrieve_endpoints(qkd_driver):
    qkd_driver.Connect()
    
    # Retrieve and validate endpoint information
    endpoints = qkd_driver.GetConfig([RESOURCE_ENDPOINTS])
    
    assert isinstance(endpoints, list), "Expected a list of endpoints"
    assert len(endpoints) > 0, "No endpoints found in the system"
    
    for endpoint in endpoints:
        assert isinstance(endpoint, tuple), "Each endpoint entry must be a tuple"
        assert isinstance(endpoint[1], dict), "Endpoint data must be a dictionary"
        if isinstance(endpoint[1], Exception):
            pytest.fail(f"Error retrieving endpoints: {endpoint[1]}")
    
    # Output for debugging
    print_data("Endpoints", endpoints)