scripts/run_tests_locally-device-morpheus.sh

+#!/bin/bash
+
+PROJECTDIR=`pwd`
+
+cd $PROJECTDIR/src
+RCFILE=$PROJECTDIR/coverage/.coveragerc
+
+# Run unitary tests and analyze coverage of code at same time
+coverage run --rcfile=$RCFILE --append -m pytest -s --log-level=INFO --verbose \
+    device/tests/test_unitary_morpheus.py

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

+import logging, requests, threading, json
+from typing import Any, Iterator, List, Optional, Tuple, Union, Dict
+from queue import Queue
+from common.method_wrappers.Decorator import MetricsPool, metered_subclass_method
+from device.service.driver_api._Driver import _Driver
+
+LOGGER = logging.getLogger(__name__)
+
+DRIVER_NAME = 'morpheus'
+METRICS_POOL = MetricsPool('Device', 'Driver', labels={'driver': DRIVER_NAME})
+
+class MorpheusApiDriver(_Driver):
+    def __init__(self, address: str, port: int, **settings) -> None:
+        super().__init__(DRIVER_NAME, address, port, **settings)
+        self.__lock = threading.Lock()
+        self.__started = threading.Event()
+        self.__terminate = threading.Event()
+        scheme = self.settings.get('scheme', 'http')
+        self.__morpheus_root = '{:s}://{:s}:{:d}'.format(scheme, self.address, int(self.port))
+        self.__timeout = int(self.settings.get('timeout', 120))
+        self.__headers = {'Accept': 'application/yang-data+json', 'Content-Type': 'application/yang-data+json'}
+
+        self.__detection_thread = None
+        self.__pipeline_error_thread = None
+
+        size = self.settings.get('queue_events_size', 10)
+        self.__pipeline_error_queue = Queue(maxsize=size)
+        self.__detection_queue = Queue(maxsize=size)
+
+    def Connect(self) -> bool:
+        url = self.__morpheus_root + '/restconf/data/naudit-morpheus:morpheus'
+        with self.__lock:
+            if self.__started.is_set(): return True
+            try:
+                requests.get(url, headers=self.__headers, timeout=self.__timeout, verify=False)
+                self.__started.set()
+                return True
+            except requests.exceptions.Timeout:
+                LOGGER.exception('Timeout connecting {:s}'.format(str(self.__morpheus_root)))
+                return False
+            except Exception:  # pylint: disable=broad-except
+                LOGGER.exception('Exception connecting {:s}'.format(str(self.__morpheus_root)))
+                return False
+
+    def Disconnect(self) -> bool:
+        with self.__lock:
+            try:
+                if self.__detection_thread and self.__detection_thread.is_alive():
+                    self.UnsubscribeDetectionEvent()
+
+                if self.__pipeline_thread and self.__pipeline_thread.is_alive():
+                    self.UnsubscribePipelineError()
+            except Exception as e:
+                LOGGER.exception(f'Error during disconnect: {str(e)}')
+
+            self.__terminate.set()
+            return True
+
+    @metered_subclass_method(METRICS_POOL)
+    def GetInitialConfig(self) -> List[Tuple[str, Any]]:
+        url = self.__morpheus_root + '/restconf/data/naudit-morpheus:morpheus/config'
+        with self.__lock:
+            try:
+                response = requests.get(url, headers=self.__headers, timeout=self.__timeout, verify=False)
+                if response.ok:
+                    config = response.json()
+                    result = []
+
+                    for key, value in config.items():
+                        result.append((key, value))
+
+                    return  result
+            except Exception as e:
+                LOGGER.exception('Exception getting initial config {:s}'.format(str(self.__morpheus_root)))
+            return []
+
+    @metered_subclass_method(METRICS_POOL)
+    def GetConfig(self, resource_keys : List[str] = []) -> List[Tuple[str, Union[Any, None, Exception]]]:
+        url = self.__morpheus_root + '/restconf/data/naudit-morpheus:morpheus/config'
+        with self.__lock:
+            try:
+                response = requests.get(url, headers=self.__headers, timeout=self.__timeout, verify=False)
+                if response.ok:
+                    config = response.json()
+                    results = []
+
+                    if not resource_keys:
+                        for key, value in config.items():
+                            results.append((key, value))
+                        return results
+
+                    for key in resource_keys:
+                        try:
+                            results.append(config[key])
+                        except KeyError:
+                            results.append(None)
+                        except Exception as e:
+                            results.append(e)
+
+                    return results
+                return [(key, None) for key in resource_keys]
+            except Exception as e:
+                LOGGER.exception(f'Error getting config: {str(e)}')
+                return [(key, e) for key in resource_keys]
+
+    @metered_subclass_method(METRICS_POOL)
+    def SetConfig(self, resources: List[Tuple[str, Any]]) -> List[Union[bool, Exception]]:
+        url = self.__morpheus_root + '/restconf/data/naudit-morpheus:morpheus/config'
+        results = []
+        with self.__lock:
+            config = dict(resources)
+            try:
+                response = requests.put(url,
+                                        headers=self.__headers,
+                                        json=config,
+                                        timeout=self.__timeout,
+                                        verify=False)
+                results.append(response.ok)
+            except Exception as e:
+                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:
+            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 []
+            except Exception as e:
+                LOGGER.exception(f'Error getting state: {str(e)}')
+                return []
+
+    def StartPipeline(self) -> Union[bool, Exception]:
+        url = self.__morpheus_root + '/restconf/data/naudit-morpheus:morpheus/start'
+        with self.__lock:
+            try:
+                response = requests.post(url, headers=self.__headers, timeout=self.__timeout, verify=False)
+                response.raise_for_status()
+                return True
+            except Exception as e:
+                LOGGER.exception(f'Error starting pipeline: {e}')
+                return e
+
+    def StopPipeline(self) -> Union[bool, Exception]:
+        url = self.__morpheus_root + '/restconf/data/naudit-morpheus:morpheus/stop'
+        with self.__lock:
+            try:
+                response = requests.post(url, headers=self.__headers, timeout=self.__timeout, verify=False)
+                response.raise_for_status()
+                return True
+            except Exception as e:
+                LOGGER.exception(f'Error stopping pipeline: {e}')
+                return False
+
+    @metered_subclass_method(METRICS_POOL)
+    def SubscribeDetectionEvent(self) -> Union[bool, Exception]:
+        url = self.__morpheus_root + '/restconf/streams/naudit-morpheus:morpheus/detection-event'
+        with self.__lock:
+            try:
+                self.__detection_thread = threading.Thread(
+                        target=self.__handle_notification_stream,
+                        args=(url, self.__detection_queue),
+                        daemon=True
+                        )
+                self.__detection_thread.start()
+                return True
+            except Exception as e:
+                LOGGER.exception(f'Error subscribing to detection events: {str(e)}')
+                return e
+
+    @metered_subclass_method(METRICS_POOL)
+    def UnsubscribeDetectionEvent(self) -> Union[bool, Exception]:
+        try:
+            if self.__detection_thread and self.__detection_thread.is_alive():
+                self.__detection_thread.join(timeout=5)
+            return True
+        except Exception as e:
+            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]:
+        url = self.__morpheus_root + '/restconf/streams/naudit-morpheus:morpheus/pipeline-error'
+        with self.__lock:
+            try:
+                self.__pipeline_error_thread = threading.Thread(
+                        target=self.__handle_notification_stream,
+                        args=(url, self.__pipeline_error_queue),
+                        daemon=True
+                        )
+                self.__pipeline_error_thread.start()
+                return True
+            except Exception as e:
+                LOGGER.exception(f'Error subscribing to pipeline errors: {str(e)}')
+                return e
+
+    def UnsubscribePipelineError(self) -> Union[bool, Exception]:
+        try:
+            if self.__pipeline_error_thread and self.__pipeline_error_thread.is_alive():
+                self.__pipeline_error_thread.join(timeout=5)
+            return True
+        except Exception as e:
+            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
+            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
+
+    def __handle_notification_stream(self, url: str, queue: Queue[Any]) -> None:
+        try:
+            with requests.get(url,
+                              headers=self.__headers,
+                              stream=True,
+                              verify=False) as response:
+
+                if not response.ok:
+                    LOGGER.error(f'Error connecting to event stream: {response.status_code}')
+                    return
+
+                try:
+                    event = response.json()
+                    queue.put(event['data']['ietf-restconf:notification'])
+                except json.JSONDecodeError as e:
+                    LOGGER.error(f'Error parsing event: {e}')
+        except Exception as e:
+            LOGGER.exception(f'Error in notification stream handler: {str(e)}')

