Refactor test scripts and update frequency handling in JSON to Proto conversion (b0868e72) · Commits · TFS / controller

scripts/run_tests_locally-nbi-dscm.sh

+0 −14

Original line number	Diff line number	Diff line
		@@ -17,19 +17,11 @@

		PROJECTDIR=`pwd`

		# Activate the dscm virtual environment
		source /home/ubuntu/dscm/bin/activate

		cd $PROJECTDIR/src

		echo "======================================"
		echo "Running NBI DSCM RESTCONF Tests"
		echo "======================================"

		# test DSCM NBI functions
		/home/ubuntu/dscm/bin/python -m pytest --log-level=INFO --log-cli-level=INFO --verbose \
		nbi/tests/test_dscm_restconf.py::test_post_get_delete_leaf_optical_channel_frequency
		# nbi/tests/test_dscm_restconf.py::test_post_hub_optical_channel_frequency \

		# # test JSON to Proto conversion functions
		# /home/ubuntu/dscm/bin/python -m pytest --log-level=INFO --log-cli-level=INFO --verbose \
		@@ -38,9 +30,3 @@ echo "======================================"
		# nbi/tests/test_json_to_proto.py::test_configure_pluggable_request_hub_format \
		# nbi/tests/test_json_to_proto.py::test_configure_pluggable_request_leaf_format \
		# nbi/tests/test_json_to_proto.py::test_empty_payload


		echo ""
		echo "======================================"
		echo "Test execution completed"
		echo "======================================"

src/nbi/service/rest_server/nbi_plugins/dscm_oc/json_to_proto_conversion.py

+3 −3

