build.storehouse

Complexity

Total Complexity

24

Size/Duplication

Total Lines	131
Duplicated Lines	0 %

Test Coverage

Coverage

100%

Importance

Changes

0

10 Methods

Rating	Name	Duplication	Size	Complexity
A	StoreHouse.__init__()	0	9	3
A	StoreHouse.get_data()	0	11	4
A	StoreHouse.get_stored_box()	0	9	1
A	StoreHouse.create_box()	0	7	1
A	StoreHouse._get_or_create_a_box_from_list_of_boxes()	0	6	2
A	StoreHouse._get_box_callback()	0	6	2
A	StoreHouse.list_stored_boxes()	0	8	1
A	StoreHouse._create_box_callback()	0	7	2
A	StoreHouse.__new__()	0	9	2
B	StoreHouse.save_status()	0	37	6

"""Module to handle the storehouse."""
import threading
import time

from kytos.core import log
from kytos.core.events import KytosEvent
from napps.kytos.topology import settings


class StoreHouse:
    """Class to handle storehouse."""

    @classmethod
    def __new__(cls, *args, **kwargs):
        # pylint: disable=unused-argument
        """Make this class a Singleton."""
        instance = cls.__dict__.get("__instance__")
        if instance is not None:
            return instance
        cls.__instance__ = instance = object.__new__(cls)
        return instance

    def __init__(self, controller):
        """Create a storehouse client instance."""
        self.controller = controller
        self.namespace = 'kytos.topology.status'
        if '_lock' not in self.__dict__:
            self._lock = threading.Lock()
        if 'box' not in self.__dict__:
            self.box = None
        self.list_stored_boxes()

    def get_data(self):
        """Return the persistence box data."""
        # Wait for box retrieve from storehouse
        i = 0
        while not self.box and i < settings.STOREHOUSE_TIMEOUT:
            time.sleep(settings.STOREHOUSE_WAIT_INTERVAL)
            i += settings.STOREHOUSE_WAIT_INTERVAL
        if not self.box:
            error = 'Error retrieving persistence box from storehouse.'
            raise FileNotFoundError(error)
        return self.box.data

    def create_box(self):
        """Create a persistence box to store administrative changes."""
        content = {'namespace': self.namespace,
                   'callback': self._create_box_callback,
                   'data': {}}
        event = KytosEvent(name='kytos.storehouse.create', content=content)
        self.controller.buffers.app.put(event)

    def _create_box_callback(self, _event, data, error):
        """Execute the callback to handle create_box."""
        if error:
            log.error(f'Can\'t create persistence'
                      f'box with namespace {self.namespace}')

        self.box = data

    def list_stored_boxes(self):
        """List all persistence box stored in storehouse."""
        name = 'kytos.storehouse.list'
        content = {'namespace': self.namespace,
                   'callback': self._get_or_create_a_box_from_list_of_boxes}

        event = KytosEvent(name=name, content=content)
        self.controller.buffers.app.put(event)

    def _get_or_create_a_box_from_list_of_boxes(self, _event, data, _error):
        """Create a persistence box or retrieve the stored box."""
        if data:
            self.get_stored_box(data[0])
        else:
            self.create_box()

    def get_stored_box(self, box_id):
        """Get persistence box from storehouse."""
        content = {'namespace': self.namespace,
                   'callback': self._get_box_callback,
                   'box_id': box_id,
                   'data': {}}
        name = 'kytos.storehouse.retrieve'
        event = KytosEvent(name=name, content=content)
        self.controller.buffers.app.put(event)

    def _get_box_callback(self, _event, data, error):
        """Handle get_box method saving the box or logging with the error."""
        if error:
            log.error('Persistence box not found.')

        self.box = data

    def save_status(self, status):
        """Save the network administrative status using storehouse."""
        storehouse_done = False
        storehouse_error = None

        def _storehouse_callback(event, data, error):
            """Handle storehouse update result."""
            # pylint: disable=unused-argument
            nonlocal storehouse_done, storehouse_error

The variable storehouse_error does not seem to be defined.

The variable storehouse_done does not seem to be defined.

            storehouse_done = True
            storehouse_error = error

        with self._lock:
            self.box.data[status.get('id')] = status
            content = {'namespace': self.namespace,
                       'box_id': self.box.box_id,
                       'data': self.box.data,
                       'callback': _storehouse_callback}
            event = KytosEvent(name='kytos.storehouse.update',
                               content=content)
            self.controller.buffers.app.put(event)

            i = 0
            while not storehouse_done and i < settings.STOREHOUSE_TIMEOUT:
                time.sleep(settings.STOREHOUSE_WAIT_INTERVAL)
                i += settings.STOREHOUSE_WAIT_INTERVAL

            if not storehouse_done:
                storehouse_error = 'Timeout while waiting for storehouse'

            if storehouse_error:
                log.error('Fail to update persistence box in '
                          f'{self.namespace}.{self.box.box_id}. '
                          f'Error: {storehouse_error}')
                return

        log.info('Network administrative status saved in '
                 f'{self.namespace}.{self.box.box_id}')