src/device/tests/test_unitary_morpheus.py

+import os
+os.environ['DEVICE_EMULATED_ONLY'] = 'YES'
+
+# pylint: disable=wrong-import-position
+import json
+import logging, pytest, time
+from typing import Dict, List
+from device.service.drivers.morpheus.MorpheusApiDriver import MorpheusApiDriver
+
+logging.basicConfig(level=logging.DEBUG)
+LOGGER = logging.getLogger(__name__)
+LOGGER.setLevel(logging.DEBUG)
+
+
+##### DRIVER FIXTURE ###################################################################################################
+
+DRIVER_SETTING_ADDRESS  = '127.0.0.1'
+DRIVER_SETTING_PORT     = 8090
+
+@pytest.fixture(scope='session')
+def driver() -> MorpheusApiDriver:
+    _driver = MorpheusApiDriver(
+        DRIVER_SETTING_ADDRESS, DRIVER_SETTING_PORT,
+    )
+    _driver.Connect()
+    yield _driver
+    time.sleep(1)
+    _driver.Disconnect()
+
+
+##### TEST METHODS #####################################################################################################
+
+def print_data(label, data):
+    print(f"{label}: {json.dumps(data, indent=2)}")
+
+def test_initial_config_retrieval(driver: MorpheusApiDriver):
+    config = driver.GetInitialConfig()
+
+    assert isinstance(config, list), "Expected a list for initial config"
+    assert len(config) > 0, "Initial config should not be empty"
+
+    print_data("Initial Config", config)
+
+def test_retrieve_config(driver: MorpheusApiDriver):
+    config = driver.GetConfig(None)
+
+    assert isinstance(config, list), "Expected a list for config"
+    assert len(config) > 0, "Config should not be empty"
+
+    print_data("Config", config)
+
+def test_set_config(driver: MorpheusApiDriver):
+    results = driver.SetConfig([('traffic_type', 'UDP')])
+
+    assert len(results) == 1, "Expected only one result"
+    assert results[0] is True, "Expected a succesfull result"
+
+def test_retrieve_state(driver: MorpheusApiDriver):
+    state = driver.GetState()
+
+    assert isinstance(state, list), "Expected a a list for initial config"
+    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()
+
+    assert result is True