OpenStack Study: fake_driver.py

**** CubicPower OpenStack Study ****

# Copyright 2012 OpenStack Foundation

#

# Licensed under the Apache License, Version 2.0 (the "License"); you may

# not use this file except in compliance with the License. You may obtain

# a copy of the License at

#

# http://www.apache.org/licenses/LICENSE-2.0

#

# Unless required by applicable law or agreed to in writing, software

# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT

# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the

# License for the specific language governing permissions and limitations

# under the License.

from cinder.openstack.common import log as logging

from cinder.tests.brick.fake_lvm import FakeBrickLVM

from cinder.volume import driver

from cinder.volume.drivers import lvm

LOG = logging.getLogger(__name__)

**** CubicPower OpenStack Study ****

class FakeISCSIDriver(lvm.LVMISCSIDriver):

"""Logs calls instead of executing."""

**** CubicPower OpenStack Study ****

    def __init__(self, *args, **kwargs):

        super(FakeISCSIDriver, self).__init__(execute=self.fake_execute,

                                              *args, **kwargs)

        self.vg = FakeBrickLVM('cinder-volumes', False,

                               None, 'default',

                               self.fake_execute)

**** CubicPower OpenStack Study ****

    def check_for_setup_error(self):

        """No setup necessary in fake mode."""

        pass

**** CubicPower OpenStack Study ****

    def initialize_connection(self, volume, connector):

        volume_metadata = {}

        for metadata in volume['volume_admin_metadata']:

            volume_metadata[metadata['key']] = metadata['value']

        access_mode = volume_metadata.get('attached_mode')

        if access_mode is None:

            access_mode = ('ro'

                           if volume_metadata.get('readonly') == 'True'

                           else 'rw')

        return {

            'driver_volume_type': 'iscsi',

            'data': {'access_mode': access_mode}

        }

**** CubicPower OpenStack Study ****

    def terminate_connection(self, volume, connector, **kwargs):

        pass

    @staticmethod

**** CubicPower OpenStack Study ****

    def fake_execute(cmd, *_args, **_kwargs):

        """Execute that simply logs the command."""

        LOG.debug(_("FAKE ISCSI: %s"), cmd)

        return (None, None)

**** CubicPower OpenStack Study ****

class FakeISERDriver(FakeISCSIDriver):

"""Logs calls instead of executing."""

**** CubicPower OpenStack Study ****

    def __init__(self, *args, **kwargs):

        super(FakeISERDriver, self).__init__(execute=self.fake_execute,

                                             *args, **kwargs)

**** CubicPower OpenStack Study ****

    def initialize_connection(self, volume, connector):

        return {

            'driver_volume_type': 'iser',

            'data': {}

        }

    @staticmethod

**** CubicPower OpenStack Study ****

    def fake_execute(cmd, *_args, **_kwargs):

        """Execute that simply logs the command."""

        LOG.debug(_("FAKE ISER: %s"), cmd)

        return (None, None)

**** CubicPower OpenStack Study ****

class LoggingVolumeDriver(driver.VolumeDriver):

"""Logs and records calls, for unit tests."""

**** CubicPower OpenStack Study ****

    def check_for_setup_error(self):

        pass

**** CubicPower OpenStack Study ****

    def create_volume(self, volume):

        self.log_action('create_volume', volume)

**** CubicPower OpenStack Study ****

    def delete_volume(self, volume):

        self.clear_volume(volume)

        self.log_action('delete_volume', volume)

**** CubicPower OpenStack Study ****

    def clear_volume(self, volume):

        self.log_action('clear_volume', volume)

**** CubicPower OpenStack Study ****

    def local_path(self, volume):

        LOG.error(_("local_path not implemented"))

        raise NotImplementedError()

**** CubicPower OpenStack Study ****

    def ensure_export(self, context, volume):

        self.log_action('ensure_export', volume)

**** CubicPower OpenStack Study ****

    def create_export(self, context, volume):

        self.log_action('create_export', volume)

**** CubicPower OpenStack Study ****

    def remove_export(self, context, volume):

        self.log_action('remove_export', volume)

**** CubicPower OpenStack Study ****

    def initialize_connection(self, volume, connector):

        self.log_action('initialize_connection', volume)

**** CubicPower OpenStack Study ****

    def terminate_connection(self, volume, connector):

        self.log_action('terminate_connection', volume)

    _LOGS = []

    @staticmethod

**** CubicPower OpenStack Study ****

    def clear_logs():

        LoggingVolumeDriver._LOGS = []

    @staticmethod

**** CubicPower OpenStack Study ****

    def log_action(action, parameters):

        """Logs the command."""

        LOG.debug(_("LoggingVolumeDriver: %s") % (action))

        log_dictionary = {}

        if parameters:

            log_dictionary = dict(parameters)

        log_dictionary['action'] = action

        LOG.debug(_("LoggingVolumeDriver: %s") % (log_dictionary))

        LoggingVolumeDriver._LOGS.append(log_dictionary)

    @staticmethod

**** CubicPower OpenStack Study ****

    def all_logs():

        return LoggingVolumeDriver._LOGS

    @staticmethod

**** CubicPower OpenStack Study ****

    def logs_like(action, **kwargs):

        matches = []

        for entry in LoggingVolumeDriver._LOGS:

            if entry['action'] != action:

                continue

            match = True

            for k, v in kwargs.iteritems():

                if entry.get(k) != v:

                    match = False

                    break

            if match:

                matches.append(entry)

        return matches