CT-207 Integration tests for OffCommand; fixed errors (b9eee971) · Commits · Gianluca Marotta / csp-lmc-old

csp-lmc-common/csp_lmc_common/CspSubarray.py

+9 −7

Original line number	Original line	Diff line number	Diff line
	@@ -38,7 +38,7 @@ from tango import AttrWriteType, DeviceProxy
	from ska.base import SKASubarray, SKASubarrayStateModel		from ska.base import SKASubarray, SKASubarrayStateModel
	#from ska.base import utils		#from ska.base import utils
	from ska.base.commands import ActionCommand, ResultCode		from ska.base.commands import ActionCommand, ResultCode
	from ska.base.faults import CapabilityValidationError		from ska.base.faults import CapabilityValidationError, CommandError
	from ska.base.control_model import HealthState, AdminMode, ObsState, ObsMode		from ska.base.control_model import HealthState, AdminMode, ObsState, ObsMode
	from .utils.cspcommons import CmdExecState		from .utils.cspcommons import CmdExecState
	from .utils.decorators import transaction_id		from .utils.decorators import transaction_id
	@@ -352,7 +352,7 @@ class CspSubarray(SKASubarray):
	self.logger.info("Call Off Command")		self.logger.info("Call Off Command")
	device = self.target		device = self.target
	subelement_thr = {}		subelement_thr = {}
	self.succeeded = []		self.thread_succeeded = []
	try:		try:
	# First loop that start the sub-element threads to get the ObsState.EMPTY		# First loop that start the sub-element threads to get the ObsState.EMPTY
	for fqdn in device._sc_subarray_fqdn:		for fqdn in device._sc_subarray_fqdn:
	@@ -368,7 +368,7 @@ class CspSubarray(SKASubarray):
	# Second loop check if the thread is succeeded and send the Off command to sub-element		# Second loop check if the thread is succeeded and send the Off command to sub-element
	for fqdn in subelement_thr.keys():		for fqdn in subelement_thr.keys():
	subelement_thr[fqdn].join()		subelement_thr[fqdn].join()
	if not self.succeeded[fqdn]:		if not self.thread_succeeded[fqdn]:
	self.logger.info(f'ObsState of device {fqdn} is not EMPTY. ObsState:{device._sc_subarray_obs_state[fqdn]}')		self.logger.info(f'ObsState of device {fqdn} is not EMPTY. ObsState:{device._sc_subarray_obs_state[fqdn]}')
	# TODO: GM: anche se non sono EMPTY li mandiamo in OFF?		# TODO: GM: anche se non sono EMPTY li mandiamo in OFF?
	if fqdn != device.CbfSubarray:		if fqdn != device.CbfSubarray:
	@@ -390,12 +390,14 @@ class CspSubarray(SKASubarray):
	except tango.DevFailed as tango_err:		except tango.DevFailed as tango_err:
	message = str(tango_err.args[0].desc)		message = str(tango_err.args[0].desc)
	return ResultCode.FAILED, message		return ResultCode.FAILED, message
			super().do()
	message = "Off command completed OK"		message = "Off command completed OK"
	self.logger.info(message)		self.logger.info(message)
	return ResultCode.OK, message		return ResultCode.OK, message

	def off_sequence(self, fqdn):		def off_sequence(self, fqdn):
	#Off command send the device in the state:OFF, ObsState:EMPTY		#Off command send the device in the state:OFF, ObsState:EMPTY
			self.logger.info('off sequence started!')
	device = self.target		device = self.target
	starting_time = time.time()		starting_time = time.time()
	timeout = device._sc_subarray_cmd_duration_expected[fqdn]['off']		timeout = device._sc_subarray_cmd_duration_expected[fqdn]['off']
	@@ -409,7 +411,7 @@ class CspSubarray(SKASubarray):
	if device._sc_subarray_obs_state[fqdn] == ObsState.FAULT:		if device._sc_subarray_obs_state[fqdn] == ObsState.FAULT:
	break		break
	time.sleep(0.1)		time.sleep(0.1)
	except ska.base.commands.CommandError as msg:		except CommandError as msg:
	self.logger.info(msg)		self.logger.info(msg)
	# Try to send Restart Command.		# Try to send Restart Command.
	# Device goes to EMPTY if ObsState is: FAULT, ABORTED		# Device goes to EMPTY if ObsState is: FAULT, ABORTED
	@@ -417,15 +419,15 @@ class CspSubarray(SKASubarray):
	try:		try:
	self.logger.info('Try to execute Restart command')		self.logger.info('Try to execute Restart command')
	device._sc_subarray_obs_state[fqdn].Restart()		device._sc_subarray_obs_state[fqdn].Restart()
	except ska.base.commands.CommandError as msg:		except CommandError as msg:
	self.logger.info(msg)		self.logger.info(msg)
	time.sleep(0.1)		time.sleep(0.1)
	# if ObsState is: RESOURCING, ABORTING start again to wait end of the transition		# if ObsState is: RESOURCING, ABORTING start again to wait end of the transition

	if device._sc_subarray_obs_state[fqdn] is ObsState.EMPTY:		if device._sc_subarray_obs_state[fqdn] is ObsState.EMPTY:
	self.succeeded[fqdn] = True		self.thread_succeeded[fqdn] = True
	else:		else:
	self.succeeded[fqdn] = False		self.thread_succeeded[fqdn] = False

	class AssignResourcesCommand(SKASubarray.AssignResourcesCommand):		class AssignResourcesCommand(SKASubarray.AssignResourcesCommand):
	@transaction_id		@transaction_id

csp-lmc-mid/tests/integration/MidCspSubarray_test.py

+18 −0

Original line number	Original line	Diff line number	Diff line
	@@ -200,6 +200,24 @@ class TestCspSubarray(TestBase):
	prober_subarray_state = Probe(self.midcsp_subarray01, "State", DevState.ON, f"CSP Subarray not OFF")		prober_subarray_state = Probe(self.midcsp_subarray01, "State", DevState.ON, f"CSP Subarray not OFF")
	Poller(4, 0.2).check(prober_subarray_state)		Poller(4, 0.2).check(prober_subarray_state)

			@pytest.mark.csp_k8s
			def test_OffCommand_after_RESOURCING(self):
			self._setup_subarray()
			self.midcsp_subarray01.Off()
			prober_subarray_state = Probe(self.midcsp_subarray01, "State", DevState.OFF, f"CSP Subarray not OFF")
			Poller(4, 0.2).check(prober_subarray_state)
			prober_subarray_obsstate = Probe(self.midcsp_subarray01, "obsState", ObsState.EMPTY, f"CSP Subarray not EMPTY")
			Poller(4, 0.2).check(prober_subarray_obsstate)

			def test_OffCommand_after_CONFIGURE(self):
			self._setup_subarray()
			self._configure_scan()
			self.midcsp_subarray01.Off()
			prober_subarray_state = Probe(self.midcsp_subarray01, "State", DevState.OFF, f"CSP Subarray not OFF")
			Poller(10, 0.2).check(prober_subarray_state)
			prober_subarray_obsstate = Probe(self.midcsp_subarray01, "obsState", ObsState.EMPTY, f"CSP Subarray not EMPTY")
			Poller(10, 0.2).check(prober_subarray_obsstate)

	@pytest.mark.csp_k8s		@pytest.mark.csp_k8s
	def test_add_receptors_WITH_invalid_id(self):		def test_add_receptors_WITH_invalid_id(self):
	"""		"""