CT-147: no need to start docker containers to run tests. (33436db6) · Commits · Gianluca Marotta / csp-lmc-old

csp-lmc-common/.gitlab-ci.yml

+12 −3

Original line number	Diff line number	Diff line
		@@ -70,14 +70,23 @@ build:csp-lmc-common_image:
		paths:
		- ./$COMMON_BUILD_PATH/build/
		variables:
		COMMON_BUILD_PATH: csp-lmc-common/docker
		COMMON_BUILD_PATH: csp-lmc-common

		test:csp-lmc-common:
		extends: .test_common
		stage: test_common
		script:
		- cd $COMMON_BUILD_PATH
		- make test
		- pwd
		- python3 -m pip install -r requirements-tst.txt
		- python3 -m pip install --extra-index-url https://nexus.engageska-portugal.pt/repository/pypi/simple -r requirements.txt
		- python3 setup.py test \|tee setup_py_test.stdout
		- pwd
		- mkdir -p build/reports
		- ls -alR
		- mv setup_py_test.stdout build/csp-lmc-common-setup-test.stdout
		- mv htmlcov build/csp-lmc-common_htmlcov
		- mv coverage.xml build
		#
		# linting stage

csp-lmc-common/conftest.py

deleted100644 → 0

+0 −87

Original line number	Diff line number	Diff line
		"""
		A module defining a list of fixture functions.
		"""
		import importlib
		import pytest

		import tango
		from unittest import mock
		from tango import DeviceProxy
		from tango.test_context import DeviceTestContext

		@pytest.fixture(scope="class")
		def tango_context(request):
		"""Creates and returns a TANGO DeviceTestContext object.

		Parameters
		----------
		request: _pytest.fixtures.SubRequest
		A request object gives access to the requesting test context.
		"""
		test_properties = {
		'CspMaster': {'SkaLevel': '2',
		'CspCbf': 'mid_csp_cbf/sub_elt/master',
		'CspPss': 'mid_csp_pss/sub_elt/master',
		'CspPst': 'mid_csp_pst/sub_elt/master',
		},
		'CspSubarray': {'CspMaster': 'common/elt/master',
		'CbfSubarray': 'mid_csp_cbf/sub_elt/subarray_01',
		'PssSubarray': 'mid_csp_pss/sub_elt/subarray_01',
		'SubID' : 1,
		},
		}
		# TODO: package_name and class_name can be used in future
		# fq_test_class_name = request.cls.__module__
		# fq_test_class_name_details = fq_test_class_name.split(".")
		# package_name = fq_test_class_name_details[1]
		# class_name = module_name = fq_test_class_name_details[1]
		test_class_name = request.cls.__name__
		class_name = test_class_name.split('Test', 1)[-1]
		module = importlib.import_module("{}.{}".format("csp_lmc_common", class_name))
		klass = getattr(module, class_name)
		tango_context = DeviceTestContext(klass, properties=test_properties.get(class_name))
		tango_context.start()
		yield tango_context
		tango_context.stop()

		#
		# NOTE: if initialize_device is called, the tests crash with error because during the
		# CspMaster and CspMaster access to the TANGO DB to get the memorized attributes.
		# Need to mock the call to the TANGO DB get_device_attribute
		@pytest.fixture(scope="function")
		def initialize_device(tango_context):
		"""Re-initializes the device.

		Parameters
		----------
		tango_context: tango.test_context.DeviceTestContext
		Context to run a device without a database.
		"""
		yield tango_context.device.Init()

		@pytest.fixture(scope="class")
		def cbf_master():
		"""Create DeviceProxy for the CbfMaster device
		to test the device with the TANGO DB
		"""
		database = tango.Database()
		instance_list = database.get_device_exported_for_class('CbfMaster')
		for instance in instance_list.value_string:
		try:
		return tango.DeviceProxy(instance)
		except tango.DevFailed:
		continue


		@pytest.fixture(scope="class")
		def csp_master():
		"""Create DeviceProxy for the CspMaster device
		to test the device with the TANGO DB
		"""
		database = tango.Database()
		instance_list = database.get_device_exported_for_class('CspMaster')
		for instance in instance_list.value_string:
		try:
		return tango.DeviceProxy(instance)
		except tango.DevFailed:
		continue