Original line number	Diff line number	Diff line
		@@ -17,7 +17,7 @@ Helper functions to convert JSON payload from RESTCONF to Pluggables proto messa
		"""

		from typing import Dict, Any
		from common.proto import pluggables_pb2 # pyright: ignore[reportAttributeAccessIssue]
		from common.proto import pluggables_pb2


		def json_to_get_pluggable_request(
		@@ -183,7 +183,7 @@ def _add_dsc_groups_from_hub_format(

		# Set frequency and power from top-level payload
		if "frequency" in payload:
		subcarrier.center_frequency_hz = float(payload["frequency"]) * 1e6 # type: ignore[attr-defined]
		subcarrier.center_frequency_hz = float(payload["frequency"]) # type: ignore[attr-defined]

		if "target_output_power" in payload:
		subcarrier.target_output_power_dbm = float(payload["target_output_power"]) # type: ignore[attr-defined]
		@@ -230,7 +230,7 @@ def _add_dsc_groups_from_leaf_format(

		# Set frequency and power from channel data
		if "frequency" in channel_data:
		subcarrier.center_frequency_hz = float(channel_data["frequency"]) * 1e6 # type: ignore[attr-defined] # MHz to Hz
		subcarrier.center_frequency_hz = float(channel_data["frequency"]) # type: ignore[attr-defined] # MHz to Hz

		if "target_output_power" in channel_data:
		subcarrier.target_output_power_dbm = float(channel_data["target_output_power"]) # type: ignore[attr-defined]

src/nbi/service/rest_server/nbi_plugins/dscm_oc/routes.py

+3 −3

Original line number	Diff line number	Diff line
		@@ -90,7 +90,7 @@ def rc_get(rc_path, device_uuid=None):
		LOGGER.warning(f"Pluggable not found for device {device_uuid}: {e.details}")
		return _not_found(f"Pluggable not found: {e.details}", path=rc_path)

		except Exception as e:
		except ServiceException as e:
		LOGGER.error(f"Unexpected error getting pluggable for device {device_uuid}: {str(e)}", exc_info=True)
		return _bad_request(f"Failed to get pluggable: {str(e)}", path=rc_path)

		@@ -132,7 +132,7 @@ def rc_post(rc_path, device_uuid=None):
		LOGGER.warning(f"Invalid argument creating pluggable for device {device_uuid}: {e.details}")
		return _bad_request(f"Invalid argument: {e.details}", path=rc_path)

		except Exception as e:
		except ServiceException as e:
		LOGGER.error(f"Unexpected error creating pluggable for device {device_uuid}: {str(e)}", exc_info=True)
		return _bad_request(f"Failed to create pluggable: {str(e)}", path=rc_path)

		@@ -157,7 +157,7 @@ def rc_delete(rc_path, device_uuid=None):
		# DELETE is idempotent - return 204 even if resource doesn't exist
		return Response(status=204)

		except Exception as e:
		except ServiceException as e:
		LOGGER.error(f"Unexpected error deleting pluggable for device {device_uuid}: {str(e)}", exc_info=True)
		return _bad_request(f"Failed to delete pluggable: {str(e)}", path=rc_path)

src/nbi/tests/DSCM_MockWebServer.py

+1 −51

Original line number	Diff line number	Diff line
		@@ -13,7 +13,7 @@
		# limitations under the License.


		import logging, threading, pytest, time
		import logging, threading
		from nbi.service.NbiApplication import NbiApplication
		from nbi.service.rest_server.nbi_plugins.dscm_oc import register_dscm_oc
		from .Constants import LOCAL_HOST, NBI_SERVICE_PORT, NBI_SERVICE_PREFIX_URL
		@@ -38,53 +38,3 @@ class MockWebServer(threading.Thread):
		)
		except: # pylint: disable=bare-except
		LOGGER.exception('[MockWebServer::run] Unhandled Exception')



		# import os, pytest, time
		# from common.Constants import ServiceNameEnum
		# from common.Settings import (
		# ENVVAR_SUFIX_SERVICE_HOST, ENVVAR_SUFIX_SERVICE_PORT_HTTP,
		# get_env_var_name, get_service_baseurl_http, get_service_port_http
		# )
		# from nbi.service.rest_server.RestServer import RestServer
		# from nbi.service.rest_server.nbi_plugins.dscm_oc import register_dscm_oc
		# from nbi.service.rest_server.nbi_plugins.etsi_bwm import register_etsi_bwm_api
		# from nbi.service.rest_server.nbi_plugins.ietf_l2vpn import register_ietf_l2vpn
		# from nbi.service.rest_server.nbi_plugins.ietf_l3vpn import register_ietf_l3vpn
		# from nbi.service.rest_server.nbi_plugins.ietf_network import register_ietf_network
		# from nbi.service.rest_server.nbi_plugins.tfs_api import register_tfs_api
		# from nbi.tests.MockService_Dependencies import MockService_Dependencies


		# LOCAL_HOST = '127.0.0.1'
		# MOCKSERVICE_PORT = 10000
		# NBI_SERVICE_PORT = MOCKSERVICE_PORT + get_service_port_http(ServiceNameEnum.NBI) # avoid privileged ports
		# 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)

		# @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 : MockService_Dependencies): # pylint: disable=redefined-outer-name, unused-argument
		# _rest_server = RestServer()
		# register_etsi_bwm_api(_rest_server)
		# register_ietf_l2vpn(_rest_server)
		# register_ietf_l3vpn(_rest_server)
		# register_ietf_network(_rest_server)
		# register_tfs_api(_rest_server)
		# register_dscm_oc(_rest_server)

		# _rest_server.start()
		# time.sleep(1) # bring time for the server to start
		# yield _rest_server
		# _rest_server.shutdown()
		# _rest_server.join()

src/nbi/tests/messages/dscm_messages.py

+3 −3

Original line number	Diff line number	Diff line
		@@ -37,21 +37,21 @@ def get_leaf_payload():
		"channels": [
		{
		"name" : "channel-1",
		"frequency" : "195006250",
		"frequency" : "195006250000000",
		"target_output_power" : "-99.0",
		"operational_mode" : "1",
		"digital_subcarriers_groups": [{ "group_id": 1 }]
		},
		{
		"name" : "channel-3",
		"frequency" : "195018750",
		"frequency" : "195018750000000",
		"target_output_power" : "-99.0",
		"operational_mode" : "1",
		"digital_subcarriers_groups": [{ "group_id": 2 }]
		},
		{
		"name" : "channel-5",
		"frequency" : "195031250",
		"frequency" : "195031250000000",
		"target_output_power" : "-99.0",
		"operational_mode" : "1",
		"digital_subcarriers_groups": [{ "group_id": 3 }]