From 745cb369e4d388787e49bf799673361fbd49e1bd Mon Sep 17 00:00:00 2001 From: Waleed Akbar Date: Wed, 5 Nov 2025 08:58:57 +0000 Subject: [PATCH 1/6] Refactor test scripts and remove obsolete files. --- src/pytest.ini => pytest.ini | 0 ...able.sh => run_tests_locally-pluggable.sh} | 8 +- src/pluggables/tests/test_pluggables.py | 123 +++++++----------- 3 files changed, 54 insertions(+), 77 deletions(-) rename src/pytest.ini => pytest.ini (100%) rename scripts/{run_tests_locally-service-pluggable.sh => run_tests_locally-pluggable.sh} (86%) diff --git a/src/pytest.ini b/pytest.ini similarity index 100% rename from src/pytest.ini rename to pytest.ini diff --git a/scripts/run_tests_locally-service-pluggable.sh b/scripts/run_tests_locally-pluggable.sh similarity index 86% rename from scripts/run_tests_locally-service-pluggable.sh rename to scripts/run_tests_locally-pluggable.sh index e749bd43c..9f36aedda 100755 --- a/scripts/run_tests_locally-service-pluggable.sh +++ b/scripts/run_tests_locally-pluggable.sh @@ -18,9 +18,9 @@ cd $PROJECTDIR/src RCFILE=$PROJECTDIR/coverage/.coveragerc # to run integration test: -m integration -python3 -m pytest --log-level=info --log-cli-level=info --verbose -m "not integration" \ - pluggables/tests/test_pluggables_with_SBI.py -# python3 -m pytest --log-level=info --log-cli-level=info --verbose \ -# pluggables/tests/test_Pluggables.py +# python3 -m pytest --log-level=info --log-cli-level=info --verbose -m "not integration" \ +# pluggables/tests/test_pluggables_with_SBI.py +python3 -m pytest --log-level=info --log-cli-level=info --verbose -m "not integration"\ + pluggables/tests/test_pluggables.py echo "Bye!" diff --git a/src/pluggables/tests/test_pluggables.py b/src/pluggables/tests/test_pluggables.py index 9bce8c92c..2e2b9ac7a 100644 --- a/src/pluggables/tests/test_pluggables.py +++ b/src/pluggables/tests/test_pluggables.py @@ -16,28 +16,29 @@ import grpc import os, pytest import logging -from typing import Union from common.proto.context_pb2 import Empty from common.Constants import ServiceNameEnum from common.Settings import ( ENVVAR_SUFIX_SERVICE_HOST, ENVVAR_SUFIX_SERVICE_PORT_GRPC, get_env_var_name, get_service_port_grpc) -from common.tests.MockServicerImpl_Context import MockServicerImpl_Context -from common.proto.context_pb2_grpc import add_ContextServiceServicer_to_server from common.proto.pluggables_pb2 import ( PluggableId, Pluggable, ListPluggablesResponse, View ) -from common.tools.service.GenericGrpcService import GenericGrpcService from pluggables.client.PluggablesClient import PluggablesClient -from pluggables.service.PluggablesService import PluggablesService from pluggables.tests.testmessages import ( create_pluggable_request, create_list_pluggables_request, create_get_pluggable_request, create_delete_pluggable_request, create_configure_pluggable_request, ) - - +from pluggables.tests.CommonObjects import ( + DEVICE_HUB_UUID, DEVICE_LEAF_UUID +) +from pluggables.tests.PreparePluggablesTestScenario import ( # pylint: disable=unused-import + # be careful, order of symbols is important here! + mock_context_service, context_client, device_service, device_client, + pluggables_service, pluggables_client, test_prepare_environment +) ########################### # Tests Setup ########################### @@ -50,45 +51,6 @@ os.environ[get_env_var_name(ServiceNameEnum.PLUGGABLES, ENVVAR_SUFIX_SERVICE_POR LOGGER = logging.getLogger(__name__) -class MockContextService(GenericGrpcService): - # Mock Service implementing Context to simplify unitary tests of DSCM pluggables - - def __init__(self, bind_port: Union[str, int]) -> None: - super().__init__(bind_port, LOCAL_HOST, enable_health_servicer=False, cls_name='MockService') - - # pylint: disable=attribute-defined-outside-init - def install_servicers(self): - self.context_servicer = MockServicerImpl_Context() - add_ContextServiceServicer_to_server(self.context_servicer, self.server) - -# This fixture will be requested by test cases and last during testing session -@pytest.fixture(scope='session') -def pluggables_service(): - LOGGER.info('Initializing PluggableService...') - _service = PluggablesService() - _service.start() - - # yield the server, when test finishes, execution will resume to stop it - LOGGER.info('Yielding PluggableService...') - yield _service - - LOGGER.info('Terminating PluggableService...') - _service.stop() - - LOGGER.info('Terminated PluggableService...') - -@pytest.fixture(scope='function') -def pluggable_client(pluggables_service : PluggablesService): - LOGGER.info('Creating PluggablesClient...') - _client = PluggablesClient() - - LOGGER.info('Yielding PluggablesClient...') - yield _client - - LOGGER.info('Closing PluggablesClient...') - _client.close() - - LOGGER.info('Closed PluggablesClient...') @pytest.fixture(autouse=True) def log_all_methods(request): @@ -100,15 +62,19 @@ def log_all_methods(request): yield LOGGER.info(f" <<<<< Finished test: {request.node.name} ") + + ########################### # Test Cases ########################### # CreatePluggable Test without configuration -def test_CreatePluggable(pluggable_client : PluggablesClient): +def test_CreatePluggable(pluggables_client : PluggablesClient): LOGGER.info('Creating Pluggable for test...') - _pluggable_request = create_pluggable_request(preferred_pluggable_index=-1) - _pluggable = pluggable_client.CreatePluggable(_pluggable_request) + _pluggable_request = create_pluggable_request( + device_uuid=DEVICE_HUB_UUID, + preferred_pluggable_index=-1) + _pluggable = pluggables_client.CreatePluggable(_pluggable_request) LOGGER.info('Created Pluggable for test: %s', _pluggable) assert isinstance(_pluggable, Pluggable) assert _pluggable.id.pluggable_index == _pluggable_request.preferred_pluggable_index @@ -116,16 +82,17 @@ def test_CreatePluggable(pluggable_client : PluggablesClient): # CreatePluggable Test with configuration -def test_CreatePluggable_with_config(pluggable_client : PluggablesClient): +@pytest.mark.integration +def test_CreatePluggable_with_config(pluggables_client : PluggablesClient): LOGGER.info('Creating Pluggable with initial configuration for test...') _pluggable_request = create_pluggable_request( - device_uuid = "9bbf1937-db9e-45bc-b2c6-3214a9d42157", - preferred_pluggable_index = -1, + device_uuid = DEVICE_HUB_UUID, + preferred_pluggable_index = 0, with_initial_config = True) - _pluggable = pluggable_client.CreatePluggable(_pluggable_request) + _pluggable = pluggables_client.CreatePluggable(_pluggable_request) LOGGER.info('Created Pluggable with initial configuration for test: %s', _pluggable) assert isinstance(_pluggable, Pluggable) - assert _pluggable.id.pluggable_index == _pluggable_request.preferred_pluggable_index + assert _pluggable.id.pluggable_index == 0 assert _pluggable.id.device.device_uuid.uuid == _pluggable_request.device.device_uuid.uuid assert _pluggable.config is not None assert len(_pluggable.config.dsc_groups) == 1 @@ -134,26 +101,29 @@ def test_CreatePluggable_with_config(pluggable_client : PluggablesClient): assert len(dsc_group.subcarriers) == 2 # create pluggable request with pluggable key already exists error -def test_CreatePluggable_already_exists(pluggable_client : PluggablesClient): +def test_CreatePluggable_already_exists(pluggables_client : PluggablesClient): LOGGER.info('Creating Pluggable for test...') - _pluggable_request = create_pluggable_request(preferred_pluggable_index=5) - _pluggable = pluggable_client.CreatePluggable(_pluggable_request) + _pluggable_request = create_pluggable_request( + device_uuid=DEVICE_LEAF_UUID, + preferred_pluggable_index=5) + _pluggable = pluggables_client.CreatePluggable(_pluggable_request) LOGGER.info('Created Pluggable for test: %s', _pluggable) assert isinstance(_pluggable, Pluggable) assert _pluggable.id.pluggable_index == _pluggable_request.preferred_pluggable_index assert _pluggable.id.device.device_uuid.uuid == _pluggable_request.device.device_uuid.uuid # Try to create the same pluggable again, should raise ALREADY_EXISTS with pytest.raises(grpc.RpcError) as e: - pluggable_client.CreatePluggable(_pluggable_request) + pluggables_client.CreatePluggable(_pluggable_request) assert e.value.code() == grpc.StatusCode.ALREADY_EXISTS # ListPluggables Test -def test_ListPluggables(pluggable_client : PluggablesClient): +def test_ListPluggables(pluggables_client : PluggablesClient): LOGGER.info('Listing Pluggables for test...') _list_request = create_list_pluggables_request( + device_uuid=DEVICE_HUB_UUID, view_level = View.VIEW_CONFIG # View.VIEW_STATE ) - _pluggables = pluggable_client.ListPluggables(_list_request) + _pluggables = pluggables_client.ListPluggables(_list_request) LOGGER.info('Listed Pluggables for test: %s', _pluggables) assert isinstance(_pluggables, ListPluggablesResponse) if len(_pluggables.pluggables) != 0: @@ -165,12 +135,14 @@ def test_ListPluggables(pluggable_client : PluggablesClient): assert len(_pluggables.pluggables) == 0 # GetPluggable Test -def test_GetPluggable(pluggable_client : PluggablesClient): +def test_GetPluggable(pluggables_client : PluggablesClient): LOGGER.info('Starting GetPluggable test...') LOGGER.info('Getting Pluggable for test...') # First create a pluggable to get it later - _pluggable_request = create_pluggable_request(preferred_pluggable_index=1) - _created_pluggable = pluggable_client.CreatePluggable(_pluggable_request) + _pluggable_request = create_pluggable_request( + device_uuid=DEVICE_HUB_UUID, + preferred_pluggable_index=1) + _created_pluggable = pluggables_client.CreatePluggable(_pluggable_request) LOGGER.info('Created Pluggable for GetPluggable test: %s', _created_pluggable) _get_request = create_get_pluggable_request( @@ -178,7 +150,7 @@ def test_GetPluggable(pluggable_client : PluggablesClient): pluggable_index = _created_pluggable.id.pluggable_index, view_level = View.VIEW_FULL ) - _pluggable = pluggable_client.GetPluggable(_get_request) + _pluggable = pluggables_client.GetPluggable(_get_request) LOGGER.info('Got Pluggable for test: %s', _pluggable) assert isinstance(_pluggable, Pluggable) assert _pluggable.id.pluggable_index == _created_pluggable.id.pluggable_index @@ -186,26 +158,28 @@ def test_GetPluggable(pluggable_client : PluggablesClient): # DeletePluggable Test -def test_DeletePluggable(pluggable_client : PluggablesClient): +def test_DeletePluggable(pluggables_client : PluggablesClient): LOGGER.info('Starting DeletePluggable test...') LOGGER.info('Creating Pluggable to delete for test...') # First create a pluggable to delete it later - _pluggable_request = create_pluggable_request(preferred_pluggable_index=2) - _created_pluggable = pluggable_client.CreatePluggable(_pluggable_request) + _pluggable_request = create_pluggable_request( + device_uuid=DEVICE_LEAF_UUID, + preferred_pluggable_index=2) + _created_pluggable = pluggables_client.CreatePluggable(_pluggable_request) LOGGER.info('Created Pluggable to delete for test: %s', _created_pluggable) _delete_request = create_delete_pluggable_request( device_uuid = _created_pluggable.id.device.device_uuid.uuid, pluggable_index = _created_pluggable.id.pluggable_index ) - _response = pluggable_client.DeletePluggable(_delete_request) + _response = pluggables_client.DeletePluggable(_delete_request) LOGGER.info('Deleted Pluggable for test, response: %s', _response) assert isinstance(_response, Empty) # Try to get the deleted pluggable, should raise NOT_FOUND with pytest.raises(grpc.RpcError) as e: - pluggable_client.GetPluggable( + pluggables_client.GetPluggable( create_get_pluggable_request( device_uuid = _created_pluggable.id.device.device_uuid.uuid, pluggable_index = _created_pluggable.id.pluggable_index @@ -214,20 +188,23 @@ def test_DeletePluggable(pluggable_client : PluggablesClient): assert e.value.code() == grpc.StatusCode.NOT_FOUND # ConfigurePluggable Test -def test_ConfigurePluggable(pluggable_client : PluggablesClient): +@pytest.mark.integration +def test_ConfigurePluggable(pluggables_client : PluggablesClient): LOGGER.info('Starting ConfigurePluggable test...') LOGGER.info('Creating Pluggable to configure for test...') # First create a pluggable to configure it later - _pluggable_request = create_pluggable_request(preferred_pluggable_index=3) - _created_pluggable = pluggable_client.CreatePluggable(_pluggable_request) + _pluggable_request = create_pluggable_request( + device_uuid=DEVICE_HUB_UUID, + preferred_pluggable_index=3) + _created_pluggable = pluggables_client.CreatePluggable(_pluggable_request) LOGGER.info('Created Pluggable to configure for test: %s', _created_pluggable) _configure_request = create_configure_pluggable_request( device_uuid = _created_pluggable.id.device.device_uuid.uuid, pluggable_index = _created_pluggable.id.pluggable_index, ) - _pluggable = pluggable_client.ConfigurePluggable(_configure_request) + _pluggable = pluggables_client.ConfigurePluggable(_configure_request) LOGGER.info('Configured Pluggable for test: %s', _pluggable) assert isinstance(_pluggable, Pluggable) assert _pluggable.config is not None -- GitLab From f7da5a8dd879ab3d7bac9459882b946835d51006 Mon Sep 17 00:00:00 2001 From: gifrerenom Date: Wed, 5 Nov 2025 14:39:54 +0000 Subject: [PATCH 2/6] NBI and Pluggables components: - Reactivated Pluggables-related CI/CD tests --- src/nbi/.gitlab-ci.yml | 2 +- src/pluggables/.gitlab-ci.yml | 146 +++++++++++++++++----------------- 2 files changed, 74 insertions(+), 74 deletions(-) diff --git a/src/nbi/.gitlab-ci.yml b/src/nbi/.gitlab-ci.yml index 492c4603a..14e2f35d6 100644 --- a/src/nbi/.gitlab-ci.yml +++ b/src/nbi/.gitlab-ci.yml @@ -120,7 +120,7 @@ unit_test nbi: - docker exec -i $IMAGE_NAME bash -c "coverage run --append -m pytest --log-level=INFO --verbose $IMAGE_NAME/tests/test_ietf_l3vpn.py --junitxml=/opt/results/${IMAGE_NAME}_report_ietf_l3vpn.xml" - docker exec -i $IMAGE_NAME bash -c "coverage run --append -m pytest --log-level=INFO --verbose $IMAGE_NAME/tests/test_etsi_bwm.py --junitxml=/opt/results/${IMAGE_NAME}_report_etsi_bwm.xml" - docker exec -i $IMAGE_NAME bash -c "coverage run --append -m pytest --log-level=INFO --verbose $IMAGE_NAME/tests/test_camara_qod.py --junitxml=/opt/results/${IMAGE_NAME}_report_camara_qod.xml" - #- docker exec -i $IMAGE_NAME bash -c "coverage run --append -m pytest --log-level=INFO --verbose $IMAGE_NAME/tests/test_dscm_restconf.py --junitxml=/opt/results/${IMAGE_NAME}_report_dscm_restconf.xml" + - docker exec -i $IMAGE_NAME bash -c "coverage run --append -m pytest --log-level=INFO --verbose $IMAGE_NAME/tests/test_dscm_restconf.py --junitxml=/opt/results/${IMAGE_NAME}_report_dscm_restconf.xml" - docker exec -i $IMAGE_NAME bash -c "coverage report --include='${IMAGE_NAME}/*' --show-missing" coverage: '/TOTAL\s+\d+\s+\d+\s+(\d+%)/' after_script: diff --git a/src/pluggables/.gitlab-ci.yml b/src/pluggables/.gitlab-ci.yml index 695348a03..b9e58b0e8 100644 --- a/src/pluggables/.gitlab-ci.yml +++ b/src/pluggables/.gitlab-ci.yml @@ -40,76 +40,76 @@ build pluggables: - .gitlab-ci.yml # Apply unit test to the component -#unit_test pluggables: -# variables: -# IMAGE_NAME: 'pluggables' # name of the microservice -# IMAGE_TAG: 'latest' # tag of the container image (production, development, etc) -# stage: unit_test -# needs: -# - build pluggables -# before_script: -# - docker login -u "$CI_REGISTRY_USER" -p "$CI_REGISTRY_PASSWORD" $CI_REGISTRY -# - if docker network list | grep teraflowbridge; then echo "teraflowbridge is already created"; else docker network create -d bridge teraflowbridge; fi -# - if docker container ls | grep crdb; then docker rm -f crdb; else echo "CockroachDB container is not in the system"; fi -# - if docker volume ls | grep crdb; then docker volume rm -f crdb; else echo "CockroachDB volume is not in the system"; fi -# - if docker container ls | grep context; then docker rm -f context; else echo "context container is not in the system"; fi -# - if docker container ls | grep $IMAGE_NAME; then docker rm -f $IMAGE_NAME; else echo "$IMAGE_NAME container is not in the system"; fi -# - docker container prune -f -# script: -# - docker pull "$CI_REGISTRY_IMAGE/$IMAGE_NAME:$IMAGE_TAG" -# - docker pull "$CI_REGISTRY_IMAGE/context:$IMAGE_TAG" -# - docker pull "cockroachdb/cockroach:latest-v22.2" -# - docker volume create crdb -# - > -# docker run --name crdb -d --network=teraflowbridge -p 26257:26257 -p 8080:8080 -# --env COCKROACH_DATABASE=tfs_test --env COCKROACH_USER=tfs --env COCKROACH_PASSWORD=tfs123 -# --volume "crdb:/cockroach/cockroach-data" -# cockroachdb/cockroach:latest-v22.2 start-single-node -# - echo "Waiting for initialization..." -# - while ! docker logs crdb 2>&1 | grep -q 'finished creating default user \"tfs\"'; do sleep 1; done -# - CRDB_ADDRESS=$(docker inspect crdb --format "{{.NetworkSettings.Networks.teraflowbridge.IPAddress}}") -# - echo $CRDB_ADDRESS -# - > -# docker run --name context -d -p 1010:1010 -# --env "CRDB_URI=cockroachdb://tfs:tfs123@${CRDB_ADDRESS}:26257/tfs_test?sslmode=require" -# --network=teraflowbridge -# $CI_REGISTRY_IMAGE/context:$IMAGE_TAG -# - docker ps -a -# - CONTEXT_ADDRESS=$(docker inspect context --format "{{.NetworkSettings.Networks.teraflowbridge.IPAddress}}") -# - echo $CONTEXT_ADDRESS -# - > -# docker run --name $IMAGE_NAME -d -p 30040:30040 -# --env "CONTEXTSERVICE_SERVICE_HOST=${CONTEXT_ADDRESS}" -# --env "CONTEXTSERVICE_SERVICE_PORT_GRPC=1010" -# --volume "$PWD/src/$IMAGE_NAME/tests:/opt/results" -# --network=teraflowbridge -# $CI_REGISTRY_IMAGE/$IMAGE_NAME:$IMAGE_TAG -# - docker ps -a -# - sleep 5 -# - docker logs $IMAGE_NAME -# - > -# docker exec -i $IMAGE_NAME bash -c -# "coverage run -m pytest --log-level=INFO --verbose --junitxml=/opt/results/${IMAGE_NAME}_report.xml $IMAGE_NAME/tests/test_*.py" -# - docker exec -i $IMAGE_NAME bash -c "coverage report --include='${IMAGE_NAME}/*' --show-missing" -# coverage: '/TOTAL\s+\d+\s+\d+\s+(\d+%)/' -# after_script: -# - docker rm -f $IMAGE_NAME context crdb -# - docker volume rm -f crdb -# - docker network rm teraflowbridge -# - docker volume prune --force -# - docker image prune --force -# rules: -# - if: '$CI_PIPELINE_SOURCE == "merge_request_event" && ($CI_MERGE_REQUEST_TARGET_BRANCH_NAME == "develop" || $CI_MERGE_REQUEST_TARGET_BRANCH_NAME == $CI_DEFAULT_BRANCH)' -# - if: '$CI_PIPELINE_SOURCE == "push" && $CI_COMMIT_BRANCH == "develop"' -# - changes: -# - src/common/**/*.py -# - proto/*.proto -# - src/$IMAGE_NAME/**/*.{py,in,yml} -# - src/$IMAGE_NAME/Dockerfile -# - src/$IMAGE_NAME/tests/*.py -# - manifests/${IMAGE_NAME}service.yaml -# - .gitlab-ci.yml -# artifacts: -# when: always -# reports: -# junit: src/$IMAGE_NAME/tests/${IMAGE_NAME}_report.xml +unit_test pluggables: + variables: + IMAGE_NAME: 'pluggables' # name of the microservice + IMAGE_TAG: 'latest' # tag of the container image (production, development, etc) + stage: unit_test + needs: + - build pluggables + before_script: + - docker login -u "$CI_REGISTRY_USER" -p "$CI_REGISTRY_PASSWORD" $CI_REGISTRY + - if docker network list | grep teraflowbridge; then echo "teraflowbridge is already created"; else docker network create -d bridge teraflowbridge; fi + - if docker container ls | grep crdb; then docker rm -f crdb; else echo "CockroachDB container is not in the system"; fi + - if docker volume ls | grep crdb; then docker volume rm -f crdb; else echo "CockroachDB volume is not in the system"; fi + - if docker container ls | grep context; then docker rm -f context; else echo "context container is not in the system"; fi + - if docker container ls | grep $IMAGE_NAME; then docker rm -f $IMAGE_NAME; else echo "$IMAGE_NAME container is not in the system"; fi + - docker container prune -f + script: + - docker pull "$CI_REGISTRY_IMAGE/$IMAGE_NAME:$IMAGE_TAG" + - docker pull "$CI_REGISTRY_IMAGE/context:$IMAGE_TAG" + - docker pull "cockroachdb/cockroach:latest-v22.2" + - docker volume create crdb + - > + docker run --name crdb -d --network=teraflowbridge -p 26257:26257 -p 8080:8080 + --env COCKROACH_DATABASE=tfs_test --env COCKROACH_USER=tfs --env COCKROACH_PASSWORD=tfs123 + --volume "crdb:/cockroach/cockroach-data" + cockroachdb/cockroach:latest-v22.2 start-single-node + - echo "Waiting for initialization..." + - while ! docker logs crdb 2>&1 | grep -q 'finished creating default user \"tfs\"'; do sleep 1; done + - CRDB_ADDRESS=$(docker inspect crdb --format "{{.NetworkSettings.Networks.teraflowbridge.IPAddress}}") + - echo $CRDB_ADDRESS + - > + docker run --name context -d -p 1010:1010 + --env "CRDB_URI=cockroachdb://tfs:tfs123@${CRDB_ADDRESS}:26257/tfs_test?sslmode=require" + --network=teraflowbridge + $CI_REGISTRY_IMAGE/context:$IMAGE_TAG + - docker ps -a + - CONTEXT_ADDRESS=$(docker inspect context --format "{{.NetworkSettings.Networks.teraflowbridge.IPAddress}}") + - echo $CONTEXT_ADDRESS + - > + docker run --name $IMAGE_NAME -d -p 30040:30040 + --env "CONTEXTSERVICE_SERVICE_HOST=${CONTEXT_ADDRESS}" + --env "CONTEXTSERVICE_SERVICE_PORT_GRPC=1010" + --volume "$PWD/src/$IMAGE_NAME/tests:/opt/results" + --network=teraflowbridge + $CI_REGISTRY_IMAGE/$IMAGE_NAME:$IMAGE_TAG + - docker ps -a + - sleep 5 + - docker logs $IMAGE_NAME + - > + docker exec -i $IMAGE_NAME bash -c + "coverage run -m pytest --log-level=INFO --verbose --junitxml=/opt/results/${IMAGE_NAME}_report.xml $IMAGE_NAME/tests/test_*.py" + - docker exec -i $IMAGE_NAME bash -c "coverage report --include='${IMAGE_NAME}/*' --show-missing" + coverage: '/TOTAL\s+\d+\s+\d+\s+(\d+%)/' + after_script: + - docker rm -f $IMAGE_NAME context crdb + - docker volume rm -f crdb + - docker network rm teraflowbridge + - docker volume prune --force + - docker image prune --force + rules: + - if: '$CI_PIPELINE_SOURCE == "merge_request_event" && ($CI_MERGE_REQUEST_TARGET_BRANCH_NAME == "develop" || $CI_MERGE_REQUEST_TARGET_BRANCH_NAME == $CI_DEFAULT_BRANCH)' + - if: '$CI_PIPELINE_SOURCE == "push" && $CI_COMMIT_BRANCH == "develop"' + - changes: + - src/common/**/*.py + - proto/*.proto + - src/$IMAGE_NAME/**/*.{py,in,yml} + - src/$IMAGE_NAME/Dockerfile + - src/$IMAGE_NAME/tests/*.py + - manifests/${IMAGE_NAME}service.yaml + - .gitlab-ci.yml + artifacts: + when: always + reports: + junit: src/$IMAGE_NAME/tests/${IMAGE_NAME}_report.xml -- GitLab From 9fb170cf001d83769db30998794aed3d6f008e83 Mon Sep 17 00:00:00 2001 From: Waleed Akbar Date: Tue, 11 Nov 2025 06:25:39 +0000 Subject: [PATCH 3/6] Refactor pluggable service and tests for improved readability and consistency --- scripts/run_tests_locally-pluggable.sh | 4 +- src/device/requirements.in | 2 +- .../service/PluggablesServiceServicerImpl.py | 77 +++++++++++-------- src/pluggables/service/config_translator.py | 9 +-- .../tests/test_pluggables_with_SBI.py | 46 +++++------ src/pluggables/tests/testmessages.py | 6 +- 6 files changed, 77 insertions(+), 67 deletions(-) diff --git a/scripts/run_tests_locally-pluggable.sh b/scripts/run_tests_locally-pluggable.sh index 9f36aedda..3c8e685f5 100755 --- a/scripts/run_tests_locally-pluggable.sh +++ b/scripts/run_tests_locally-pluggable.sh @@ -18,8 +18,8 @@ cd $PROJECTDIR/src RCFILE=$PROJECTDIR/coverage/.coveragerc # to run integration test: -m integration -# python3 -m pytest --log-level=info --log-cli-level=info --verbose -m "not integration" \ -# pluggables/tests/test_pluggables_with_SBI.py +python3 -m pytest --log-level=info --log-cli-level=info --verbose -m "not integration" \ + pluggables/tests/test_pluggables_with_SBI.py python3 -m pytest --log-level=info --log-cli-level=info --verbose -m "not integration"\ pluggables/tests/test_pluggables.py diff --git a/src/device/requirements.in b/src/device/requirements.in index 10139bc2c..be939ae7e 100644 --- a/src/device/requirements.in +++ b/src/device/requirements.in @@ -34,7 +34,7 @@ paramiko==2.11.* pyang==2.6.* git+https://github.com/robshakir/pyangbind.git python-json-logger==2.0.2 -#pytz==2021.3 +pytz==2021.3 #redis==4.1.2 requests==2.27.1 requests-mock==1.9.3 diff --git a/src/pluggables/service/PluggablesServiceServicerImpl.py b/src/pluggables/service/PluggablesServiceServicerImpl.py index c34b0999f..6f450cceb 100644 --- a/src/pluggables/service/PluggablesServiceServicerImpl.py +++ b/src/pluggables/service/PluggablesServiceServicerImpl.py @@ -13,18 +13,19 @@ # limitations under the License. import json, logging, grpc -from common.method_wrappers.Decorator import MetricsPool, safe_and_metered_rpc_method -from common.proto.context_pb2 import Empty, Device, DeviceId -from common.proto.pluggables_pb2_grpc import PluggablesServiceServicer -from common.proto.pluggables_pb2 import ( - Pluggable, CreatePluggableRequest, ListPluggablesRequest, ListPluggablesResponse, - GetPluggableRequest, DeletePluggableRequest, ConfigurePluggableRequest) +from .config_translator import translate_pluggable_config_to_netconf, create_config_rule_from_dict +from common.method_wrappers.Decorator import MetricsPool, safe_and_metered_rpc_method from common.method_wrappers.ServiceExceptions import ( - NotFoundException, AlreadyExistsException, InvalidArgumentException, OperationFailedException) -from common.tools.object_factory.ConfigRule import json_config_rule_set -from context.client.ContextClient import ContextClient -from device.client.DeviceClient import DeviceClient -from .config_translator import translate_pluggable_config_to_netconf, create_config_rule_from_dict + NotFoundException, AlreadyExistsException, InvalidArgumentException, OperationFailedException) +from common.proto.context_pb2 import Empty, Device, DeviceId +from common.proto.pluggables_pb2 import ( + Pluggable, PluggableConfig, CreatePluggableRequest, ListPluggablesRequest, ListPluggablesResponse, + GetPluggableRequest, DeletePluggableRequest, ConfigurePluggableRequest) +from common.proto.pluggables_pb2_grpc import PluggablesServiceServicer +from common.tools.object_factory.ConfigRule import json_config_rule_set +from context.client.ContextClient import ContextClient +from device.client.DeviceClient import DeviceClient + LOGGER = logging.getLogger(__name__) METRICS_POOL = MetricsPool('Pluggables', 'ServicegRPC') @@ -32,9 +33,9 @@ METRICS_POOL = MetricsPool('Pluggables', 'ServicegRPC') class PluggablesServiceServicerImpl(PluggablesServiceServicer): def __init__(self): LOGGER.info('Initiate PluggablesService') - self.pluggables = {} # In-memory store for pluggables + self.pluggables = {} # In-memory store for pluggables self.context_client = ContextClient() - self.device_client = DeviceClient() + self.device_client = DeviceClient() def _push_config_to_device(self, device_uuid: str, pluggable_index: int, pluggable_config): # type: ignore """ @@ -56,21 +57,22 @@ class PluggablesServiceServicerImpl(PluggablesServiceServicer): # Translate pluggable config to NETCONF format component_name = f"channel-{pluggable_index}" - netconf_config = translate_pluggable_config_to_netconf(pluggable_config, component_name=component_name) + config = translate_pluggable_config_to_netconf(pluggable_config, component_name=component_name) - LOGGER.info(f"Translated pluggable config to NETCONF format: {netconf_config}") + LOGGER.info(f"Translated pluggable config to NETCONF format: {config}") # Step 2: Create configuration rule with generic pluggable template # Use template index 1 for standard pluggable configuration template_index = 1 - resource_key = f"/pluggable/{template_index}/config" + resource_key = f"/pluggable/{template_index}/config" # Create config rule dict and convert to protobuf - config_json = json.dumps(netconf_config) + config_json = json.dumps(config) config_rule_dict = json_config_rule_set(resource_key, config_json) - config_rule = create_config_rule_from_dict(config_rule_dict) + config_rule = create_config_rule_from_dict(config_rule_dict) # Step 3: Create a minimal Device object with only the DSCM config rule + # TODO: In future, if device config merging is needed, fetch existing device config first config_device = Device() config_device.device_id.device_uuid.uuid = device_uuid # type: ignore config_device.device_config.config_rules.append(config_rule) # type: ignore @@ -195,25 +197,38 @@ class PluggablesServiceServicerImpl(PluggablesServiceServicer): LOGGER.info(f'No matching pluggable found: device={device_uuid}, index={pluggable_index}') raise NotFoundException('Pluggable', pluggable_key) - # Remove pluggable config from device - # TODO: Verify deletion works with actual hub and leaf devices - # + # Remove pluggable config from device before deleting from memory + # Use empty config to signal deletion to the device driver try: - pluggable = self.pluggables[pluggable_key] - # Create empty config to trigger deletion - from common.proto.pluggables_pb2 import PluggableConfig - empty_config = PluggableConfig() - empty_config.id.device.device_uuid.uuid = device_uuid - empty_config.id.pluggable_index = pluggable_index + del_config = PluggableConfig() + del_config.id.device.device_uuid.uuid = device_uuid + del_config.id.pluggable_index = pluggable_index + del_config.operational_mode = 0 # Operational mode + del_config.channel_name = "channel-1" # Channel name for component + + group_1 = del_config.dsc_groups.add() + group_1.id.pluggable.device.device_uuid.uuid = device_uuid + group_1.id.pluggable.pluggable_index = pluggable_index + group_1.id.group_index = 0 + + subcarrier_1 = group_1.subcarriers.add() + subcarrier_1.id.group.pluggable.device.device_uuid.uuid = device_uuid + subcarrier_1.id.group.pluggable.pluggable_index = pluggable_index + subcarrier_1.id.group.group_index = 0 + subcarrier_1.id.subcarrier_index = 0 + subcarrier_1.active = False + + # ADD MORE SUBCARRIERS IF NEEDED LOGGER.info(f"Removing configuration from device {device_uuid}") - self._push_config_to_device(device_uuid, pluggable_index, empty_config) - except Exception as e: - LOGGER.error(f"Failed to remove config from device: {e}") + self._push_config_to_device(device_uuid, pluggable_index, del_config) + except (grpc.RpcError, InvalidArgumentException, OperationFailedException) as e: + LOGGER.warning(f"Failed to remove config from device (continuing with memory deletion): {e}") # Continue with deletion from memory even if device config removal fails + # Always delete from memory (even if device operation failed) del self.pluggables[pluggable_key] - LOGGER.info(f"Deleted pluggable: device={device_uuid}, index={pluggable_index}") + LOGGER.info(f"Deleted pluggable from memory: device={device_uuid}, index={pluggable_index}") return Empty() diff --git a/src/pluggables/service/config_translator.py b/src/pluggables/service/config_translator.py index 4c0e8c765..501e48b3f 100644 --- a/src/pluggables/service/config_translator.py +++ b/src/pluggables/service/config_translator.py @@ -33,18 +33,17 @@ def translate_pluggable_config_to_netconf( component_name: str = "channel-1" # Fallback if channel_name not provided (channel-1 for HUB and channel-1/3/5 for LEAF) ) -> Dict[str, Any]: """ - Translate PluggableConfig protobuf message to the format expected by NetConfDriver. + Translate PluggableConfig protobuf message to the format expected by OpenConfig. Args: pluggable_config: PluggableConfig message containing DSC groups and subcarriers component_name: Fallback name if channel_name is not specified in config (default: "channel-1") Returns: - Dictionary in the format expected by NetConfDriver templates: + Dictionary in the format expected by OpenConfig templates: """ if not pluggable_config or not pluggable_config.dsc_groups: LOGGER.warning("Empty pluggable config provided") return { - "name": channel_name, "operation": "delete" } if hasattr(pluggable_config, 'channel_name') and pluggable_config.channel_name: @@ -54,11 +53,11 @@ def translate_pluggable_config_to_netconf( channel_name = component_name LOGGER.debug(f"Using fallback component_name: {channel_name}") - if not hasattr(pluggable_config, 'center_frequency_mhz') or pluggable_config.center_frequency_mhz <= 0: + if not hasattr(pluggable_config, 'center_frequency_mhz'): raise ValueError("center_frequency_mhz is required and must be greater than 0 in PluggableConfig") center_frequency_mhz = int(pluggable_config.center_frequency_mhz) - if not hasattr(pluggable_config, 'operational_mode') or pluggable_config.operational_mode <= 0: + if not hasattr(pluggable_config, 'operational_mode'): raise ValueError("operational_mode is required and must be greater than 0 in PluggableConfig") operational_mode = pluggable_config.operational_mode diff --git a/src/pluggables/tests/test_pluggables_with_SBI.py b/src/pluggables/tests/test_pluggables_with_SBI.py index d93142942..3ddf46e2e 100644 --- a/src/pluggables/tests/test_pluggables_with_SBI.py +++ b/src/pluggables/tests/test_pluggables_with_SBI.py @@ -83,14 +83,14 @@ def test_create_pluggable_hub_with_config(pluggables_client: PluggablesClient): _pluggable = pluggables_client.CreatePluggable(_request) assert isinstance(_pluggable, Pluggable) - assert _pluggable.id.device.device_uuid.uuid == DEVICE_HUB_UUID - assert _pluggable.id.pluggable_index == 2 - assert len(_pluggable.config.dsc_groups) == 1 # Should be 1, not 2 (check testmessages.py) + assert _pluggable.id.device.device_uuid.uuid == DEVICE_HUB_UUID # pyright: ignore[reportAttributeAccessIssue] + assert _pluggable.id.pluggable_index == 2 # pyright: ignore[reportAttributeAccessIssue] + assert len(_pluggable.config.dsc_groups) == 1 # Should be 1, not 2 (check testmessages.py) # pyright: ignore[reportAttributeAccessIssue] # Verify DSC group configuration - dsc_group = _pluggable.config.dsc_groups[0] - assert dsc_group.group_size == 4 # From testmessages.py - assert len(dsc_group.subcarriers) == 2 + dsc_group = _pluggable.config.dsc_groups[0] # pyright: ignore[reportAttributeAccessIssue] + assert dsc_group.group_size == 4 # From testmessages.py # pyright: ignore[reportAttributeAccessIssue] + assert len(dsc_group.subcarriers) == 2 # pyright: ignore[reportAttributeAccessIssue] LOGGER.info(f'Created Pluggable on Hub with {len(dsc_group.subcarriers)} subcarriers') @@ -112,9 +112,9 @@ def test_create_pluggable_leaf_with_config(pluggables_client: PluggablesClient): _pluggable = pluggables_client.CreatePluggable(_request) assert isinstance(_pluggable, Pluggable) - assert _pluggable.id.device.device_uuid.uuid == DEVICE_LEAF_UUID - assert _pluggable.id.pluggable_index == 1 # Should be 1, not 0 - assert len(_pluggable.config.dsc_groups) == 1 + assert _pluggable.id.device.device_uuid.uuid == DEVICE_LEAF_UUID # pyright: ignore[reportAttributeAccessIssue] + assert _pluggable.id.pluggable_index == 1 # Should be 1, not 0 # pyright: ignore[reportAttributeAccessIssue] + assert len(_pluggable.config.dsc_groups) == 1 # pyright: ignore[reportAttributeAccessIssue] LOGGER.info(f'Created Pluggable on Leaf: {_pluggable.id}') @@ -131,7 +131,7 @@ def test_configure_pluggable_hub(pluggables_client: PluggablesClient): with_initial_config=False ) _created = pluggables_client.CreatePluggable(_create_request) - assert _created.id.pluggable_index == 3 + assert _created.id.pluggable_index == 3 # pyright: ignore[reportAttributeAccessIssue] # Now configure it _config_request = create_configure_pluggable_request( @@ -143,14 +143,14 @@ def test_configure_pluggable_hub(pluggables_client: PluggablesClient): _configured = pluggables_client.ConfigurePluggable(_config_request) assert isinstance(_configured, Pluggable) - assert _configured.id.device.device_uuid.uuid == DEVICE_HUB_UUID - assert _configured.id.pluggable_index == 3 - assert len(_configured.config.dsc_groups) == 1 + assert _configured.id.device.device_uuid.uuid == DEVICE_HUB_UUID # pyright: ignore[reportAttributeAccessIssue] + assert _configured.id.pluggable_index == 3 # pyright: ignore[reportAttributeAccessIssue] + assert len(_configured.config.dsc_groups) == 1 # pyright: ignore[reportAttributeAccessIssue] # Verify configuration was applied - dsc_group = _configured.config.dsc_groups[0] - assert dsc_group.group_size == 2 - assert len(dsc_group.subcarriers) == 2 + dsc_group = _configured.config.dsc_groups[0] # pyright: ignore[reportAttributeAccessIssue] + assert dsc_group.group_size == 2 # pyright: ignore[reportAttributeAccessIssue] + assert len(dsc_group.subcarriers) == 2 # pyright: ignore[reportAttributeAccessIssue] LOGGER.info(f'Configured Pluggable on Hub with {len(dsc_group.subcarriers)} subcarriers') @@ -181,9 +181,9 @@ def test_get_pluggable(pluggables_client: PluggablesClient): _retrieved = pluggables_client.GetPluggable(_get_request) assert isinstance(_retrieved, Pluggable) - assert _retrieved.id.device.device_uuid.uuid == DEVICE_HUB_UUID - assert _retrieved.id.pluggable_index == 4 - assert len(_retrieved.config.dsc_groups) == len(_created.config.dsc_groups) + assert _retrieved.id.device.device_uuid.uuid == DEVICE_HUB_UUID # pyright: ignore[reportAttributeAccessIssue] + assert _retrieved.id.pluggable_index == 4 # pyright: ignore[reportAttributeAccessIssue] + assert len(_retrieved.config.dsc_groups) == len(_created.config.dsc_groups) # pyright: ignore[reportAttributeAccessIssue] LOGGER.info(f'Retrieved Pluggable: {_retrieved.id}') @@ -203,8 +203,8 @@ def test_list_pluggables(pluggables_client: PluggablesClient): assert len(_response.pluggables) >= 1 # At least one from previous tests for pluggable in _response.pluggables: - assert pluggable.id.device.device_uuid.uuid == DEVICE_HUB_UUID - LOGGER.info(f'Found Pluggable: index={pluggable.id.pluggable_index}') + assert pluggable.id.device.device_uuid.uuid == DEVICE_HUB_UUID # pyright: ignore[reportAttributeAccessIssue] + LOGGER.info(f'Found Pluggable: index={pluggable.id.pluggable_index}') # pyright: ignore[reportAttributeAccessIssue] # Number 7. @pytest.mark.integration @@ -222,7 +222,7 @@ def test_delete_pluggable(pluggables_client: PluggablesClient): with_initial_config=True ) _created = pluggables_client.CreatePluggable(_create_request) - assert _created.id.pluggable_index == 2 + assert _created.id.pluggable_index == 2 # pyright: ignore[reportAttributeAccessIssue] # Delete it _delete_request = create_delete_pluggable_request( @@ -257,7 +257,7 @@ def test_pluggable_already_exists_error(pluggables_client: PluggablesClient): # Create first time - should succeed _pluggable = pluggables_client.CreatePluggable(_request) - assert _pluggable.id.pluggable_index == 3 # Should be 3, not 5 + assert _pluggable.id.pluggable_index == 3 # Should be 3, not 5 # pyright: ignore[reportAttributeAccessIssue] # Try to create again - should fail with pytest.raises(grpc.RpcError) as e: diff --git a/src/pluggables/tests/testmessages.py b/src/pluggables/tests/testmessages.py index c18924bab..7cc5e0df1 100644 --- a/src/pluggables/tests/testmessages.py +++ b/src/pluggables/tests/testmessages.py @@ -235,11 +235,7 @@ def create_configure_pluggable_request( subcarrier_2.target_output_power_dbm = -10.0 subcarrier_2.center_frequency_hz = 193100075000000 # 193.175 THz subcarrier_2.symbol_rate_baud = 64000000000 # 64 GBaud - - # Set update mask if provided - if update_mask_paths: - _request.update_mask.paths.extend(update_mask_paths) - + _request.view_level = view_level _request.apply_timeout_seconds = apply_timeout_seconds -- GitLab From 242ee4051ba741847b41be96650e9c164c7fca36 Mon Sep 17 00:00:00 2001 From: Waleed Akbar Date: Tue, 11 Nov 2025 06:40:08 +0000 Subject: [PATCH 4/6] Fix import path for register_dscm_oc in MockWebServer --- src/nbi/tests/DSCM_MockWebServer.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/nbi/tests/DSCM_MockWebServer.py b/src/nbi/tests/DSCM_MockWebServer.py index 604da6810..1a718d65b 100644 --- a/src/nbi/tests/DSCM_MockWebServer.py +++ b/src/nbi/tests/DSCM_MockWebServer.py @@ -15,7 +15,7 @@ import logging, threading from nbi.service.NbiApplication import NbiApplication -from nbi.service.rest_server.nbi_plugins.dscm_oc import register_dscm_oc +from nbi.service.dscm_oc import register_dscm_oc from .Constants import LOCAL_HOST, NBI_SERVICE_PORT, NBI_SERVICE_PREFIX_URL -- GitLab From 9f21dfac400914196e3e6521c781c06e00971d84 Mon Sep 17 00:00:00 2001 From: Waleed Akbar Date: Tue, 11 Nov 2025 06:45:45 +0000 Subject: [PATCH 5/6] Comment out integration tests for pluggable creation and configuration in test files --- src/pluggables/tests/test_pluggables.py | 84 ++--- .../tests/test_pluggables_with_SBI.py | 286 +++++++++--------- 2 files changed, 185 insertions(+), 185 deletions(-) diff --git a/src/pluggables/tests/test_pluggables.py b/src/pluggables/tests/test_pluggables.py index 2e2b9ac7a..d6d4c8955 100644 --- a/src/pluggables/tests/test_pluggables.py +++ b/src/pluggables/tests/test_pluggables.py @@ -81,24 +81,24 @@ def test_CreatePluggable(pluggables_client : PluggablesClient): assert _pluggable.id.device.device_uuid.uuid == _pluggable_request.device.device_uuid.uuid -# CreatePluggable Test with configuration -@pytest.mark.integration -def test_CreatePluggable_with_config(pluggables_client : PluggablesClient): - LOGGER.info('Creating Pluggable with initial configuration for test...') - _pluggable_request = create_pluggable_request( - device_uuid = DEVICE_HUB_UUID, - preferred_pluggable_index = 0, - with_initial_config = True) - _pluggable = pluggables_client.CreatePluggable(_pluggable_request) - LOGGER.info('Created Pluggable with initial configuration for test: %s', _pluggable) - assert isinstance(_pluggable, Pluggable) - assert _pluggable.id.pluggable_index == 0 - assert _pluggable.id.device.device_uuid.uuid == _pluggable_request.device.device_uuid.uuid - assert _pluggable.config is not None - assert len(_pluggable.config.dsc_groups) == 1 - dsc_group = _pluggable.config.dsc_groups[0] - assert dsc_group.group_size == 4 - assert len(dsc_group.subcarriers) == 2 +# # CreatePluggable Test with configuration +# @pytest.mark.integration +# def test_CreatePluggable_with_config(pluggables_client : PluggablesClient): +# LOGGER.info('Creating Pluggable with initial configuration for test...') +# _pluggable_request = create_pluggable_request( +# device_uuid = DEVICE_HUB_UUID, +# preferred_pluggable_index = 0, +# with_initial_config = True) +# _pluggable = pluggables_client.CreatePluggable(_pluggable_request) +# LOGGER.info('Created Pluggable with initial configuration for test: %s', _pluggable) +# assert isinstance(_pluggable, Pluggable) +# assert _pluggable.id.pluggable_index == 0 +# assert _pluggable.id.device.device_uuid.uuid == _pluggable_request.device.device_uuid.uuid +# assert _pluggable.config is not None +# assert len(_pluggable.config.dsc_groups) == 1 +# dsc_group = _pluggable.config.dsc_groups[0] +# assert dsc_group.group_size == 4 +# assert len(dsc_group.subcarriers) == 2 # create pluggable request with pluggable key already exists error def test_CreatePluggable_already_exists(pluggables_client : PluggablesClient): @@ -188,27 +188,27 @@ def test_DeletePluggable(pluggables_client : PluggablesClient): assert e.value.code() == grpc.StatusCode.NOT_FOUND # ConfigurePluggable Test -@pytest.mark.integration -def test_ConfigurePluggable(pluggables_client : PluggablesClient): - LOGGER.info('Starting ConfigurePluggable test...') - LOGGER.info('Creating Pluggable to configure for test...') - - # First create a pluggable to configure it later - _pluggable_request = create_pluggable_request( - device_uuid=DEVICE_HUB_UUID, - preferred_pluggable_index=3) - _created_pluggable = pluggables_client.CreatePluggable(_pluggable_request) - LOGGER.info('Created Pluggable to configure for test: %s', _created_pluggable) - - _configure_request = create_configure_pluggable_request( - device_uuid = _created_pluggable.id.device.device_uuid.uuid, - pluggable_index = _created_pluggable.id.pluggable_index, - ) - _pluggable = pluggables_client.ConfigurePluggable(_configure_request) - LOGGER.info('Configured Pluggable for test: %s', _pluggable) - assert isinstance(_pluggable, Pluggable) - assert _pluggable.config is not None - assert len(_pluggable.config.dsc_groups) == 1 - dsc_group = _pluggable.config.dsc_groups[0] - assert dsc_group.group_size == 2 - assert len(dsc_group.subcarriers) == 2 +# @pytest.mark.integration +# def test_ConfigurePluggable(pluggables_client : PluggablesClient): +# LOGGER.info('Starting ConfigurePluggable test...') +# LOGGER.info('Creating Pluggable to configure for test...') + +# # First create a pluggable to configure it later +# _pluggable_request = create_pluggable_request( +# device_uuid=DEVICE_HUB_UUID, +# preferred_pluggable_index=3) +# _created_pluggable = pluggables_client.CreatePluggable(_pluggable_request) +# LOGGER.info('Created Pluggable to configure for test: %s', _created_pluggable) + +# _configure_request = create_configure_pluggable_request( +# device_uuid = _created_pluggable.id.device.device_uuid.uuid, +# pluggable_index = _created_pluggable.id.pluggable_index, +# ) +# _pluggable = pluggables_client.ConfigurePluggable(_configure_request) +# LOGGER.info('Configured Pluggable for test: %s', _pluggable) +# assert isinstance(_pluggable, Pluggable) +# assert _pluggable.config is not None +# assert len(_pluggable.config.dsc_groups) == 1 +# dsc_group = _pluggable.config.dsc_groups[0] +# assert dsc_group.group_size == 2 +# assert len(dsc_group.subcarriers) == 2 diff --git a/src/pluggables/tests/test_pluggables_with_SBI.py b/src/pluggables/tests/test_pluggables_with_SBI.py index 3ddf46e2e..0b8a5f60a 100644 --- a/src/pluggables/tests/test_pluggables_with_SBI.py +++ b/src/pluggables/tests/test_pluggables_with_SBI.py @@ -66,126 +66,126 @@ def test_create_pluggable_hub_without_config(pluggables_client: PluggablesClient LOGGER.info(f'Created Pluggable on Hub: {_pluggable.id}') # Number 2. -@pytest.mark.integration -def test_create_pluggable_hub_with_config(pluggables_client: PluggablesClient): - """Test creating a pluggable on Hub device with initial configuration +# @pytest.mark.integration +# def test_create_pluggable_hub_with_config(pluggables_client: PluggablesClient): +# """Test creating a pluggable on Hub device with initial configuration - Requires: Real NETCONF device at 10.30.7.7:2023 - """ - LOGGER.info('Creating Pluggable on Hub device with config...') +# Requires: Real NETCONF device at 10.30.7.7:2023 +# """ +# LOGGER.info('Creating Pluggable on Hub device with config...') - _request = create_pluggable_request( - device_uuid=DEVICE_HUB_UUID, - preferred_pluggable_index=2, # Use index 2 to avoid conflict with test #1 - with_initial_config=True - ) +# _request = create_pluggable_request( +# device_uuid=DEVICE_HUB_UUID, +# preferred_pluggable_index=2, # Use index 2 to avoid conflict with test #1 +# with_initial_config=True +# ) - _pluggable = pluggables_client.CreatePluggable(_request) +# _pluggable = pluggables_client.CreatePluggable(_request) - assert isinstance(_pluggable, Pluggable) - assert _pluggable.id.device.device_uuid.uuid == DEVICE_HUB_UUID # pyright: ignore[reportAttributeAccessIssue] - assert _pluggable.id.pluggable_index == 2 # pyright: ignore[reportAttributeAccessIssue] - assert len(_pluggable.config.dsc_groups) == 1 # Should be 1, not 2 (check testmessages.py) # pyright: ignore[reportAttributeAccessIssue] +# assert isinstance(_pluggable, Pluggable) +# assert _pluggable.id.device.device_uuid.uuid == DEVICE_HUB_UUID # pyright: ignore[reportAttributeAccessIssue] +# assert _pluggable.id.pluggable_index == 2 # pyright: ignore[reportAttributeAccessIssue] +# assert len(_pluggable.config.dsc_groups) == 1 # Should be 1, not 2 (check testmessages.py) # pyright: ignore[reportAttributeAccessIssue] - # Verify DSC group configuration - dsc_group = _pluggable.config.dsc_groups[0] # pyright: ignore[reportAttributeAccessIssue] - assert dsc_group.group_size == 4 # From testmessages.py # pyright: ignore[reportAttributeAccessIssue] - assert len(dsc_group.subcarriers) == 2 # pyright: ignore[reportAttributeAccessIssue] +# # Verify DSC group configuration +# dsc_group = _pluggable.config.dsc_groups[0] # pyright: ignore[reportAttributeAccessIssue] +# assert dsc_group.group_size == 4 # From testmessages.py # pyright: ignore[reportAttributeAccessIssue] +# assert len(dsc_group.subcarriers) == 2 # pyright: ignore[reportAttributeAccessIssue] - LOGGER.info(f'Created Pluggable on Hub with {len(dsc_group.subcarriers)} subcarriers') +# LOGGER.info(f'Created Pluggable on Hub with {len(dsc_group.subcarriers)} subcarriers') -# Number 3. -@pytest.mark.integration -def test_create_pluggable_leaf_with_config(pluggables_client: PluggablesClient): - """Test creating a pluggable on Leaf device with initial configuration +# # Number 3. +# @pytest.mark.integration +# def test_create_pluggable_leaf_with_config(pluggables_client: PluggablesClient): +# """Test creating a pluggable on Leaf device with initial configuration - Requires: Real NETCONF device at 10.30.7.8:2023 - """ - LOGGER.info('Creating Pluggable on Leaf device with config...') +# Requires: Real NETCONF device at 10.30.7.8:2023 +# """ +# LOGGER.info('Creating Pluggable on Leaf device with config...') - _request = create_pluggable_request( - device_uuid=DEVICE_LEAF_UUID, - preferred_pluggable_index=1, - with_initial_config=True - ) +# _request = create_pluggable_request( +# device_uuid=DEVICE_LEAF_UUID, +# preferred_pluggable_index=1, +# with_initial_config=True +# ) - _pluggable = pluggables_client.CreatePluggable(_request) +# _pluggable = pluggables_client.CreatePluggable(_request) - assert isinstance(_pluggable, Pluggable) - assert _pluggable.id.device.device_uuid.uuid == DEVICE_LEAF_UUID # pyright: ignore[reportAttributeAccessIssue] - assert _pluggable.id.pluggable_index == 1 # Should be 1, not 0 # pyright: ignore[reportAttributeAccessIssue] - assert len(_pluggable.config.dsc_groups) == 1 # pyright: ignore[reportAttributeAccessIssue] +# assert isinstance(_pluggable, Pluggable) +# assert _pluggable.id.device.device_uuid.uuid == DEVICE_LEAF_UUID # pyright: ignore[reportAttributeAccessIssue] +# assert _pluggable.id.pluggable_index == 1 # Should be 1, not 0 # pyright: ignore[reportAttributeAccessIssue] +# assert len(_pluggable.config.dsc_groups) == 1 # pyright: ignore[reportAttributeAccessIssue] - LOGGER.info(f'Created Pluggable on Leaf: {_pluggable.id}') +# LOGGER.info(f'Created Pluggable on Leaf: {_pluggable.id}') -# Number 4. -@pytest.mark.integration -def test_configure_pluggable_hub(pluggables_client: PluggablesClient): - """Test configuring an existing pluggable on Hub device""" - LOGGER.info('Configuring existing Pluggable on Hub device...') - - # First, create a pluggable without config - _create_request = create_pluggable_request( - device_uuid=DEVICE_HUB_UUID, - preferred_pluggable_index=3, # Use index 3 to avoid conflicts - with_initial_config=False - ) - _created = pluggables_client.CreatePluggable(_create_request) - assert _created.id.pluggable_index == 3 # pyright: ignore[reportAttributeAccessIssue] - - # Now configure it - _config_request = create_configure_pluggable_request( - device_uuid=DEVICE_HUB_UUID, - pluggable_index=3, # Match the created index - view_level=View.VIEW_FULL - ) - - _configured = pluggables_client.ConfigurePluggable(_config_request) - - assert isinstance(_configured, Pluggable) - assert _configured.id.device.device_uuid.uuid == DEVICE_HUB_UUID # pyright: ignore[reportAttributeAccessIssue] - assert _configured.id.pluggable_index == 3 # pyright: ignore[reportAttributeAccessIssue] - assert len(_configured.config.dsc_groups) == 1 # pyright: ignore[reportAttributeAccessIssue] - - # Verify configuration was applied - dsc_group = _configured.config.dsc_groups[0] # pyright: ignore[reportAttributeAccessIssue] - assert dsc_group.group_size == 2 # pyright: ignore[reportAttributeAccessIssue] - assert len(dsc_group.subcarriers) == 2 # pyright: ignore[reportAttributeAccessIssue] - - LOGGER.info(f'Configured Pluggable on Hub with {len(dsc_group.subcarriers)} subcarriers') +# # Number 4. +# @pytest.mark.integration +# def test_configure_pluggable_hub(pluggables_client: PluggablesClient): +# """Test configuring an existing pluggable on Hub device""" +# LOGGER.info('Configuring existing Pluggable on Hub device...') + +# # First, create a pluggable without config +# _create_request = create_pluggable_request( +# device_uuid=DEVICE_HUB_UUID, +# preferred_pluggable_index=3, # Use index 3 to avoid conflicts +# with_initial_config=False +# ) +# _created = pluggables_client.CreatePluggable(_create_request) +# assert _created.id.pluggable_index == 3 # pyright: ignore[reportAttributeAccessIssue] + +# # Now configure it +# _config_request = create_configure_pluggable_request( +# device_uuid=DEVICE_HUB_UUID, +# pluggable_index=3, # Match the created index +# view_level=View.VIEW_FULL +# ) + +# _configured = pluggables_client.ConfigurePluggable(_config_request) + +# assert isinstance(_configured, Pluggable) +# assert _configured.id.device.device_uuid.uuid == DEVICE_HUB_UUID # pyright: ignore[reportAttributeAccessIssue] +# assert _configured.id.pluggable_index == 3 # pyright: ignore[reportAttributeAccessIssue] +# assert len(_configured.config.dsc_groups) == 1 # pyright: ignore[reportAttributeAccessIssue] + +# # Verify configuration was applied +# dsc_group = _configured.config.dsc_groups[0] # pyright: ignore[reportAttributeAccessIssue] +# assert dsc_group.group_size == 2 # pyright: ignore[reportAttributeAccessIssue] +# assert len(dsc_group.subcarriers) == 2 # pyright: ignore[reportAttributeAccessIssue] + +# LOGGER.info(f'Configured Pluggable on Hub with {len(dsc_group.subcarriers)} subcarriers') -# Number 5. -@pytest.mark.integration -def test_get_pluggable(pluggables_client: PluggablesClient): - """Test retrieving an existing pluggable - - Requires: Real NETCONF device at 10.30.7.7:2023 - """ - LOGGER.info('Getting existing Pluggable...') - - # Create a pluggable first - _create_request = create_pluggable_request( - device_uuid=DEVICE_HUB_UUID, - preferred_pluggable_index=4, # Use index 4 to avoid conflicts - with_initial_config=True - ) - _created = pluggables_client.CreatePluggable(_create_request) - - # Now get it - _get_request = create_get_pluggable_request( - device_uuid=DEVICE_HUB_UUID, - pluggable_index=4, # Match the created index - view_level=View.VIEW_FULL - ) - - _retrieved = pluggables_client.GetPluggable(_get_request) - - assert isinstance(_retrieved, Pluggable) - assert _retrieved.id.device.device_uuid.uuid == DEVICE_HUB_UUID # pyright: ignore[reportAttributeAccessIssue] - assert _retrieved.id.pluggable_index == 4 # pyright: ignore[reportAttributeAccessIssue] - assert len(_retrieved.config.dsc_groups) == len(_created.config.dsc_groups) # pyright: ignore[reportAttributeAccessIssue] - - LOGGER.info(f'Retrieved Pluggable: {_retrieved.id}') +# # Number 5. +# @pytest.mark.integration +# def test_get_pluggable(pluggables_client: PluggablesClient): +# """Test retrieving an existing pluggable + +# Requires: Real NETCONF device at 10.30.7.7:2023 +# """ +# LOGGER.info('Getting existing Pluggable...') + +# # Create a pluggable first +# _create_request = create_pluggable_request( +# device_uuid=DEVICE_HUB_UUID, +# preferred_pluggable_index=4, # Use index 4 to avoid conflicts +# with_initial_config=True +# ) +# _created = pluggables_client.CreatePluggable(_create_request) + +# # Now get it +# _get_request = create_get_pluggable_request( +# device_uuid=DEVICE_HUB_UUID, +# pluggable_index=4, # Match the created index +# view_level=View.VIEW_FULL +# ) + +# _retrieved = pluggables_client.GetPluggable(_get_request) + +# assert isinstance(_retrieved, Pluggable) +# assert _retrieved.id.device.device_uuid.uuid == DEVICE_HUB_UUID # pyright: ignore[reportAttributeAccessIssue] +# assert _retrieved.id.pluggable_index == 4 # pyright: ignore[reportAttributeAccessIssue] +# assert len(_retrieved.config.dsc_groups) == len(_created.config.dsc_groups) # pyright: ignore[reportAttributeAccessIssue] + +# LOGGER.info(f'Retrieved Pluggable: {_retrieved.id}') # Number 6. def test_list_pluggables(pluggables_client: PluggablesClient): @@ -206,43 +206,43 @@ def test_list_pluggables(pluggables_client: PluggablesClient): assert pluggable.id.device.device_uuid.uuid == DEVICE_HUB_UUID # pyright: ignore[reportAttributeAccessIssue] LOGGER.info(f'Found Pluggable: index={pluggable.id.pluggable_index}') # pyright: ignore[reportAttributeAccessIssue] -# Number 7. -@pytest.mark.integration -def test_delete_pluggable(pluggables_client: PluggablesClient): - """Test deleting a pluggable - - Requires: Real NETCONF device at 10.30.7.8:2023 - """ - LOGGER.info('Deleting Pluggable...') - - # Create a pluggable to delete - _create_request = create_pluggable_request( - device_uuid=DEVICE_LEAF_UUID, - preferred_pluggable_index=2, # Use index 2 to avoid conflict with test #3 - with_initial_config=True - ) - _created = pluggables_client.CreatePluggable(_create_request) - assert _created.id.pluggable_index == 2 # pyright: ignore[reportAttributeAccessIssue] - - # Delete it - _delete_request = create_delete_pluggable_request( - device_uuid=DEVICE_LEAF_UUID, - pluggable_index=2 - ) - - _response = pluggables_client.DeletePluggable(_delete_request) - assert isinstance(_response, Empty) - - # Verify it's deleted - with pytest.raises(grpc.RpcError) as e: - _get_request = create_get_pluggable_request( - device_uuid=DEVICE_LEAF_UUID, - pluggable_index=2 - ) - pluggables_client.GetPluggable(_get_request) - - assert e.value.code() == grpc.StatusCode.NOT_FOUND - LOGGER.info('Successfully deleted Pluggable and verified removal') +# # Number 7. +# @pytest.mark.integration +# def test_delete_pluggable(pluggables_client: PluggablesClient): +# """Test deleting a pluggable + +# Requires: Real NETCONF device at 10.30.7.8:2023 +# """ +# LOGGER.info('Deleting Pluggable...') + +# # Create a pluggable to delete +# _create_request = create_pluggable_request( +# device_uuid=DEVICE_LEAF_UUID, +# preferred_pluggable_index=2, # Use index 2 to avoid conflict with test #3 +# with_initial_config=True +# ) +# _created = pluggables_client.CreatePluggable(_create_request) +# assert _created.id.pluggable_index == 2 # pyright: ignore[reportAttributeAccessIssue] + +# # Delete it +# _delete_request = create_delete_pluggable_request( +# device_uuid=DEVICE_LEAF_UUID, +# pluggable_index=2 +# ) + +# _response = pluggables_client.DeletePluggable(_delete_request) +# assert isinstance(_response, Empty) + +# # Verify it's deleted +# with pytest.raises(grpc.RpcError) as e: +# _get_request = create_get_pluggable_request( +# device_uuid=DEVICE_LEAF_UUID, +# pluggable_index=2 +# ) +# pluggables_client.GetPluggable(_get_request) + +# assert e.value.code() == grpc.StatusCode.NOT_FOUND +# LOGGER.info('Successfully deleted Pluggable and verified removal') # Number 8. def test_pluggable_already_exists_error(pluggables_client: PluggablesClient): -- GitLab From a0f18c277c3560e6631d16e47ee7085455866f7b Mon Sep 17 00:00:00 2001 From: Waleed Akbar Date: Tue, 11 Nov 2025 09:11:30 +0000 Subject: [PATCH 6/6] Updated NBI and Pluggable Service - preferred_pluggable_index values in pluggable tests for consistency. - Corrected nbi_service_import --- src/nbi/tests/test_dscm_restconf.py | 33 +++++++++++++++++-------- src/pluggables/tests/test_pluggables.py | 4 +-- 2 files changed, 25 insertions(+), 12 deletions(-) diff --git a/src/nbi/tests/test_dscm_restconf.py b/src/nbi/tests/test_dscm_restconf.py index 6ee0a946b..dae731992 100644 --- a/src/nbi/tests/test_dscm_restconf.py +++ b/src/nbi/tests/test_dscm_restconf.py @@ -14,19 +14,20 @@ from typing import Union import logging -import os, pytest +import os, pytest, time import requests -from .DSCM_MockWebServer import nbi_service_rest -from .messages.dscm_messages import get_hub_payload, get_leaf_payload -from common.Constants import ServiceNameEnum +from .DSCM_MockWebServer import MockWebServer +from .messages.dscm_messages import get_hub_payload, get_leaf_payload +from common.Constants import ServiceNameEnum +from nbi.service.NbiApplication import NbiApplication from common.proto.context_pb2_grpc import add_ContextServiceServicer_to_server from common.Settings import ( ENVVAR_SUFIX_SERVICE_HOST, ENVVAR_SUFIX_SERVICE_PORT_GRPC, get_env_var_name, get_service_port_grpc) -from common.tests.MockServicerImpl_Context import MockServicerImpl_Context +from common.tests.MockServicerImpl_Context import MockServicerImpl_Context from common.tools.service.GenericGrpcService import GenericGrpcService -from pluggables.client.PluggablesClient import PluggablesClient -from pluggables.service.PluggablesService import PluggablesService +from pluggables.client.PluggablesClient import PluggablesClient +from pluggables.service.PluggablesService import PluggablesService LOGGER = logging.getLogger(__name__) @@ -42,9 +43,9 @@ HEADERS = { "Accept" : "application/yang-data+json", LOCAL_HOST = '127.0.0.1' -DSCMPLUGGABLE_SERVICE_PORT = get_service_port_grpc(ServiceNameEnum.DSCMPLUGGABLE) -os.environ[get_env_var_name(ServiceNameEnum.DSCMPLUGGABLE, ENVVAR_SUFIX_SERVICE_HOST )] = str(LOCAL_HOST) -os.environ[get_env_var_name(ServiceNameEnum.DSCMPLUGGABLE, ENVVAR_SUFIX_SERVICE_PORT_GRPC)] = str(DSCMPLUGGABLE_SERVICE_PORT) +DSCMPLUGGABLE_SERVICE_PORT = get_service_port_grpc(ServiceNameEnum.PLUGGABLES) +os.environ[get_env_var_name(ServiceNameEnum.PLUGGABLES, ENVVAR_SUFIX_SERVICE_HOST )] = str(LOCAL_HOST) +os.environ[get_env_var_name(ServiceNameEnum.PLUGGABLES, ENVVAR_SUFIX_SERVICE_PORT_GRPC)] = str(DSCMPLUGGABLE_SERVICE_PORT) class MockContextService(GenericGrpcService): def __init__(self, bind_port: Union[str, int]) -> None: @@ -55,6 +56,18 @@ class MockContextService(GenericGrpcService): add_ContextServiceServicer_to_server(self.context_servicer, self.server) +@pytest.fixture(scope='session') +def nbi_service_rest() -> NbiApplication: + """Fixture to start the NBI REST service with DSCM plugin.""" + mock_web_server = MockWebServer() + mock_web_server.start() + time.sleep(1) # give time for the server to start + + nbi_app = mock_web_server.nbi_app + yield nbi_app + + # Cleanup: web server runs as daemon thread, so will terminate with main process + @pytest.fixture(scope='session') def pluggables_service(): LOGGER.info('Initializing DscmPluggableService...') diff --git a/src/pluggables/tests/test_pluggables.py b/src/pluggables/tests/test_pluggables.py index d6d4c8955..34ea05679 100644 --- a/src/pluggables/tests/test_pluggables.py +++ b/src/pluggables/tests/test_pluggables.py @@ -73,7 +73,7 @@ def test_CreatePluggable(pluggables_client : PluggablesClient): LOGGER.info('Creating Pluggable for test...') _pluggable_request = create_pluggable_request( device_uuid=DEVICE_HUB_UUID, - preferred_pluggable_index=-1) + preferred_pluggable_index=1) _pluggable = pluggables_client.CreatePluggable(_pluggable_request) LOGGER.info('Created Pluggable for test: %s', _pluggable) assert isinstance(_pluggable, Pluggable) @@ -141,7 +141,7 @@ def test_GetPluggable(pluggables_client : PluggablesClient): # First create a pluggable to get it later _pluggable_request = create_pluggable_request( device_uuid=DEVICE_HUB_UUID, - preferred_pluggable_index=1) + preferred_pluggable_index=10) _created_pluggable = pluggables_client.CreatePluggable(_pluggable_request) LOGGER.info('Created Pluggable for GetPluggable test: %s', _created_pluggable) -- GitLab