csp-lmc-common/csp_lmc_common/CspMaster.py

+24 −24

Original line number	Diff line number	Diff line
		@@ -1212,12 +1212,12 @@ class CspMaster(SKAMaster):
		self._build_state = '{}, {}, {}'.format(release.name, release.version, release.description)
		self._version_id = release.version
		# connect to TANGO DB
		csp_tango_db = tango.Database()
		#csp_tango_db = tango.Database()
		# read the CSP memorized attributes from the TANGO DB.
		# Note: a memorized attribute has defined the attribute
		# property '__value'
		attribute_properties = csp_tango_db.get_device_attribute_property(self.get_name(),
		{'adminMode': ['__value']})
		#attribute_properties = csp_tango_db.get_device_attribute_property(self.get_name(),
		# {'adminMode': ['__value']})
		# set init values for the CSP Element and Sub-element SCM states
		self.set_state(tango.DevState.INIT)
		self._health_state = HealthState.OK
		@@ -1233,16 +1233,16 @@ class CspMaster(SKAMaster):

		# build a dictionary with the (attr_name, value) of the memorized attributes
		# use memorized atrtibute if present, otherwise the default one
		memorized_attr_dict = {attr_name : int(value[0]) for attr_name, db_key in attribute_properties.items()
		for key, value in db_key.items()
		if key == '__value'}
		try:
		self._admin_mode = memorized_attr_dict['adminMode']
		if self._admin_mode not in [AdminMode.ONLINE, AdminMode.MAINTENANCE]:
		self._health_state = HealthState.UNKNOWN
		self.set_state(tango.DevState.DISABLE)
		except KeyError as key_err:
		self.logger.info("Key {} not found".format(key_err))
		#memorized_attr_dict = {attr_name : int(value[0]) for attr_name, db_key in attribute_properties.items()
		# for key, value in db_key.items()
		# if key == '__value'}
		#try:
		# self._admin_mode = memorized_attr_dict['adminMode']
		# if self._admin_mode not in [AdminMode.ONLINE, AdminMode.MAINTENANCE]:
		# self._health_state = HealthState.UNKNOWN
		# self.set_state(tango.DevState.DISABLE)
		#except KeyError as key_err:
		# self.logger.info("Key {} not found".format(key_err))

		# initialize list with CSP sub-element FQDNs
		self._se_fqdn = []
		@@ -1260,17 +1260,17 @@ class CspMaster(SKAMaster):
		# read the sub-elements adminMode (memorized) attributes from
		# the TANGO DB.
		# Note: a memorized attribute has defined the attribute property '__value'
		for fqdn in self._se_fqdn:
		attribute_properties = csp_tango_db.get_device_attribute_property(fqdn,
		{'adminMode': ['__value']})
		self.logger.info("fqdn: {} attribute_properties: {}".format(fqdn, attribute_properties))
		try:
		admin_mode_memorized = attribute_properties['adminMode']['__value']
		self._se_admin_mode[fqdn] = int(admin_mode_memorized[0])
		except KeyError as key_err:
		msg = ("No key {} found for sub-element {}"
		" adminMode attribute".format(key_err, fqdn))
		self.logger.info(msg)
		#for fqdn in self._se_fqdn:
		# attribute_properties = csp_tango_db.get_device_attribute_property(fqdn,
		# {'adminMode': ['__value']})
		# self.logger.info("fqdn: {} attribute_properties: {}".format(fqdn, attribute_properties))
		# try:
		# admin_mode_memorized = attribute_properties['adminMode']['__value']
		# self._se_admin_mode[fqdn] = int(admin_mode_memorized[0])
		# except KeyError as key_err:
		# msg = ("No key {} found for sub-element {}"
		# " adminMode attribute".format(key_err, fqdn))
		# self.logger.info(msg)

		# _se_proxies: the sub-element proxies
		# implemented as dictionary:

