refactor(drivers): respect the driver's interface (60436188) · Commits · TFS / controller

src/device/service/drivers/morpheus/MorpheusApiDriver.py

+85 −51

Original line number	Diff line number	Diff line
		import logging, requests, threading, json
		import logging, requests, threading, json, time
		from typing import Any, Iterator, List, Optional, Tuple, Union, Dict
		from queue import Queue
		from common.method_wrappers.Decorator import MetricsPool, metered_subclass_method
		@@ -46,10 +46,10 @@ class MorpheusApiDriver(_Driver):
		with self.__lock:
		try:
		if self.__detection_thread and self.__detection_thread.is_alive():
		self.UnsubscribeDetectionEvent()
		self.__unsubscribe_detection_event()

		if self.__pipeline_thread and self.__pipeline_thread.is_alive():
		self.UnsubscribePipelineError()
		self.__unsubscribe_pipeline_error()
		except Exception as e:
		LOGGER.exception(f'Error during disconnect: {str(e)}')

		@@ -120,23 +120,72 @@ class MorpheusApiDriver(_Driver):
		results.append(e)
		return results

		def GetState(self) -> List[Tuple[str, Any]]:
		url = self.__morpheus_root + '/restconf/data/naudit-morpheus:morpheus/state'
		with self.__lock:
		def GetState(self, blocking=False, terminate: Optional[threading.Event] = None) -> Iterator[Tuple[float, str, Any]]:
		while True:
		if self.__terminate.is_set(): break
		if terminate is not None and terminate.is_set(): break

		internal_state = self.__get_state()
		if internal_state is not None:
		timestamp = time.time()
		yield(timestamp, 'state', internal_state)

		pipeline_error_empty = False
		detection_event_empty = False

		try:
		response = requests.get(url, headers=self.__headers, timeout=self.__timeout, verify=False)
		if response.ok:
		state = response.json()
		result = []
		for key, value in state.items():
		result.append((key, value))
		return result
		return []
		error = self.__pipeline_error_queue.get(block=False, timeout=0.1)
		if error is not None:
		yield (error.get('eventTime', time.time()), 'pipeline_error', error.get('event'),)
		except queue.Empty:
		pipeline_error_empty = True

		try:
		event = self.__detection_queue.get(block=False, timeout=0.1)
		if event is not None:
		yield (event.get('eventTime', time.time()), 'detection_event', error.get('event'),)
		except queue.Empty:
		detection_event_empty = True

		if pipeline_error_empty and detection_event_empty:
		if blocking:
		continue
		return

		@metered_subclass_method(METRICS_POOL)
		def SubscribeState(self, subscriptions: List[Tuple[str, float, float]]) -> List[Union[bool,Exception]]:
		results = []
		rollback_stack = []
		operations = [
		(self.__subscribe_detection_event, self.__unsubscribe_detection_event),
		(self.__subscribe_pipeline_error, self.__unsubscribe_pipeline_error),
		(self.__start_pipeline, self.__stop_pipeline),
		]
		for i, (sub_op, unsub_op) in enumerate(operations):
		result = sub_op()
		reuslts.append(result)
		if isinstance(result, Exception):
		while rollback_stack:
		rollback_op = rollback_stack.pop()
		try:
		rollback_op()
		except Exception as e:
		LOGGER.exception(f'Error getting state: {str(e)}')
		return []
		LOGGER.exception(f'Error during subscription rollback operation: {e}')

		return results

		rollback_stack.append(unsub_op)
		return results

		def StartPipeline(self) -> Union[bool, Exception]:
		@metered_subclass_method(METRICS_POOL)
		def UnsubscribeState(self, subscriptions: List[Tuple[str,float,float]]) -> List[Union[bool, Exception]]:
		results = []
		results.append(self.__stop_pipeline())
		results.append(self.__unsubscribe_pipeline_error())
		results.append(self.__unsubscribe_detection_event())
		return results

		def __start_pipeline(self) -> Union[bool, Exception]:
		url = self.__morpheus_root + '/restconf/data/naudit-morpheus:morpheus/start'
		with self.__lock:
		try:
		@@ -147,7 +196,7 @@ class MorpheusApiDriver(_Driver):
		LOGGER.exception(f'Error starting pipeline: {e}')
		return e

		def StopPipeline(self) -> Union[bool, Exception]:
		def __stop_pipeline(self) -> Union[bool, Exception]:
		url = self.__morpheus_root + '/restconf/data/naudit-morpheus:morpheus/stop'
		with self.__lock:
		try:
		@@ -156,10 +205,9 @@ class MorpheusApiDriver(_Driver):
		return True
		except Exception as e:
		LOGGER.exception(f'Error stopping pipeline: {e}')
		return False
		return e

		@metered_subclass_method(METRICS_POOL)
		def SubscribeDetectionEvent(self) -> Union[bool, Exception]:
		def __subscribe_detection_event(self) -> Union[bool, Exception]:
		url = self.__morpheus_root + '/restconf/streams/naudit-morpheus:morpheus/detection-event'
		with self.__lock:
		try:
		@@ -174,8 +222,7 @@ class MorpheusApiDriver(_Driver):
		LOGGER.exception(f'Error subscribing to detection events: {str(e)}')
		return e

		@metered_subclass_method(METRICS_POOL)
		def UnsubscribeDetectionEvent(self) -> Union[bool, Exception]:
		def __unsubscribe_detection_event(self) -> Union[bool, Exception]:
		try:
		if self.__detection_thread and self.__detection_thread.is_alive():
		self.__detection_thread.join(timeout=5)
		@@ -184,20 +231,7 @@ class MorpheusApiDriver(_Driver):
		LOGGER.exception(f'Error unsubscribing from detection events: {str(e)}')
		return e

		def GetDetectionEvent(self, blocking=False, terminate : Optional[threading.Event] = None) -> Iterator[Dict]:
		while True:
		if self.__terminate.is_set(): break
		if terminate is not None and terminate.is_set(): break
		try:
		event = self.__detection_queue.get(block=blocking, timeout=0.1)
		if event is not None:
		yield event
		except queue.Empty:
		if blocking:
		continue
		return

		def SubscribePipelineError(self) -> Union[bool, Exception]:
		def __subscribe_pipeline_error(self) -> Union[bool, Exception]:
		url = self.__morpheus_root + '/restconf/streams/naudit-morpheus:morpheus/pipeline-error'
		with self.__lock:
		try:
		@@ -212,7 +246,7 @@ class MorpheusApiDriver(_Driver):
		LOGGER.exception(f'Error subscribing to pipeline errors: {str(e)}')
		return e

		def UnsubscribePipelineError(self) -> Union[bool, Exception]:
		def __unsubscribe_pipeline_error(self) -> Union[bool, Exception]:
		try:
		if self.__pipeline_error_thread and self.__pipeline_error_thread.is_alive():
		self.__pipeline_error_thread.join(timeout=5)
		@@ -221,18 +255,18 @@ class MorpheusApiDriver(_Driver):
		LOGGER.exception(f'Error unsubscribing from pipeline errors: {str(e)}')
		return e

		def GetPipelineError(self, blocking=False, terminate: Optional[threading.Event] = None) -> Iterator[Dict]:
		while True:
		if self.__terminate.is_set(): break
		if terminate is not None and terminate.is_set(): break
		def __get_state(self) -> Dict:
		url = self.__morpheus_root + '/restconf/data/naudit-morpheus:morpheus/state'
		with self.__lock:
		try:
		error = self.__pipeline_error_queue.get(block=blocking, timeout=0.1)
		if error is not None:
		yield error
		except queue.Empty:
		if blocking:
		continue
		return
		response = requests.get(url, headers=self.__headers, timeout=self.__timeout, verify=False)
		if response.ok:
		state = response.json()
		return state
		except Exception as e:
		LOGGER.exception(f'Error getting internal state: {e}')
		return None


		def __handle_notification_stream(self, url: str, queue: Queue[Any]) -> None:
		try:

