Commit 4730fd4f authored by Lluis Gifre Renom's avatar Lluis Gifre Renom
Browse files

Device component - QKD Driver: 

- Removed useless QKD tests
- Corrected fetch of qkd node details
parent 33a983c2

src/device/.gitlab-ci.yml

+0 −1
Original line number Diff line number Diff line
@@ -122,7 +122,6 @@ unit_test_device: 
    - sleep 5

    - docker ps -a

    - docker logs $IMAGE_NAME

    - docker exec -i $IMAGE_NAME bash -c "coverage run --append -m pytest --log-level=INFO --verbose device/tests/qkd/unit/test_qkd_mock_connectivity.py"

    - docker exec -i $IMAGE_NAME bash -c "coverage run --append -m pytest --log-level=INFO --verbose device/tests/qkd/unit/test_qkd_compliance.py"

    - docker exec -i $IMAGE_NAME bash -c "coverage run --append -m pytest --log-level=INFO --verbose device/tests/qkd/unit/test_mock_qkd_node.py"

    - docker exec -i $IMAGE_NAME bash -c "coverage run --append -m pytest --log-level=INFO --verbose device/tests/qkd/unit/test_qkd_error_handling.py"

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

+7 −2
Original line number Diff line number Diff line
@@ -224,7 +224,12 @@ def fetch_node(url: str, resource_key: str, headers: Dict[str, str], auth: Optio 
    try:

        r = requests.get(url, timeout=timeout, verify=False, auth=auth, headers=headers)

        r.raise_for_status()

        result.append((resource_key, r.json().get('qkd_node', {})))

        data = r.json()

        data.pop('qkdn_capabilities', None)

        data.pop('qkd_applications', None)

        data.pop('qkd_interfaces', None)

        data.pop('qkd_links', None)

        result.append((resource_key, data))

    except requests.RequestException as e:

        LOGGER.error(f"Error fetching node from {url}: {e}")

        result.append((resource_key, e))

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

deleted100644 → 0
+0 −35
Original line number Diff line number Diff line 
# Copyright 2022-2025 ETSI 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 requests

from requests.exceptions import ConnectionError



MOCK_QKD_ADDRESS = '127.0.0.1'

MOCK_PORT = 11111





def test_mock_qkd_node_responses():

    response = requests.get(f'http://{MOCK_QKD_ADDRESS}:{MOCK_PORT}/restconf/data/etsi-qkd-sdn-node:qkd_node')

    assert response.status_code == 200

    data = response.json()

    assert 'qkd_node' in data



def test_mock_node_failure_scenarios():

    try:

        response = requests.get(f'http://{MOCK_QKD_ADDRESS}:12345/restconf/data/etsi-qkd-sdn-node:qkd_node')

    except ConnectionError as e:

        assert isinstance(e, ConnectionError)

    else:

        pytest.fail("ConnectionError not raised as expected")
 
 No newline at end of file

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

deleted100644 → 0
+0 −32
Original line number Diff line number Diff line 
# Copyright 2022-2025 ETSI 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.



# tests/unit/test_qkd_performance.py



import pytest, time

from device.service.drivers.qkd.QKDDriver2 import QKDDriver



MOCK_QKD_ADDRRESS = '127.0.0.1'

MOCK_PORT = 11111



def test_performance_under_load():

    driver = QKDDriver(address=MOCK_QKD_ADDRRESS, port=MOCK_PORT, username='user', password='pass')

    driver.Connect()

    

    start_time = time.time()

    for _ in range(1000):

        driver.GetConfig(['/qkd_interfaces/qkd_interface'])

    end_time = time.time()

    

    assert (end_time - start_time) < 60