st2tests.mocks.MockSensorService

Complexity

Total Complexity

13

Size/Duplication

Total Lines	93
Duplicated Lines	0 %

# Licensed to the StackStorm, Inc ('StackStorm') under one or more
class MockSensorService(SensorService):
    """
    Mock SensorService for use in testing.
    """

    def __init__(self, sensor_wrapper):
        self._sensor_wrapper = sensor_wrapper

        # Holds a mock logger instance
        # We use a Mock class so use can assert logger was called with particular arguments
        self._logger = Mock(spec=RootLogger)

        # Holds mock KeyValuePair objects
        # Key is a KeyValuePair name and value is the KeyValuePair object
        self._datastore_items = {}

        # Holds a list of triggers which were dispatched
        self.dispatched_triggers = []

    def get_logger(self, name):
        """
        Return mock logger instance.

        Keep in mind that this method returns Mock class instance which means you can use all the
        usual Mock class methods to assert that a particular message has been logged / logger has
        been called with particular arguments.
        """
        return self._logger

    def dispatch_with_context(self, trigger, payload=None, trace_context=None):
        item = {
            'trigger': trigger,
            'payload': payload,
            'trace_context': trace_context
        }
        self.dispatched_triggers.append(item)

    def list_values(self, local=True, prefix=None):
        """
        Return a list of all values stored in a dictionary which is local to this class.
        """
        key_prefix = self._get_full_key_prefix(local=local, prefix=prefix)

        if not key_prefix:
            return self._datastore_items.values()

        result = []
        for name, kvp in self._datastore_items.items():
            if name.startswith(key_prefix):
                result.append(kvp)

        return result

    def get_value(self, name, local=True):
        """
        Return a particular value stored in a dictionary which is local to this class.
        """
        name = self._get_full_key_name(name=name, local=local)

        if name not in self._datastore_items:
            return None

        kvp = self._datastore_items[name]
        return kvp.value

    def set_value(self, name, value, ttl=None, local=True):
        """
        Store a value in a dictionary which is local to this class.
        """
        if ttl:
            raise ValueError('MockSensorService.set_value doesn\'t support "ttl" argument')

        name = self._get_full_key_name(name=name, local=local)

        instance = KeyValuePair()
        instance.id = name
        instance.name = name
        instance.value = value

        self._datastore_items[name] = instance
        return True

    def delete_value(self, name, local=True):
        """
        Delete a value from a dictionary which is local to this class.
        """
        name = self._get_full_key_name(name=name, local=local)

        if name not in self._datastore_items:
            return False

        del self._datastore_items[name]
        return True