"""
import eventlet
+
eventlet.monkey_patch()
# See http://code.google.com/p/python-nose/issues/detail?id=373
import os
import shutil
+from oslo.config import cfg
+
from cinder.db.sqlalchemy.api import get_engine
-from cinder import flags
-FLAGS = flags.FLAGS
_DB = None
+CONF = cfg.CONF
+
def reset_db():
- if FLAGS.database.connection == "sqlite://":
+ if CONF.database.connection == "sqlite://":
engine = get_engine()
engine.dispose()
conn = engine.connect()
conn.connection.executescript(_DB)
else:
- shutil.copyfile(os.path.join(FLAGS.state_path, FLAGS.sqlite_clean_db),
- os.path.join(FLAGS.state_path, FLAGS.sqlite_db))
+ shutil.copyfile(os.path.join(CONF.state_path,
+ CONF.sqlite_clean_db),
+ os.path.join(CONF.state_path, CONF.sqlite_db))
def setup():
from cinder.db import migration
from cinder.tests import fake_flags
- fake_flags.set_defaults(FLAGS)
+ fake_flags.set_defaults(CONF)
- if FLAGS.database.connection == "sqlite://":
+ if CONF.database.connection == "sqlite://":
if migration.db_version() > 1:
return
else:
- testdb = os.path.join(FLAGS.state_path, FLAGS.sqlite_db)
+ testdb = os.path.join(CONF.state_path, CONF.sqlite_db)
if os.path.exists(testdb):
return
migration.db_sync()
- if FLAGS.database.connection == "sqlite://":
+ if CONF.database.connection == "sqlite://":
global _DB
engine = get_engine()
conn = engine.connect()
_DB = "".join(line for line in conn.connection.iterdump())
else:
- cleandb = os.path.join(FLAGS.state_path, FLAGS.sqlite_clean_db)
+ cleandb = os.path.join(CONF.state_path, CONF.sqlite_clean_db)
shutil.copyfile(testdb, cleandb)
# License for the specific language governing permissions and limitations
# under the License.
+
from lxml import etree
import webob
from cinder.api.contrib import extended_snapshot_attributes
from cinder import exception
-from cinder import flags
from cinder.openstack.common import jsonutils
from cinder import test
from cinder.tests.api import fakes
from cinder import volume
-FLAGS = flags.FLAGS
-
-
UUID1 = '00000000-0000-0000-0000-000000000001'
UUID2 = '00000000-0000-0000-0000-000000000002'
from cinder.api.contrib import hosts as os_hosts
from cinder import context
from cinder import db
-from cinder import flags
from cinder.openstack.common import log as logging
from cinder.openstack.common import timeutils
from cinder import test
-FLAGS = flags.FLAGS
LOG = logging.getLogger(__name__)
created_time = datetime.datetime(2012, 11, 14, 1, 20, 41, 95099)
curr_time = timeutils.utcnow()
from cinder.api.contrib import volume_actions
from cinder import exception
-from cinder import flags
from cinder.openstack.common import jsonutils
from cinder.openstack.common.rpc import common as rpc_common
from cinder import test
from cinder.volume import api as volume_api
-FLAGS = flags.FLAGS
-
-
def fake_volume_api(*args, **kwargs):
return True
# under the License.
import StringIO
+
+from oslo.config import cfg
import webob
from cinder.api.middleware import sizelimit
-from cinder import flags
from cinder import test
-FLAGS = flags.FLAGS
-MAX_REQUEST_BODY_SIZE = FLAGS.osapi_max_request_body_size
+
+CONF = cfg.CONF
+
+MAX_REQUEST_BODY_SIZE = CONF.osapi_max_request_body_size
class TestLimitingReader(test.TestCase):
import iso8601
from lxml import etree
+from oslo.config import cfg
import webob
from cinder.api.v1 import router
from cinder.api import xmlutil
-from cinder import flags
from cinder.openstack.common import jsonutils
from cinder import test
-FLAGS = flags.FLAGS
+
NS = "{http://docs.openstack.org/common/api/v1.0}"
+CONF = cfg.CONF
+
+
class ExtensionTestCase(test.TestCase):
def setUp(self):
super(ExtensionTestCase, self).setUp()
- ext_list = FLAGS.osapi_volume_extension[:]
+ ext_list = CONF.osapi_volume_extension[:]
fox = ('cinder.tests.api.extensions.foxinsocks.Foxinsocks')
if fox not in ext_list:
ext_list.append(fox)
from cinder.api.v1 import snapshots
from cinder.api.v1 import volumes
from cinder.api import versions
-from cinder import flags
from cinder.openstack.common import log as logging
from cinder import test
from cinder.tests.api import fakes
-FLAGS = flags.FLAGS
LOG = logging.getLogger(__name__)
from cinder.api.v1 import snapshots
from cinder import db
from cinder import exception
-from cinder import flags
from cinder.openstack.common import log as logging
from cinder import test
from cinder.tests.api import fakes
from cinder import volume
-FLAGS = flags.FLAGS
LOG = logging.getLogger(__name__)
UUID = '00000000-0000-0000-0000-000000000001'
import datetime
from lxml import etree
+from oslo.config import cfg
import webob
from cinder.api import extensions
from cinder import context
from cinder import db
from cinder import exception
-from cinder import flags
from cinder import test
from cinder.tests.api import fakes
from cinder.tests.api.v2 import stubs
from cinder.volume import api as volume_api
-FLAGS = flags.FLAGS
NS = '{http://docs.openstack.org/volume/api/v1}'
TEST_SNAPSHOT_UUID = '00000000-0000-0000-0000-000000000001'
+CONF = cfg.CONF
+
def stub_snapshot_get(self, context, snapshot_id):
if snapshot_id != TEST_SNAPSHOT_UUID:
self.assertEqual(res_dict, expected)
def test_volume_create_with_type(self):
- vol_type = FLAGS.default_volume_type
+ vol_type = CONF.default_volume_type
db.volume_type_create(context.get_admin_context(),
dict(name=vol_type, extra_specs={}))
db_vol_type = db.volume_type_get_by_name(context.get_admin_context(),
from cinder.api.v2 import snapshots
from cinder import db
from cinder import exception
-from cinder import flags
from cinder.openstack.common import log as logging
from cinder import test
from cinder.tests.api import fakes
from cinder import volume
-FLAGS = flags.FLAGS
LOG = logging.getLogger(__name__)
UUID = '00000000-0000-0000-0000-000000000001'
# License for the specific language governing permissions and limitations
# under the License.
+
import datetime
from lxml import etree
+from oslo.config import cfg
import webob
from cinder.api import extensions
from cinder import context
from cinder import db
from cinder import exception
-from cinder import flags
from cinder import test
from cinder.tests.api import fakes
from cinder.tests.api.v2 import stubs
from cinder.volume import api as volume_api
-FLAGS = flags.FLAGS
+CONF = cfg.CONF
+
NS = '{http://docs.openstack.org/api/openstack-volume/2.0/content}'
TEST_SNAPSHOT_UUID = '00000000-0000-0000-0000-000000000001'
self.assertEqual(res_dict, expected)
def test_volume_create_with_type(self):
- vol_type = db.volume_type_create(context.get_admin_context(),
- dict(name=FLAGS.default_volume_type,
- extra_specs={}))
+ vol_type = db.volume_type_create(
+ context.get_admin_context(),
+ dict(name=CONF.default_volume_type, extra_specs={})
+ )
db_vol_type = db.volume_type_get(context.get_admin_context(),
vol_type.id)
# License for the specific language governing permissions and limitations
# under the License.
+
from oslo.config import cfg
-from cinder import flags
-FLAGS = flags.FLAGS
-FLAGS.register_opt(cfg.IntOpt('answer', default=42, help='test flag'))
+CONF = cfg.CONF
+CONF.register_opt(cfg.IntOpt('answer', default=42, help='test flag'))
import uuid
from cinder import exception
-from cinder import flags
import cinder.image.glance
from cinder.openstack.common import log as logging
LOG = logging.getLogger(__name__)
-FLAGS = flags.FLAGS
-
-
class _FakeImageService(object):
"""Mock (fake) image service for unit testing."""
import datetime
import glanceclient.exc
+from glanceclient.v2.client import Client as glanceclient_v2
+from oslo.config import cfg
from cinder import context
from cinder import exception
-from cinder import flags
from cinder.image import glance
from cinder import test
from cinder.tests.glance import stubs as glance_stubs
-from glanceclient.v2.client import Client as glanceclient_v2
-FLAGS = flags.FLAGS
+CONF = cfg.CONF
class NullWriter(object):
9292)
self.assertEquals(client_wrapper_v2.client.__module__,
'glanceclient.v2.client')
- FLAGS.reset()
+ CONF.reset()
def test_glance_version_by_arg(self):
"""Test glance version set by arg to GlanceClientWrapper"""
# License for the specific language governing permissions and limitations
# under the License.
-from cinder import flags
+
+from oslo.config import cfg
+
from cinder.openstack.common import log as logging
from cinder.tests.integrated import integrated_helpers
-FLAGS = flags.FLAGS
+CONF = cfg.CONF
+
LOG = logging.getLogger(__name__)
class ExtensionsTest(integrated_helpers._IntegratedTestBase):
def _get_flags(self):
f = super(ExtensionsTest, self)._get_flags()
- f['osapi_volume_extension'] = FLAGS.osapi_volume_extension[:]
+ f['osapi_volume_extension'] = CONF.osapi_volume_extension[:]
f['osapi_volume_extension'].append(
'cinder.tests.api.extensions.foxinsocks.Foxinsocks')
return f
# License for the specific language governing permissions and limitations
# under the License.
+
from oslo.config import cfg
-from cinder import flags
-FLAGS = flags.FLAGS
-FLAGS.register_opt(cfg.IntOpt('runtime_answer', default=54, help='test flag'))
+CONF = cfg.CONF
+CONF.register_opt(cfg.IntOpt('runtime_answer', default=54, help='test flag'))
Tests For HostManager
"""
+from oslo.config import cfg
from cinder import db
from cinder import exception
-from cinder import flags
from cinder.openstack.common.scheduler import filters
from cinder.openstack.common import timeutils
from cinder.scheduler import host_manager
from cinder.tests.scheduler import fakes
-FLAGS = flags.FLAGS
+CONF = cfg.CONF
class FakeFilterClass1(filters.BaseHostFilter):
def test_get_all_host_states(self):
context = 'fake_context'
- topic = FLAGS.volume_topic
+ topic = CONF.volume_topic
self.mox.StubOutWithMock(db, 'service_get_all_by_topic')
self.mox.StubOutWithMock(host_manager.LOG, 'warn')
Unit Tests for cinder.scheduler.rpcapi
"""
+
+from oslo.config import cfg
+
from cinder import context
-from cinder import flags
from cinder.openstack.common import rpc
from cinder.scheduler import rpcapi as scheduler_rpcapi
from cinder import test
-FLAGS = flags.FLAGS
+CONF = cfg.CONF
class SchedulerRpcAPITestCase(test.TestCase):
retval = getattr(rpcapi, method)(ctxt, **kwargs)
self.assertEqual(retval, expected_retval)
- expected_args = [ctxt, FLAGS.scheduler_topic, expected_msg]
+ expected_args = [ctxt, CONF.scheduler_topic, expected_msg]
for arg, expected_arg in zip(self.fake_args, expected_args):
self.assertEqual(arg, expected_arg)
Tests For Scheduler
"""
-
from cinder import context
from cinder import db
from cinder import exception
-from cinder import flags
from cinder.openstack.common import timeutils
from cinder.scheduler import driver
from cinder.scheduler import manager
from cinder import utils
-FLAGS = flags.FLAGS
-
-
class SchedulerManagerTestCase(test.TestCase):
"""Test case for scheduler manager."""
import tempfile
+from oslo.config import cfg
+
from cinder import context
from cinder import db
from cinder import exception
-from cinder import flags
from cinder.openstack.common import importutils
from cinder.openstack.common import log as logging
from cinder.openstack.common import timeutils
from cinder import test
-FLAGS = flags.FLAGS
+
+CONF = cfg.CONF
+
LOG = logging.getLogger(__name__)
self.flags(connection_type='fake',
volumes_dir=vol_tmpdir)
self.backup_mgr = \
- importutils.import_object(FLAGS.backup_manager)
+ importutils.import_object(CONF.backup_manager)
self.backup_mgr.host = 'testhost'
self.ctxt = context.get_admin_context()
backup['container'] = container
backup['status'] = status
backup['fail_reason'] = ''
- backup['service'] = FLAGS.backup_service
+ backup['service'] = CONF.backup_service
backup['size'] = size
backup['object_count'] = object_count
return db.backup_create(self.ctxt, backup)['id']
import tempfile
import zlib
+from swiftclient import client as swift
+
from cinder.backup.services.swift import SwiftBackupService
from cinder import context
from cinder import db
from cinder import exception
-from cinder import flags
from cinder.openstack.common import log as logging
from cinder import test
from cinder.tests.backup.fake_swift_client import FakeSwiftClient
-from swiftclient import client as swift
-FLAGS = flags.FLAGS
LOG = logging.getLogger(__name__)
"""Unit tests for cinder.db.api."""
+import datetime
+
+from oslo.config import cfg
+
from cinder import context
from cinder import db
from cinder import exception
-from cinder import flags
from cinder.openstack.common import uuidutils
from cinder.quota import ReservableResource
from cinder import test
-from datetime import datetime
-from datetime import timedelta
-FLAGS = flags.FLAGS
+
+CONF = cfg.CONF
def _quota_reserve(context, project_id):
resource,
get_sync(resource, i), 'quota_res_%d' % i)
deltas[resource] = i
- return db.quota_reserve(context, resources, quotas, deltas,
- datetime.utcnow(), datetime.utcnow(),
- timedelta(days=1), project_id)
+ return db.quota_reserve(
+ context, resources, quotas, deltas,
+ datetime.datetime.utcnow(), datetime.datetime.utcnow(),
+ datetime.timedelta(days=1), project_id
+ )
class ModelsObjectComparatorMixin(object):
def test_service_get_all_volume_sorted(self):
values = [
- ({'host': 'h1', 'binary': 'a', 'topic': FLAGS.volume_topic}, 100),
- ({'host': 'h2', 'binary': 'b', 'topic': FLAGS.volume_topic}, 200),
- ({'host': 'h3', 'binary': 'b', 'topic': FLAGS.volume_topic}, 300)]
+ ({'host': 'h1', 'binary': 'a', 'topic': CONF.volume_topic},
+ 100),
+ ({'host': 'h2', 'binary': 'b', 'topic': CONF.volume_topic},
+ 200),
+ ({'host': 'h3', 'binary': 'b', 'topic': CONF.volume_topic},
+ 300)]
services = []
for vals, size in values:
services.append(self._create_service(vals))
def setUp(self):
super(DBAPIReservationTestCase, self).setUp()
- self.values = {'uuid': 'sample-uuid',
- 'project_id': 'project1',
- 'resource': 'resource',
- 'delta': 42,
- 'expire': datetime.utcnow() + timedelta(days=1),
- 'usage': {'id': 1}}
+ self.values = {
+ 'uuid': 'sample-uuid',
+ 'project_id': 'project1',
+ 'resource': 'resource',
+ 'delta': 42,
+ 'expire': (datetime.datetime.utcnow() +
+ datetime.timedelta(days=1)),
+ 'usage': {'id': 1}
+ }
def test_reservation_create(self):
reservation = db.reservation_create(self.ctxt, **self.values)
'project1'))
def test_reservation_expire(self):
- self.values['expire'] = datetime.utcnow() + timedelta(days=1)
+ self.values['expire'] = datetime.datetime.utcnow() + \
+ datetime.timedelta(days=1)
reservations = _quota_reserve(self.ctxt, 'project1')
db.reservation_expire(self.ctxt)
# License for the specific language governing permissions and limitations
# under the License.
+
+from oslo.config import cfg
+
from cinder import context
-from cinder import flags
from cinder.openstack.common import importutils
from cinder import test
from cinder.volume.drivers.solidfire import SolidFire
-FLAGS = flags.FLAGS
+
+CONF = cfg.CONF
RBD_MODULE = "cinder.volume.drivers.rbd.RBDDriver"
SHEEPDOG_MODULE = "cinder.volume.drivers.sheepdog.SheepdogDriver"
def setUp(self):
super(VolumeDriverCompatibility, self).setUp()
- self.manager = importutils.import_object(FLAGS.volume_manager)
+ self.manager = importutils.import_object(CONF.volume_manager)
self.context = context.get_admin_context()
def tearDown(self):
# License for the specific language governing permissions and limitations
# under the License.
-import mox
import os
import shutil
import tempfile
from xml.dom.minidom import Document
+import mox
+
from cinder import exception
-from cinder import flags
from cinder.openstack.common import log as logging
from cinder import test
from cinder.volume import configuration as conf
from cinder.volume.drivers.emc.emc_smis_common import EMCSMISCommon
from cinder.volume.drivers.emc.emc_smis_iscsi import EMCSMISISCSIDriver
+
CINDER_EMC_CONFIG_FILE = '/etc/cinder/cinder_emc_config.xml'
LOG = logging.getLogger(__name__)
# under the License.
"""Unit tests for the GlusterFS driver module."""
-import __builtin__
+
import errno
import os
from cinder import exception
from cinder.exception import ProcessExecutionError
from cinder import test
-
from cinder.volume import configuration as conf
from cinder.volume.drivers import glusterfs
import base64
import urllib2
-import cinder.flags
+from oslo.config import cfg
+
from cinder import test
from cinder.volume.drivers import nexenta
from cinder.volume.drivers.nexenta import jsonrpc
from cinder.volume.drivers.nexenta import volume
-FLAGS = cinder.flags.FLAGS
+
+CONF = cfg.CONF
class TestNexentaDriver(test.TestCase):
self.assertEquals(
retval,
{'provider_location':
- '%s:%s,1 %s%s 0' % (FLAGS.nexenta_host,
- FLAGS.nexenta_iscsi_target_portal_port,
- FLAGS.nexenta_target_prefix,
+ '%s:%s,1 %s%s 0' % (CONF.nexenta_host,
+ CONF.nexenta_iscsi_target_portal_port,
+ CONF.nexenta_target_prefix,
self.TEST_VOLUME_NAME)})
def __get_test(i):
'available': '5368709120G',
'health': 'ONLINE'}
self.nms_mock.volume.get_child_props(
- FLAGS.nexenta_volume,
+ CONF.nexenta_volume,
'health|size|used|available').AndReturn(stats)
self.mox.ReplayAll()
stats = self.drv.get_volume_stats(True)
import StringIO
import urllib2
+from oslo.config import cfg
+
from cinder import context
from cinder import exception
-from cinder import flags
import cinder.openstack.common.policy
from cinder.openstack.common import policy as common_policy
from cinder import policy
from cinder import test
from cinder import utils
-FLAGS = flags.FLAGS
+
+CONF = cfg.CONF
class PolicyFileTestCase(test.TestCase):
rules = {
'context_is_admin': [["role:administrator"], ["role:johnny-admin"]]
}
- brain = common_policy.Brain(rules, FLAGS.policy_default_rule)
+ brain = common_policy.Brain(rules, CONF.policy_default_rule)
common_policy.set_brain(brain)
ctx = context.RequestContext('fake', 'fake', roles=['johnny-admin'])
self.assert_(ctx.is_admin)
"admin_or_owner": [["role:admin"], ["project_id:%(project_id)s"]],
"default": [["rule:admin_or_owner"]],
}
- brain = common_policy.Brain(rules, FLAGS.policy_default_rule)
+ brain = common_policy.Brain(rules, CONF.policy_default_rule)
common_policy.set_brain(brain)
ctx = context.RequestContext('fake', 'fake')
self.assertFalse(ctx.is_admin)
# License for the specific language governing permissions and limitations
# under the License.
+
import datetime
+from oslo.config import cfg
+
from cinder import context
from cinder import db
from cinder.db.sqlalchemy import api as sqa_api
from cinder.db.sqlalchemy import models as sqa_models
from cinder import exception
-from cinder import flags
from cinder.openstack.common import rpc
from cinder.openstack.common import timeutils
from cinder import quota
from cinder import volume
-FLAGS = flags.FLAGS
+CONF = cfg.CONF
class QuotaIntegrationTestCase(test.TestCase):
def test_too_many_volumes(self):
volume_ids = []
- for i in range(FLAGS.quota_volumes):
+ for i in range(CONF.quota_volumes):
vol_ref = self._create_volume()
volume_ids.append(vol_ref['id'])
self.assertRaises(exception.QuotaError,
# Make sure the snapshot volume_size isn't included in usage.
vol_type = db.volume_type_create(self.context,
- dict(name=FLAGS.default_volume_type))
+ dict(name=CONF.default_volume_type))
vol_ref2 = volume.API().create(self.context, 10, '', '')
usages = db.quota_usage_get_all_by_project(self.context,
self.project_id)
Unit Tests for remote procedure calls using queue
"""
+
import mox
from oslo.config import cfg
from cinder import context
from cinder import db
from cinder import exception
-from cinder import flags
from cinder import manager
from cinder import service
from cinder import test
from cinder import wsgi
+
test_service_opts = [
cfg.StrOpt("fake_manager",
default="cinder.tests.test_service.FakeManager",
default=0,
help="Port number to bind test service to"), ]
-flags.FLAGS.register_opts(test_service_opts)
+CONF = cfg.CONF
+CONF.register_opts(test_service_opts)
class FakeManager(manager.Manager):
# License for the specific language governing permissions and limitations
# under the License.
+
import __builtin__
import datetime
import hashlib
import os
-import os.path
import paramiko
import StringIO
import tempfile
import uuid
import mox
+from oslo.config import cfg
import cinder
from cinder import exception
-from cinder import flags
-from cinder.openstack.common import strutils
from cinder.openstack.common import timeutils
from cinder import test
from cinder import utils
-FLAGS = flags.FLAGS
+CONF = cfg.CONF
class ExecuteTestCase(test.TestCase):
def test_generate_glance_url(self):
generated_url = utils.generate_glance_url()
- actual_url = "http://%s:%d" % (FLAGS.glance_host, FLAGS.glance_port)
+ actual_url = "http://%s:%d" % (CONF.glance_host,
+ CONF.glance_port)
self.assertEqual(generated_url, actual_url)
def test_read_cached_file(self):
import datetime
import os
-
-import mox
import shutil
import tempfile
+import mox
+from oslo.config import cfg
+
from cinder.brick.iscsi import iscsi
from cinder import context
from cinder import db
from cinder import exception
-from cinder import flags
from cinder.image import image_utils
from cinder.openstack.common import importutils
from cinder.openstack.common.notifier import api as notifier_api
from cinder.volume import configuration as conf
from cinder.volume import driver
+
QUOTAS = quota.QUOTAS
-FLAGS = flags.FLAGS
+
+CONF = cfg.CONF
class VolumeTestCase(test.TestCase):
self.flags(connection_type='fake',
volumes_dir=vol_tmpdir,
notification_driver=[test_notifier.__name__])
- self.volume = importutils.import_object(FLAGS.volume_manager)
+ self.volume = importutils.import_object(CONF.volume_manager)
self.context = context.get_admin_context()
self.stubs.Set(iscsi.TgtAdm, '_get_target', self.fake_get_target)
fake_image.stub_out_image_service(self.stubs)
def tearDown(self):
try:
- shutil.rmtree(FLAGS.volumes_dir)
+ shutil.rmtree(CONF.volumes_dir)
except OSError:
pass
notifier_api._reset_drivers()
vol['image_id'] = image_id
vol['user_id'] = 'fake'
vol['project_id'] = 'fake'
- vol['availability_zone'] = FLAGS.storage_availability_zone
+ vol['availability_zone'] = CONF.storage_availability_zone
vol['status'] = status
vol['attach_status'] = "detached"
- vol['host'] = FLAGS.host
+ vol['host'] = CONF.host
if metadata is not None:
vol['metadata'] = metadata
return db.volume_create(context.get_admin_context(), vol)
self.assert_(iscsi_target not in targets)
targets.append(iscsi_target)
- total_slots = FLAGS.iscsi_num_targets
+ total_slots = CONF.iscsi_num_targets
for _index in xrange(total_slots):
self._create_volume()
for volume_id in volume_ids:
vol_tmpdir = tempfile.mkdtemp()
self.flags(volume_driver=self.driver_name,
volumes_dir=vol_tmpdir)
- self.volume = importutils.import_object(FLAGS.volume_manager)
+ self.volume = importutils.import_object(CONF.volume_manager)
self.context = context.get_admin_context()
self.output = ""
self.stubs.Set(iscsi.TgtAdm, '_get_target', self.fake_get_target)
def tearDown(self):
try:
- shutil.rmtree(FLAGS.volumes_dir)
+ shutil.rmtree(CONF.volumes_dir)
except OSError:
pass
super(DriverTestCase, self).tearDown()
iscsi_driver = driver.ISCSIDriver(configuration=configuration)
iscsi_driver._execute = lambda *a, **kw: \
- ("%s dummy" % FLAGS.iscsi_ip_address, '')
+ ("%s dummy" % CONF.iscsi_ip_address, '')
volume = {"name": "dummy",
"host": "0.0.0.0"}
iscsi_driver._do_iscsi_discovery(volume)
"""Tests for the configuration wrapper in volume drivers."""
+
from oslo.config import cfg
-from cinder import flags
from cinder.openstack.common import log as logging
from cinder import test
from cinder.volume import configuration
-from cinder.volume import driver
LOG = logging.getLogger(__name__)
-FLAGS = flags.FLAGS
volume_opts = [
cfg.IntOpt('int_opt', default=1),
]
-FLAGS.register_opts(volume_opts)
-FLAGS.register_opts(more_volume_opts)
+CONF = cfg.CONF
+CONF.register_opts(volume_opts)
+CONF.register_opts(more_volume_opts)
class VolumeConfigurationTest(test.TestCase):
def test_group_grafts_opts(self):
c = configuration.Configuration(volume_opts, config_group='foo')
- self.assertEquals(c.str_opt, FLAGS.foo.str_opt)
- self.assertEquals(c.bool_opt, FLAGS.foo.bool_opt)
+ self.assertEquals(c.str_opt, CONF.foo.str_opt)
+ self.assertEquals(c.bool_opt, CONF.foo.bool_opt)
def test_opts_no_group(self):
c = configuration.Configuration(volume_opts)
- self.assertEquals(c.str_opt, FLAGS.str_opt)
- self.assertEquals(c.bool_opt, FLAGS.bool_opt)
+ self.assertEquals(c.str_opt, CONF.str_opt)
+ self.assertEquals(c.bool_opt, CONF.bool_opt)
def test_grafting_multiple_opts(self):
c = configuration.Configuration(volume_opts, config_group='foo')
c.append_config_values(more_volume_opts)
- self.assertEquals(c.str_opt, FLAGS.foo.str_opt)
- self.assertEquals(c.bool_opt, FLAGS.foo.bool_opt)
- self.assertEquals(c.int_opt, FLAGS.foo.int_opt)
+ self.assertEquals(c.str_opt, CONF.foo.str_opt)
+ self.assertEquals(c.bool_opt, CONF.foo.bool_opt)
+ self.assertEquals(c.int_opt, CONF.foo.int_opt)
def test_safe_get(self):
c = configuration.Configuration(volume_opts, config_group='foo')
"""
+from oslo.config import cfg
+
from cinder import context
from cinder import db
-from cinder import flags
from cinder.openstack.common import jsonutils
from cinder.openstack.common import rpc
from cinder import test
from cinder.volume import rpcapi as volume_rpcapi
-FLAGS = flags.FLAGS
+CONF = cfg.CONF
class VolumeRpcAPITestCase(test.TestCase):
self.context = context.get_admin_context()
vol = {}
vol['host'] = 'fake_host'
- vol['availability_zone'] = FLAGS.storage_availability_zone
+ vol['availability_zone'] = CONF.storage_availability_zone
vol['status'] = "available"
vol['attach_status'] = "detached"
volume = db.volume_create(self.context, vol)
host = kwargs['host']
else:
host = kwargs['volume']['host']
- expected_topic = '%s.%s' % (FLAGS.volume_topic, host)
+ expected_topic = '%s.%s' % (CONF.volume_topic, host)
self.fake_args = None
self.fake_kwargs = None
# License for the specific language governing permissions and limitations
# under the License.
"""Unit Tests for volume transfers."""
+
+
import datetime
from cinder import context
from cinder import db
from cinder import exception
-from cinder import flags
from cinder.openstack.common import log as logging
from cinder import test
from cinder.transfer import api as transfer_api
-from cinder.volume import api as cinder_api
+
LOG = logging.getLogger(__name__)
"""
Unit Tests for volume types code
"""
+
+
import time
from cinder import context
from cinder.db.sqlalchemy import api as db_api
from cinder.db.sqlalchemy import models
from cinder import exception
-from cinder import flags
from cinder.openstack.common import log as logging
from cinder import test
from cinder.tests import fake_flags
from cinder.volume import volume_types
-FLAGS = flags.FLAGS
+
LOG = logging.getLogger(__name__)
"""Tests For miscellaneous util methods used with volume."""
+
+from oslo.config import cfg
+
from cinder import context
from cinder import db
-from cinder import flags
from cinder.openstack.common import importutils
from cinder.openstack.common import log as logging
from cinder.openstack.common.notifier import api as notifier_api
LOG = logging.getLogger(__name__)
-FLAGS = flags.FLAGS
+
+CONF = cfg.CONF
class UsageInfoTestCase(test.TestCase):
self.flags(connection_type='fake',
host='fake',
notification_driver=[test_notifier.__name__])
- self.volume = importutils.import_object(FLAGS.volume_manager)
+ self.volume = importutils.import_object(CONF.volume_manager)
self.user_id = 'fake'
self.project_id = 'fake'
self.snapshot_id = 'fake'
vol['snapshot_id'] = self.snapshot_id
vol['user_id'] = self.user_id
vol['project_id'] = self.project_id
- vol['host'] = FLAGS.host
- vol['availability_zone'] = FLAGS.storage_availability_zone
+ vol['host'] = CONF.host
+ vol['availability_zone'] = CONF.storage_availability_zone
vol['status'] = "creating"
vol['attach_status'] = "detached"
vol['size'] = self.volume_size
"""
Unit tests for Windows Server 2012 OpenStack Cinder volume driver
"""
+
+
import sys
from oslo.config import cfg
from cinder.tests.windows import windowsutils
from cinder.volume.drivers import windows
+
CONF = cfg.CONF
# License for the specific language governing permissions and limitations
# under the License.
+
import contextlib
import StringIO
import mock
import mox
-from oslo.config import cfg
from cinder.db import api as db_api
from cinder import exception
from cinder import test
from cinder.volume import configuration as conf
-from cinder.volume import driver as parent_driver
from cinder.volume.drivers.xenapi import lib
from cinder.volume.drivers.xenapi import sm as driver
from cinder.volume.drivers.xenapi import tools
# Erik Zaadi <erikz@il.ibm.com>
# Avishay Traeger <avishay@il.ibm.com>
+
import mox
+from oslo.config import cfg
from cinder import exception
-from cinder import flags
from cinder import test
from cinder.volume import configuration as conf
from cinder.volume.drivers import xiv
-FLAGS = flags.FLAGS
-
FAKE = "fake"
VOLUME = {'size': 16,
'name': FAKE,
CONNECTOR = {'initiator': "iqn.2012-07.org.fake:01:948f189c4695", }
+CONF = cfg.CONF
+
class XIVFakeProxyDriver(object):
"""Fake XIV Proxy Driver."""
self.volumes = {}
def setup(self, context):
- if self.xiv_info['xiv_user'] != FLAGS.san_login:
+ if self.xiv_info['xiv_user'] != CONF.san_login:
raise self.exception.NotAuthorized()
- if self.xiv_info['xiv_address'] != FLAGS.san_ip:
+ if self.xiv_info['xiv_address'] != CONF.san_ip:
raise self.exception.HostNotFound(host='fake')
def create_volume(self, volume):
"""Test that the san flags are passed to the XIV proxy."""
self.assertEquals(self.driver.xiv_proxy.xiv_info['xiv_user'],
- FLAGS.san_login)
+ CONF.san_login)
self.assertEquals(self.driver.xiv_proxy.xiv_info['xiv_pass'],
- FLAGS.san_password)
+ CONF.san_password)
self.assertEquals(self.driver.xiv_proxy.xiv_info['xiv_address'],
- FLAGS.san_ip)
+ CONF.san_ip)
self.assertEquals(self.driver.xiv_proxy.xiv_info['xiv_vol_pool'],
- FLAGS.san_clustername)
+ CONF.san_clustername)
def test_setup_should_fail_if_credentials_are_invalid(self):
"""Test that the xiv_proxy validates credentials."""
# License for the specific language governing permissions and limitations
#
+
import os
import cinder.context
-FLAGS = cinder.flags.FLAGS
-
def get_test_admin_context():
return cinder.context.get_admin_context()
"""
import os
-import sys
-from cinder import flags
+from oslo.config import cfg
# Check needed for unit testing on Unix
if os.name == 'nt':
import wmi
-FLAGS = flags.FLAGS
+
+CONF = cfg.CONF
class WindowsUtils(object):
def _get_vhd_path(self, volume_name):
'''Gets the path disk of the volume.'''
- base_vhd_folder = FLAGS.windows_iscsi_lun_path
+ base_vhd_folder = CONF.windows_iscsi_lun_path
return os.path.join(base_vhd_folder, volume_name + ".vhd")
def delete_snapshot(self, name):
Code Review / openstack-build / cinder-build.git / commitdiff