csp-lmc-common/csp_lmc_common/CspSubarray.py

+21 −21

Original line number	Diff line number	Diff line
		@@ -156,25 +156,25 @@ class CspSubarray(SKASubarray):
		device._sc_subarray_obs_state = defaultdict(lambda: ObsState.IDLE)
		device._sc_subarray_obs_mode = defaultdict(lambda: ObsMode.IDLE)
		device._mutex_obs_state = threading.Lock()
		device._csp_tango_db = tango.Database()
		#device._csp_tango_db = tango.Database()
		# read the CSP memorized attributes from the TANGO DB.
		# Note: a memorized attribute has defined the attribute
		# property '__value'
		attribute_properties = device._csp_tango_db.get_device_attribute_property(device.get_name(),
		{'adminMode': ['__value']})
		#attribute_properties = device._csp_tango_db.get_device_attribute_property(device.get_name(),
		# {'adminMode': ['__value']})
		# build a dictionary with the (attr_name, value) of the memorized attributes
		# use memorized atrtibute if present, otherwise the default one
		memorized_attr_dict = {attr_name : int(value[0]) for attr_name, db_key in attribute_properties.items()
		for key, value in db_key.items()
		if key == '__value'}
		try:
		device._admin_mode = memorized_attr_dict['adminMode']
		if device._admin_mode not in [AdminMode.ONLINE, AdminMode.MAINTENANCE]:
		device._health_state = HealthState.UNKNOWN
		device.set_state(tango.DevState.DISABLE)

		except KeyError as key_err:
		device.logger.info("Key {} not found".format(key_err))
		#memorized_attr_dict = {attr_name : int(value[0]) for attr_name, db_key in attribute_properties.items()
		# for key, value in db_key.items()
		# if key == '__value'}
		#try:
		# device._admin_mode = memorized_attr_dict['adminMode']
		# if device._admin_mode not in [AdminMode.ONLINE, AdminMode.MAINTENANCE]:
		# device._health_state = HealthState.UNKNOWN
		# device.set_state(tango.DevState.DISABLE)

		#except KeyError as key_err:
		# device.logger.info("Key {} not found".format(key_err))

		# list of sub-array sub-component FQDNs
		device._sc_subarray_fqdn = []
		@@ -1653,14 +1653,14 @@ class CspSubarray(SKASubarray):
		return
		# read the sub-componet adminMode (memorized) attribute from
		# the CSP.LMC TANGO DB.
		attribute_properties = self._csp_tango_db.get_device_attribute_property(fqdn,
		{'adminMode': ['__value']})
		self.logger.debug("fqdn: {} attribute_properties: {}".format(fqdn, attribute_properties))
		try:
		admin_mode_memorized = attribute_properties['adminMode']['__value']
		self._sc_subarray_admin_mode[fqdn] = int(admin_mode_memorized[0])
		except KeyError as key_error:
		self.logger.warning("No key {} found".format(str(key_error)))
		#attribute_properties = self._csp_tango_db.get_device_attribute_property(fqdn,
		# {'adminMode': ['__value']})
		#self.logger.debug("fqdn: {} attribute_properties: {}".format(fqdn, attribute_properties))
		#try:
		# admin_mode_memorized = attribute_properties['adminMode']['__value']
		# self._sc_subarray_admin_mode[fqdn] = int(admin_mode_memorized[0])
		#except KeyError as key_error:
		# self.logger.warning("No key {} found".format(str(key_error)))
		try:
		log_msg = "Trying connection to " + str(fqdn) + " device"
		self.logger.info(log_msg)

csp-lmc-common/csp_lmc_common/init.py

+8 −0

Original line number	Diff line number	Diff line
		__all__ = [
		"CspMaster",
		"CspSubarray",
		]

		from .CspMaster import CspMaster
		from .CspSubarray import CspSubarray