src/device/tests/test_unitary_morpheus.py

+10 −26

Original line number	Diff line number	Diff line
		@@ -3,6 +3,7 @@ os.environ['DEVICE_EMULATED_ONLY'] = 'YES'

		# pylint: disable=wrong-import-position
		import json
		import threading
		import logging, pytest, time
		from typing import Dict, List
		from device.service.drivers.morpheus.MorpheusApiDriver import MorpheusApiDriver
		@@ -56,36 +57,19 @@ def test_set_config(driver: MorpheusApiDriver):
		assert results[0] is True, "Expected a succesfull result"

		def test_retrieve_state(driver: MorpheusApiDriver):
		results = driver.SubscribeState()

		assert all(isinstance(result, bool) and result for result in results), \
		f"Subscription error: {results}"

		state = driver.GetState()

		assert isinstance(state, list), "Expected a a list for initial config"
		assert isinstance(state, iter), "Expected an iterator for state"
		assert len(state) > 0, " State should not be empty"

		print_data("State", state)

		def test_pipeline(driver: MorpheusApiDriver):
		result = driver.StartPipeline()

		assert result is True

		result = driver.StopPipeline()

		assert result is True

		def test_subscription_detection(driver: MorpheusApiDriver):
		result = driver.SubscribeDetectionEvent()

		assert result is True

		result = driver.UnsubscribeDetectionEvent()

		assert result is True

		def test_subscription_error(driver: MorpheusApiDriver):
		result = driver.SubscribePipelineError()

		assert result is True

		result = driver.UnsubscribePipelineError()
		results = driver.UnsubscribeState()

		assert result is True
		assert all(isinstance(result, bool) and result for result in results), \
		f"Unsubscription error: {results}"