--- /dev/null
+# 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.
+
+import contextlib
+import datetime
+import StringIO
+import sys
+
+import mock
+from oslo.config import cfg
+import rtslib
+
+from cinder.cmd import all as cinder_all
+from cinder.cmd import api as cinder_api
+from cinder.cmd import backup as cinder_backup
+from cinder.cmd import manage as cinder_manage
+from cinder.cmd import rtstool as cinder_rtstool
+from cinder.cmd import scheduler as cinder_scheduler
+from cinder.cmd import volume as cinder_volume
+from cinder.cmd import volume_usage_audit
+from cinder import context
+from cinder import test
+from cinder import version
+
+CONF = cfg.CONF
+
+
+class TestCinderApiCmd(test.TestCase):
+ """Unit test cases for python modules under cinder/cmd."""
+
+ def setUp(self):
+ super(TestCinderApiCmd, self).setUp()
+ sys.argv = ['cinder-api']
+ CONF(sys.argv[1:], project='cinder', version=version.version_string())
+
+ def tearDown(self):
+ super(TestCinderApiCmd, self).tearDown()
+
+ @mock.patch('cinder.service.WSGIService')
+ @mock.patch('cinder.service.process_launcher')
+ @mock.patch('cinder.rpc.init')
+ @mock.patch('cinder.utils.monkey_patch')
+ @mock.patch('cinder.openstack.common.log.setup')
+ def test_main(self, log_setup, monkey_patch, rpc_init, process_launcher,
+ wsgi_service):
+ launcher = process_launcher.return_value
+ server = wsgi_service.return_value
+ server.workers = mock.sentinel.worker_count
+
+ cinder_api.main()
+
+ self.assertEqual(CONF.project, 'cinder')
+ self.assertEqual(CONF.version, version.version_string())
+ log_setup.assert_called_once_with("cinder")
+ monkey_patch.assert_called_once_with()
+ rpc_init.assert_called_once_with(CONF)
+ process_launcher.assert_called_once_with()
+ wsgi_service.assert_called_once_with('osapi_volume')
+ launcher.launch_service.assert_called_once_with(server,
+ workers=server.workers)
+ launcher.wait.assert_called_once_with()
+
+
+class TestCinderBackupCmd(test.TestCase):
+
+ def setUp(self):
+ super(TestCinderBackupCmd, self).setUp()
+ sys.argv = ['cinder-backup']
+ CONF(sys.argv[1:], project='cinder', version=version.version_string())
+
+ def tearDown(self):
+ super(TestCinderBackupCmd, self).tearDown()
+
+ @mock.patch('cinder.service.wait')
+ @mock.patch('cinder.service.serve')
+ @mock.patch('cinder.service.Service.create')
+ @mock.patch('cinder.utils.monkey_patch')
+ @mock.patch('cinder.openstack.common.log.setup')
+ def test_main(self, log_setup, monkey_patch, service_create, service_serve,
+ service_wait):
+ server = service_create.return_value
+
+ cinder_backup.main()
+
+ self.assertEqual(CONF.project, 'cinder')
+ self.assertEqual(CONF.version, version.version_string())
+ log_setup.assert_called_once_with("cinder")
+ monkey_patch.assert_called_once_with()
+ service_create.assert_called_once_with(binary='cinder-backup')
+ service_serve.assert_called_once_with(server)
+ service_wait.assert_called_once_with()
+
+
+class TestCinderAllCmd(test.TestCase):
+
+ def setUp(self):
+ super(TestCinderAllCmd, self).setUp()
+ sys.argv = ['cinder-all']
+ CONF(sys.argv[1:], project='cinder', version=version.version_string())
+
+ def tearDown(self):
+ super(TestCinderAllCmd, self).tearDown()
+
+ @mock.patch('cinder.service.Service.create')
+ @mock.patch('cinder.service.WSGIService')
+ @mock.patch('cinder.service.process_launcher')
+ @mock.patch('cinder.utils.monkey_patch')
+ @mock.patch('cinder.openstack.common.log.getLogger')
+ @mock.patch('cinder.openstack.common.log.setup')
+ def test_main(self, log_setup, get_logger, monkey_patch, process_launcher,
+ wsgi_service, service_create):
+ launcher = process_launcher.return_value
+ server = wsgi_service.return_value
+ server.workers = mock.sentinel.worker_count
+ service = service_create.return_value
+
+ cinder_all.main()
+
+ self.assertEqual(CONF.project, 'cinder')
+ self.assertEqual(CONF.version, version.version_string())
+ log_setup.assert_called_once_with("cinder")
+ get_logger.assert_called_once_with('cinder.all')
+ monkey_patch.assert_called_once_with()
+ process_launcher.assert_called_once_with()
+ wsgi_service.assert_called_once_with('osapi_volume')
+ launcher.launch_service.assert_any_call(server, workers=server.workers)
+
+ service_create.assert_has_calls([mock.call(binary='cinder-volume'),
+ mock.call(binary='cinder-scheduler'),
+ mock.call(binary='cinder-backup')])
+ self.assertEqual(3, service_create.call_count)
+ launcher.launch_service.assert_has_calls([mock.call(service)] * 3)
+ self.assertEqual(4, launcher.launch_service.call_count)
+
+ launcher.wait.assert_called_once_with()
+
+ @mock.patch('cinder.service.Service.create')
+ @mock.patch('cinder.service.WSGIService')
+ @mock.patch('cinder.service.process_launcher')
+ @mock.patch('cinder.utils.monkey_patch')
+ @mock.patch('cinder.openstack.common.log.getLogger')
+ @mock.patch('cinder.openstack.common.log.setup')
+ def test_main_load_osapi_volume_exception(self, log_setup, get_logger,
+ monkey_patch, process_launcher,
+ wsgi_service, service_create):
+ launcher = process_launcher.return_value
+ server = wsgi_service.return_value
+ server.workers = mock.sentinel.worker_count
+ mock_log = get_logger.return_value
+
+ for ex in (Exception(), SystemExit()):
+ launcher.launch_service.side_effect = ex
+
+ cinder_all.main()
+
+ self.assertEqual(CONF.project, 'cinder')
+ self.assertEqual(CONF.version, version.version_string())
+ log_setup.assert_called_once_with("cinder")
+ get_logger.assert_called_once_with('cinder.all')
+ monkey_patch.assert_called_once_with()
+ process_launcher.assert_called_once_with()
+ wsgi_service.assert_called_once_with('osapi_volume')
+ launcher.launch_service.assert_any_call(server,
+ workers=server.workers)
+ self.assertTrue(mock_log.exception.called)
+
+ # Reset for the next exception
+ log_setup.reset_mock()
+ get_logger.reset_mock()
+ monkey_patch.reset_mock()
+ process_launcher.reset_mock()
+ wsgi_service.reset_mock()
+ mock_log.reset_mock()
+
+ @mock.patch('cinder.service.Service.create')
+ @mock.patch('cinder.service.WSGIService')
+ @mock.patch('cinder.service.process_launcher')
+ @mock.patch('cinder.utils.monkey_patch')
+ @mock.patch('cinder.openstack.common.log.getLogger')
+ @mock.patch('cinder.openstack.common.log.setup')
+ def test_main_load_binary_exception(self, log_setup, get_logger,
+ monkey_patch, process_launcher,
+ wsgi_service, service_create):
+ launcher = process_launcher.return_value
+ server = wsgi_service.return_value
+ server.workers = mock.sentinel.worker_count
+ service = service_create.return_value
+ mock_log = get_logger.return_value
+
+ def launch_service(*args, **kwargs):
+ if service in args:
+ raise Exception()
+
+ launcher.launch_service.side_effect = launch_service
+
+ cinder_all.main()
+
+ self.assertEqual(CONF.project, 'cinder')
+ self.assertEqual(CONF.version, version.version_string())
+ log_setup.assert_called_once_with("cinder")
+ get_logger.assert_called_once_with('cinder.all')
+ monkey_patch.assert_called_once_with()
+ process_launcher.assert_called_once_with()
+ wsgi_service.assert_called_once_with('osapi_volume')
+ launcher.launch_service.assert_any_call(server,
+ workers=server.workers)
+ for binary in ['cinder-volume', 'cinder-scheduler', 'cinder-backup']:
+ service_create.assert_any_call(binary=binary)
+ launcher.launch_service.assert_called_with(service)
+ self.assertTrue(mock_log.exception.called)
+
+
+class TestCinderSchedulerCmd(test.TestCase):
+
+ def setUp(self):
+ super(TestCinderSchedulerCmd, self).setUp()
+ sys.argv = ['cinder-scheduler']
+ CONF(sys.argv[1:], project='cinder', version=version.version_string())
+
+ def tearDown(self):
+ super(TestCinderSchedulerCmd, self).tearDown()
+
+ @mock.patch('cinder.service.wait')
+ @mock.patch('cinder.service.serve')
+ @mock.patch('cinder.service.Service.create')
+ @mock.patch('cinder.utils.monkey_patch')
+ @mock.patch('cinder.openstack.common.log.setup')
+ def test_main(self, log_setup, monkey_patch, service_create,
+ service_serve, service_wait):
+ server = service_create.return_value
+
+ cinder_scheduler.main()
+
+ self.assertEqual(CONF.project, 'cinder')
+ self.assertEqual(CONF.version, version.version_string())
+ log_setup.assert_called_once_with("cinder")
+ monkey_patch.assert_called_once_with()
+ service_create.assert_called_once_with(binary='cinder-scheduler')
+ service_serve.assert_called_once_with(server)
+ service_wait.assert_called_once_with()
+
+
+class TestCinderVolumeCmd(test.TestCase):
+
+ def setUp(self):
+ super(TestCinderVolumeCmd, self).setUp()
+ sys.argv = ['cinder-volume']
+ CONF(sys.argv[1:], project='cinder', version=version.version_string())
+
+ def tearDown(self):
+ super(TestCinderVolumeCmd, self).tearDown()
+
+ @mock.patch('cinder.service.get_launcher')
+ @mock.patch('cinder.service.Service.create')
+ @mock.patch('cinder.utils.monkey_patch')
+ @mock.patch('cinder.openstack.common.log.setup')
+ def test_main(self, log_setup, monkey_patch, service_create,
+ get_launcher):
+ CONF.set_override('enabled_backends', None)
+ launcher = get_launcher.return_value
+ server = service_create.return_value
+
+ cinder_volume.main()
+
+ self.assertEqual(CONF.project, 'cinder')
+ self.assertEqual(CONF.version, version.version_string())
+ log_setup.assert_called_once_with("cinder")
+ monkey_patch.assert_called_once_with()
+ get_launcher.assert_called_once_with()
+ service_create.assert_called_once_with(binary='cinder-volume')
+ launcher.launch_service.assert_called_once_with(server)
+ launcher.wait.assert_called_once_with()
+
+ @mock.patch('cinder.service.get_launcher')
+ @mock.patch('cinder.service.Service.create')
+ @mock.patch('cinder.utils.monkey_patch')
+ @mock.patch('cinder.openstack.common.log.setup')
+ def test_main_with_backends(self, log_setup, monkey_patch, service_create,
+ get_launcher):
+ backends = ['backend1', 'backend2']
+ CONF.set_override('enabled_backends', backends)
+ launcher = get_launcher.return_value
+
+ cinder_volume.main()
+
+ self.assertEqual(CONF.project, 'cinder')
+ self.assertEqual(CONF.version, version.version_string())
+ log_setup.assert_called_once_with("cinder")
+ monkey_patch.assert_called_once_with()
+ get_launcher.assert_called_once_with()
+ self.assertEqual(len(backends), service_create.call_count)
+ self.assertEqual(len(backends), launcher.launch_service.call_count)
+ launcher.wait.assert_called_once_with()
+
+
+class TestCinderManageCmd(test.TestCase):
+
+ def setUp(self):
+ super(TestCinderManageCmd, self).setUp()
+ sys.argv = ['cinder-manage']
+ CONF(sys.argv[1:], project='cinder', version=version.version_string())
+
+ def tearDown(self):
+ super(TestCinderManageCmd, self).tearDown()
+
+ @mock.patch('cinder.openstack.common.uuidutils.is_uuid_like')
+ def test_param2id(self, is_uuid_like):
+ mock_object_id = mock.MagicMock()
+ is_uuid_like.return_value = True
+
+ object_id = cinder_manage.param2id(mock_object_id)
+ self.assertEqual(mock_object_id, object_id)
+ is_uuid_like.assert_called_once_with(mock_object_id)
+
+ @mock.patch('cinder.openstack.common.uuidutils.is_uuid_like')
+ def test_param2id_int_string(self, is_uuid_like):
+ object_id_str = '10'
+ is_uuid_like.return_value = False
+
+ object_id = cinder_manage.param2id(object_id_str)
+ self.assertEqual(10, object_id)
+ is_uuid_like.assert_called_once_with(object_id_str)
+
+ @mock.patch('cinder.db.migration.db_sync')
+ def test_db_commands_sync(self, db_sync):
+ version = mock.MagicMock()
+ db_cmds = cinder_manage.DbCommands()
+ db_cmds.sync(version=version)
+ db_sync.assert_called_once_with(version)
+
+ @mock.patch('cinder.db.migration.db_version')
+ def test_db_commands_version(self, db_version):
+ db_cmds = cinder_manage.DbCommands()
+ db_cmds.version()
+ db_version.assert_called_once_with()
+
+ @mock.patch('cinder.version.version_string')
+ def test_versions_commands_list(self, version_string):
+ version_cmds = cinder_manage.VersionCommands()
+ version_cmds.list()
+ version_string.assert_called_once_with()
+
+ @mock.patch('cinder.version.version_string')
+ def test_versions_commands_call(self, version_string):
+ version_cmds = cinder_manage.VersionCommands()
+ version_cmds.__call__()
+ version_string.assert_called_once_with()
+
+ @mock.patch('cinder.db.service_get_all')
+ @mock.patch('cinder.context.get_admin_context')
+ def test_host_commands_list(self, get_admin_context, service_get_all):
+ get_admin_context.return_value = mock.sentinel.ctxt
+ service_get_all.return_value = [{'host': 'fake-host',
+ 'availability_zone': 'fake-az'}]
+
+ with mock.patch('sys.stdout', new=StringIO.StringIO()) as fake_out:
+ expected_out = ("%(host)-25s\t%(zone)-15s\n" %
+ {'host': 'host', 'zone': 'zone'})
+ expected_out += ("%(host)-25s\t%(availability_zone)-15s\n" %
+ {'host': 'fake-host',
+ 'availability_zone': 'fake-az'})
+ host_cmds = cinder_manage.HostCommands()
+ host_cmds.list()
+
+ get_admin_context.assert_called_once_with()
+ service_get_all.assert_called_once_with(mock.sentinel.ctxt)
+ self.assertEqual(expected_out, fake_out.getvalue())
+
+ @mock.patch('cinder.db.service_get_all')
+ @mock.patch('cinder.context.get_admin_context')
+ def test_host_commands_list_with_zone(self, get_admin_context,
+ service_get_all):
+ get_admin_context.return_value = mock.sentinel.ctxt
+ service_get_all.return_value = [{'host': 'fake-host',
+ 'availability_zone': 'fake-az1'},
+ {'host': 'fake-host',
+ 'availability_zone': 'fake-az2'}]
+
+ with mock.patch('sys.stdout', new=StringIO.StringIO()) as fake_out:
+ expected_out = ("%(host)-25s\t%(zone)-15s\n" %
+ {'host': 'host', 'zone': 'zone'})
+ expected_out += ("%(host)-25s\t%(availability_zone)-15s\n" %
+ {'host': 'fake-host',
+ 'availability_zone': 'fake-az1'})
+ host_cmds = cinder_manage.HostCommands()
+ host_cmds.list(zone='fake-az1')
+
+ get_admin_context.assert_called_once_with()
+ service_get_all.assert_called_once_with(mock.sentinel.ctxt)
+ self.assertEqual(expected_out, fake_out.getvalue())
+
+ @mock.patch('cinder.rpc.get_client')
+ @mock.patch('cinder.rpc.init')
+ @mock.patch('cinder.rpc.initialized', return_value=False)
+ @mock.patch('oslo.messaging.Target')
+ def test_volume_commands_init(self, messaging_target, rpc_initialized,
+ rpc_init, get_client):
+ CONF.set_override('volume_topic', 'fake-topic')
+ mock_target = messaging_target.return_value
+ mock_rpc_client = get_client.return_value
+
+ volume_cmds = cinder_manage.VolumeCommands()
+ rpc_client = volume_cmds.rpc_client()
+
+ rpc_initialized.assert_called_once_with()
+ rpc_init.assert_called_once_with(CONF)
+ messaging_target.assert_called_once_with(topic='fake-topic')
+ get_client.assert_called_once_with(mock_target)
+ self.assertEqual(mock_rpc_client, rpc_client)
+
+ @mock.patch('cinder.db.volume_get')
+ @mock.patch('cinder.context.get_admin_context')
+ @mock.patch('cinder.rpc.get_client')
+ @mock.patch('cinder.rpc.init')
+ def test_volume_commands_delete(self, rpc_init, get_client,
+ get_admin_context, volume_get):
+ ctxt = context.RequestContext('fake-user', 'fake-project')
+ get_admin_context.return_value = ctxt
+ mock_client = mock.MagicMock()
+ cctxt = mock.MagicMock()
+ mock_client.prepare.return_value = cctxt
+ get_client.return_value = mock_client
+ volume_id = '123'
+ volume = {'id': volume_id, 'host': 'fake-host', 'status': 'available'}
+ volume_get.return_value = volume
+
+ volume_cmds = cinder_manage.VolumeCommands()
+ volume_cmds._client = mock_client
+ volume_cmds.delete(volume_id)
+
+ volume_get.assert_called_once_with(ctxt, 123)
+ mock_client.prepare.assert_called_once_with(server=volume['host'])
+ cctxt.cast.assert_called_once_with(ctxt, 'delete_volume',
+ volume_id=volume['id'])
+
+ @mock.patch('cinder.db.volume_destroy')
+ @mock.patch('cinder.db.volume_get')
+ @mock.patch('cinder.context.get_admin_context')
+ @mock.patch('cinder.rpc.init')
+ def test_volume_commands_delete_no_host(self, rpc_init, get_admin_context,
+ volume_get, volume_destroy):
+ ctxt = context.RequestContext('fake-user', 'fake-project')
+ get_admin_context.return_value = ctxt
+ volume_id = '123'
+ volume = {'id': volume_id, 'host': None, 'status': 'available'}
+ volume_get.return_value = volume
+
+ with mock.patch('sys.stdout', new=StringIO.StringIO()) as fake_out:
+ expected_out = ('Volume not yet assigned to host.\n'
+ 'Deleting volume from database and skipping'
+ ' rpc.\n')
+ volume_cmds = cinder_manage.VolumeCommands()
+ volume_cmds.delete(volume_id)
+
+ get_admin_context.assert_called_once_with()
+ volume_get.assert_called_once_with(ctxt, 123)
+ volume_destroy.assert_called_once_with(ctxt, 123)
+ self.assertEqual(expected_out, fake_out.getvalue())
+
+ @mock.patch('cinder.db.volume_destroy')
+ @mock.patch('cinder.db.volume_get')
+ @mock.patch('cinder.context.get_admin_context')
+ @mock.patch('cinder.rpc.init')
+ def test_volume_commands_delete_volume_in_use(self, rpc_init,
+ get_admin_context,
+ volume_get, volume_destroy):
+ ctxt = context.RequestContext('fake-user', 'fake-project')
+ get_admin_context.return_value = ctxt
+ volume_id = '123'
+ volume = {'id': volume_id, 'host': 'fake-host', 'status': 'in-use'}
+ volume_get.return_value = volume
+
+ with mock.patch('sys.stdout', new=StringIO.StringIO()) as fake_out:
+ expected_out = ('Volume is in-use.\n'
+ 'Detach volume from instance and then try'
+ ' again.\n')
+ volume_cmds = cinder_manage.VolumeCommands()
+ volume_cmds.delete(volume_id)
+
+ volume_get.assert_called_once_with(ctxt, 123)
+ self.assertEqual(expected_out, fake_out.getvalue())
+
+ def test_config_commands_list(self):
+ with mock.patch('sys.stdout', new=StringIO.StringIO()) as fake_out:
+ expected_out = ''
+ for key, value in CONF.iteritems():
+ expected_out += '%s = %s' % (key, value) + '\n'
+
+ config_cmds = cinder_manage.ConfigCommands()
+ config_cmds.list()
+
+ self.assertEqual(expected_out, fake_out.getvalue())
+
+ def test_config_commands_list_param(self):
+ with mock.patch('sys.stdout', new=StringIO.StringIO()) as fake_out:
+ CONF.set_override('host', 'fake')
+ expected_out = 'host = fake\n'
+
+ config_cmds = cinder_manage.ConfigCommands()
+ config_cmds.list(param='host')
+
+ self.assertEqual(expected_out, fake_out.getvalue())
+
+ def test_get_log_commands_no_errors(self):
+ with mock.patch('sys.stdout', new=StringIO.StringIO()) as fake_out:
+ CONF.set_override('log_dir', None)
+ expected_out = 'No errors in logfiles!\n'
+
+ get_log_cmds = cinder_manage.GetLogCommands()
+ get_log_cmds.errors()
+
+ self.assertEqual(expected_out, fake_out.getvalue())
+
+ @mock.patch('__builtin__.open')
+ @mock.patch('os.listdir')
+ def test_get_log_commands_errors(self, listdir, open):
+ CONF.set_override('log_dir', 'fake-dir')
+ listdir.return_value = ['fake-error.log']
+
+ with mock.patch('sys.stdout', new=StringIO.StringIO()) as fake_out:
+ open.return_value = StringIO.StringIO(
+ '[ ERROR ] fake-error-message')
+ expected_out = ('fake-dir/fake-error.log:-\n'
+ 'Line 1 : [ ERROR ] fake-error-message\n')
+
+ get_log_cmds = cinder_manage.GetLogCommands()
+ get_log_cmds.errors()
+
+ self.assertEqual(expected_out, fake_out.getvalue())
+ open.assert_called_once_with('fake-dir/fake-error.log', 'r')
+ listdir.assert_called_once_with(CONF.log_dir)
+
+ @mock.patch('__builtin__.open')
+ @mock.patch('os.path.exists')
+ def test_get_log_commands_syslog_no_log_file(self, path_exists, open):
+ path_exists.return_value = False
+
+ get_log_cmds = cinder_manage.GetLogCommands()
+ exit = self.assertRaises(SystemExit, get_log_cmds.syslog)
+
+ self.assertEqual(exit.code, 1)
+ path_exists.assert_any_call('/var/log/syslog')
+ path_exists.assert_any_call('/var/log/messages')
+
+ @mock.patch('cinder.db.backup_get_all')
+ @mock.patch('cinder.context.get_admin_context')
+ def test_backup_commands_list(self, get_admin_context, backup_get_all):
+ ctxt = context.RequestContext('fake-user', 'fake-project')
+ get_admin_context.return_value = ctxt
+ backup = {'id': 1,
+ 'user_id': 'fake-user-id',
+ 'project_id': 'fake-project-id',
+ 'host': 'fake-host',
+ 'display_name': 'fake-display-name',
+ 'container': 'fake-container',
+ 'status': 'fake-status',
+ 'size': 123,
+ 'object_count': 1}
+ backup_get_all.return_value = [backup]
+ with mock.patch('sys.stdout', new=StringIO.StringIO()) as fake_out:
+ hdr = ('%-32s\t%-32s\t%-32s\t%-24s\t%-24s\t%-12s\t%-12s\t%-12s'
+ '\t%-12s')
+ header = hdr % ('ID',
+ 'User ID',
+ 'Project ID',
+ 'Host',
+ 'Name',
+ 'Container',
+ 'Status',
+ 'Size',
+ 'Object Count')
+ res = ('%-32s\t%-32s\t%-32s\t%-24s\t%-24s\t%-12s\t%-12s\t%-12d'
+ '\t%-12s')
+ resource = res % (backup['id'],
+ backup['user_id'],
+ backup['project_id'],
+ backup['host'],
+ backup['display_name'],
+ backup['container'],
+ backup['status'],
+ backup['size'],
+ 1)
+ expected_out = header + '\n' + resource + '\n'
+
+ backup_cmds = cinder_manage.BackupCommands()
+ backup_cmds.list()
+
+ get_admin_context.assert_called_once_with()
+ backup_get_all.assert_called_once_with(ctxt)
+ self.assertEqual(expected_out, fake_out.getvalue())
+
+ @mock.patch('cinder.utils.service_is_up')
+ @mock.patch('cinder.db.service_get_all')
+ @mock.patch('cinder.context.get_admin_context')
+ def test_service_commands_list(self, get_admin_context, service_get_all,
+ service_is_up):
+ ctxt = context.RequestContext('fake-user', 'fake-project')
+ get_admin_context.return_value = ctxt
+ service = {'binary': 'cinder-binary',
+ 'host': 'fake-host.fake-domain',
+ 'availability_zone': 'fake-zone',
+ 'updated_at': '2014-06-30 11:22:33',
+ 'disabled': False}
+ service_get_all.return_value = [service]
+ service_is_up.return_value = True
+ with mock.patch('sys.stdout', new=StringIO.StringIO()) as fake_out:
+ format = "%-16s %-36s %-16s %-10s %-5s %-10s"
+ print_format = format % ('Binary',
+ 'Host',
+ 'Zone',
+ 'Status',
+ 'State',
+ 'Updated At')
+ service_format = format % (service['binary'],
+ service['host'].partition('.')[0],
+ service['availability_zone'],
+ 'enabled',
+ ':-)',
+ service['updated_at'])
+ expected_out = print_format + '\n' + service_format + '\n'
+
+ service_cmds = cinder_manage.ServiceCommands()
+ service_cmds.list()
+
+ self.assertEqual(expected_out, fake_out.getvalue())
+ get_admin_context.assert_called_with()
+ service_get_all.assert_called_with(ctxt)
+ service_is_up.assert_called_with(service)
+
+ @mock.patch('oslo.config.cfg.ConfigOpts.register_cli_opt')
+ def test_main_argv_lt_2(self, register_cli_opt):
+ script_name = 'cinder-manage'
+ sys.argv = [script_name]
+ CONF(sys.argv[1:], project='cinder', version=version.version_string())
+
+ exit = self.assertRaises(SystemExit, cinder_manage.main)
+
+ self.assertTrue(register_cli_opt.called)
+ self.assertEqual(exit.code, 2)
+
+ @mock.patch('oslo.config.cfg.ConfigOpts.__call__')
+ @mock.patch('cinder.openstack.common.log.setup')
+ @mock.patch('oslo.config.cfg.ConfigOpts.register_cli_opt')
+ def test_main_sudo_failed(self, register_cli_opt, log_setup,
+ config_opts_call):
+ script_name = 'cinder-manage'
+ sys.argv = [script_name, 'fake_category', 'fake_action']
+ config_opts_call.side_effect = cfg.ConfigFilesNotFoundError(
+ mock.sentinel._namespace)
+
+ exit = self.assertRaises(SystemExit, cinder_manage.main)
+
+ self.assertTrue(register_cli_opt.called)
+ config_opts_call.assert_called_once_with(
+ sys.argv[1:], project='cinder',
+ version=version.version_string())
+ self.assertFalse(log_setup.called)
+ self.assertEqual(exit.code, 2)
+
+ @mock.patch('oslo.config.cfg.ConfigOpts.__call__')
+ @mock.patch('oslo.config.cfg.ConfigOpts.register_cli_opt')
+ def test_main(self, register_cli_opt, config_opts_call):
+ script_name = 'cinder-manage'
+ sys.argv = [script_name, 'config', 'list']
+ action_fn = mock.MagicMock()
+ CONF.category = mock.MagicMock(action_fn=action_fn)
+
+ cinder_manage.main()
+
+ self.assertTrue(register_cli_opt.called)
+ config_opts_call.assert_called_once_with(
+ sys.argv[1:], project='cinder', version=version.version_string())
+ self.assertTrue(action_fn.called)
+
+
+class TestCinderRtstoolCmd(test.TestCase):
+
+ def setUp(self):
+ super(TestCinderRtstoolCmd, self).setUp()
+ sys.argv = ['cinder-rtstool']
+ CONF(sys.argv[1:], project='cinder', version=version.version_string())
+
+ def tearDown(self):
+ super(TestCinderRtstoolCmd, self).tearDown()
+
+ @mock.patch('rtslib.root.RTSRoot')
+ def test_create_rtsllib_error(self, rtsroot):
+ rtsroot.side_effect = rtslib.utils.RTSLibError()
+
+ self.assertRaises(rtslib.utils.RTSLibError, cinder_rtstool.create,
+ mock.sentinel.backing_device,
+ mock.sentinel.name,
+ mock.sentinel.userid,
+ mock.sentinel.password)
+
+ def _test_create_rtsllib_error_network_portal(self, ip):
+ with contextlib.nested(
+ mock.patch('rtslib.NetworkPortal'),
+ mock.patch('rtslib.LUN'),
+ mock.patch('rtslib.TPG'),
+ mock.patch('rtslib.FabricModule'),
+ mock.patch('rtslib.Target'),
+ mock.patch('rtslib.BlockStorageObject'),
+ mock.patch('rtslib.root.RTSRoot')
+ ) as (network_portal, lun, tpg, fabric_module, target,
+ block_storage_object, rts_root):
+ root_new = mock.MagicMock(storage_objects=mock.MagicMock())
+ rts_root.return_value = root_new
+ block_storage_object.return_value = mock.sentinel.so_new
+ target.return_value = mock.sentinel.target_new
+ fabric_module.return_value = mock.sentinel.fabric_new
+ tpg_new = tpg.return_value
+ lun.return_value = mock.sentinel.lun_new
+
+ if ip == '0.0.0.0':
+ network_portal.side_effect = rtslib.utils.RTSLibError()
+ self.assertRaises(rtslib.utils.RTSLibError,
+ cinder_rtstool.create,
+ mock.sentinel.backing_device,
+ mock.sentinel.name,
+ mock.sentinel.userid,
+ mock.sentinel.password)
+ else:
+ cinder_rtstool.create(mock.sentinel.backing_device,
+ mock.sentinel.name,
+ mock.sentinel.userid,
+ mock.sentinel.password)
+
+ rts_root.assert_called_once_with()
+ block_storage_object.assert_called_once_with(
+ name=mock.sentinel.name, dev=mock.sentinel.backing_device)
+ target.assert_called_once_with(mock.sentinel.fabric_new,
+ mock.sentinel.name, 'create')
+ fabric_module.assert_called_once_with('iscsi')
+ tpg.assert_called_once_with(mock.sentinel.target_new,
+ mode='create')
+ tpg_new.set_attribute.assert_called_once_with('authentication',
+ '1')
+ lun.assert_called_once_with(tpg_new,
+ storage_object=mock.sentinel.so_new)
+ self.assertEqual(1, tpg_new.enable)
+ network_portal.assert_any_call(tpg_new, ip, 3260,
+ mode='any')
+
+ if ip == '::0':
+ network_portal.assert_any_call(tpg_new, ip, 3260, mode='any')
+
+ def test_create_rtsllib_error_network_portal_ipv4(self):
+ self._test_create_rtsllib_error_network_portal('0.0.0.0')
+
+ def test_create_rtsllib_error_network_portal_ipv6(self):
+ self._test_create_rtsllib_error_network_portal('::0')
+
+ def _test_create(self, ip):
+ with contextlib.nested(
+ mock.patch('rtslib.NetworkPortal'),
+ mock.patch('rtslib.LUN'),
+ mock.patch('rtslib.TPG'),
+ mock.patch('rtslib.FabricModule'),
+ mock.patch('rtslib.Target'),
+ mock.patch('rtslib.BlockStorageObject'),
+ mock.patch('rtslib.root.RTSRoot')
+ ) as (network_portal, lun, tpg, fabric_module, target,
+ block_storage_object, rts_root):
+ root_new = mock.MagicMock(storage_objects=mock.MagicMock())
+ rts_root.return_value = root_new
+ block_storage_object.return_value = mock.sentinel.so_new
+ target.return_value = mock.sentinel.target_new
+ fabric_module.return_value = mock.sentinel.fabric_new
+ tpg_new = tpg.return_value
+ lun.return_value = mock.sentinel.lun_new
+
+ def network_portal_exception(*args, **kwargs):
+ if set([tpg_new, '::0', 3260]).issubset(list(args)):
+ raise rtslib.utils.RTSLibError()
+ else:
+ pass
+
+ cinder_rtstool.create(mock.sentinel.backing_device,
+ mock.sentinel.name,
+ mock.sentinel.userid,
+ mock.sentinel.password)
+
+ rts_root.assert_called_once_with()
+ block_storage_object.assert_called_once_with(
+ name=mock.sentinel.name, dev=mock.sentinel.backing_device)
+ target.assert_called_once_with(mock.sentinel.fabric_new,
+ mock.sentinel.name, 'create')
+ fabric_module.assert_called_once_with('iscsi')
+ tpg.assert_called_once_with(mock.sentinel.target_new,
+ mode='create')
+ tpg_new.set_attribute.assert_called_once_with('authentication',
+ '1')
+ lun.assert_called_once_with(tpg_new,
+ storage_object=mock.sentinel.so_new)
+ self.assertEqual(1, tpg_new.enable)
+ network_portal.assert_any_call(tpg_new, ip, 3260,
+ mode='any')
+
+ if ip == '::0':
+ network_portal.assert_any_call(tpg_new, ip, 3260, mode='any')
+
+ def test_create_ipv4(self):
+ self._test_create('0.0.0.0')
+
+ def test_create_ipv6(self):
+ self._test_create('::0')
+
+ @mock.patch('rtslib.root.RTSRoot')
+ def test_add_initiator_rtslib_error(self, rtsroot):
+ rtsroot.side_effect = rtslib.utils.RTSLibError()
+
+ self.assertRaises(rtslib.utils.RTSLibError,
+ cinder_rtstool.add_initiator,
+ mock.sentinel.target_iqn,
+ mock.sentinel.initiator_iqn,
+ mock.sentinel.userid,
+ mock.sentinel.password)
+
+ @mock.patch('rtslib.root.RTSRoot')
+ def test_add_initiator_rtstool_error(self, rtsroot):
+ rtsroot.targets.return_value = {}
+
+ self.assertRaises(cinder_rtstool.RtstoolError,
+ cinder_rtstool.add_initiator,
+ mock.sentinel.target_iqn,
+ mock.sentinel.initiator_iqn,
+ mock.sentinel.userid,
+ mock.sentinel.password)
+
+ @mock.patch('rtslib.MappedLUN')
+ @mock.patch('rtslib.NodeACL')
+ @mock.patch('rtslib.root.RTSRoot')
+ def test_add_initiator_acl_exists(self, rtsroot, node_acl, mapped_lun):
+ target_iqn = mock.MagicMock()
+ target_iqn.tpgs.return_value = \
+ [{'node_acls': mock.sentinel.initiator_iqn}]
+ acl = {'node_wwn': mock.sentinel.initiator_iqn}
+ tpg = mock.MagicMock(node_acls=[acl])
+ tpgs = mock.MagicMock()
+ tpgs.next.return_value = tpg
+ target = mock.MagicMock(tpgs=tpgs, wwn=target_iqn)
+ rtsroot.return_value = mock.MagicMock(targets=[target])
+
+ cinder_rtstool.add_initiator(target_iqn,
+ mock.sentinel.initiator_iqn,
+ mock.sentinel.userid,
+ mock.sentinel.password)
+ self.assertFalse(node_acl.called)
+ self.assertFalse(mapped_lun.called)
+
+ @mock.patch('rtslib.MappedLUN')
+ @mock.patch('rtslib.NodeACL')
+ @mock.patch('rtslib.root.RTSRoot')
+ def test_add_initiator(self, rtsroot, node_acl, mapped_lun):
+ target_iqn = mock.MagicMock()
+ target_iqn.tpgs.return_value = \
+ [{'node_acls': mock.sentinel.initiator_iqn}]
+ tpg = mock.MagicMock()
+ target = mock.MagicMock(tpgs=tpg, wwn=target_iqn)
+ rtsroot.return_value = mock.MagicMock(targets=[target])
+
+ acl_new = mock.MagicMock(chap_userid=mock.sentinel.userid,
+ chap_password=mock.sentinel.password)
+ node_acl.return_value = acl_new
+
+ cinder_rtstool.add_initiator(target_iqn,
+ mock.sentinel.initiator_iqn,
+ mock.sentinel.userid,
+ mock.sentinel.password)
+ node_acl.assert_called_once_with(tpg.next(),
+ mock.sentinel.initiator_iqn,
+ mode='create')
+ mapped_lun.assert_called_once_with(acl_new, 0, tpg_lun=0)
+
+ @mock.patch('rtslib.root.RTSRoot')
+ def test_get_targets(self, rtsroot):
+ target = mock.MagicMock()
+ target.dump.return_value = {'wwn': 'fake-wwn'}
+ rtsroot.return_value = mock.MagicMock(targets=[target])
+
+ with mock.patch('sys.stdout', new=StringIO.StringIO()) as fake_out:
+ cinder_rtstool.get_targets()
+
+ self.assertEqual(str(target.wwn), fake_out.getvalue().strip())
+
+ @mock.patch('rtslib.root.RTSRoot')
+ def test_delete(self, rtsroot):
+ target = mock.MagicMock(wwn=mock.sentinel.iqn)
+ storage_object = mock.MagicMock()
+ name = mock.PropertyMock(return_value=mock.sentinel.iqn)
+ type(storage_object).name = name
+ rtsroot.return_value = mock.MagicMock(
+ targets=[target], storage_objects=[storage_object])
+
+ cinder_rtstool.delete(mock.sentinel.iqn)
+
+ target.delete.assert_called_once_with()
+ storage_object.delete.assert_called_once_with()
+
+ def test_usage(self):
+ exit = self.assertRaises(SystemExit, cinder_rtstool.usage)
+
+ self.assertEqual(exit.code, 1)
+
+ @mock.patch('cinder.cmd.rtstool.usage')
+ def test_main_argc_lt_2(self, usage):
+ usage.side_effect = SystemExit(1)
+ sys.argv = ['cinder-rtstool']
+
+ exit = self.assertRaises(SystemExit, cinder_rtstool.usage)
+
+ self.assertTrue(usage.called)
+ self.assertEqual(exit.code, 1)
+
+ def test_main_create_argv_lt_6(self):
+ sys.argv = ['cinder-rtstool', 'create']
+ self._test_main_check_argv()
+
+ def test_main_create_argv_gt_7(self):
+ sys.argv = ['cinder-rtstool', 'create', 'fake-arg1', 'fake-arg2',
+ 'fake-arg3', 'fake-arg4', 'fake-arg5', 'fake-arg6']
+ self._test_main_check_argv()
+
+ def test_main_add_initiator_argv_lt_6(self):
+ sys.argv = ['cinder-rtstool', 'add-initiator']
+ self._test_main_check_argv()
+
+ def test_main_delete_argv_lt_3(self):
+ sys.argv = ['cinder-rtstool', 'delete']
+ self._test_main_check_argv()
+
+ def test_main_no_action(self):
+ sys.argv = ['cinder-rtstool']
+ self._test_main_check_argv()
+
+ def _test_main_check_argv(self):
+ with mock.patch('cinder.cmd.rtstool.usage') as usage:
+ usage.side_effect = SystemExit(1)
+ sys.argv = ['cinder-rtstool', 'create']
+
+ exit = self.assertRaises(SystemExit, cinder_rtstool.main)
+
+ self.assertTrue(usage.called)
+ self.assertEqual(exit.code, 1)
+
+ def test_main_create(self):
+ with mock.patch('cinder.cmd.rtstool.create') as create:
+ sys.argv = ['cinder-rtstool',
+ 'create',
+ mock.sentinel.backing_device,
+ mock.sentinel.name,
+ mock.sentinel.userid,
+ mock.sentinel.password,
+ mock.sentinel.initiator_iqns]
+
+ rc = cinder_rtstool.main()
+
+ create.assert_called_once_with(mock.sentinel.backing_device,
+ mock.sentinel.name,
+ mock.sentinel.userid,
+ mock.sentinel.password,
+ mock.sentinel.initiator_iqns)
+ self.assertEqual(0, rc)
+
+ def test_main_add_initiator(self):
+ with mock.patch('cinder.cmd.rtstool.add_initiator') as add_initiator:
+ sys.argv = ['cinder-rtstool',
+ 'add-initiator',
+ mock.sentinel.target_iqn,
+ mock.sentinel.userid,
+ mock.sentinel.password,
+ mock.sentinel.initiator_iqns]
+
+ rc = cinder_rtstool.main()
+
+ add_initiator.assert_called_once_with(
+ mock.sentinel.target_iqn, mock.sentinel.initiator_iqns,
+ mock.sentinel.userid, mock.sentinel.password)
+ self.assertEqual(0, rc)
+
+ def test_main_get_targets(self):
+ with mock.patch('cinder.cmd.rtstool.get_targets') as get_targets:
+ sys.argv = ['cinder-rtstool', 'get-targets']
+
+ rc = cinder_rtstool.main()
+
+ get_targets.assert_called_once_with()
+ self.assertEqual(0, rc)
+
+ def test_main_delete(self):
+ with mock.patch('cinder.cmd.rtstool.delete') as delete:
+ sys.argv = ['cinder-rtstool', 'delete', mock.sentinel.iqn]
+
+ rc = cinder_rtstool.main()
+
+ delete.assert_called_once_with(mock.sentinel.iqn)
+ self.assertEqual(0, rc)
+
+ def test_main_verify(self):
+ with mock.patch('cinder.cmd.rtstool.verify_rtslib') as verify_rtslib:
+ sys.argv = ['cinder-rtstool', 'verify']
+
+ rc = cinder_rtstool.main()
+
+ verify_rtslib.assert_called_once_with()
+ self.assertEqual(0, rc)
+
+
+class TestCinderVolumeUsageAuditCmd(test.TestCase):
+
+ def setUp(self):
+ super(TestCinderVolumeUsageAuditCmd, self).setUp()
+ sys.argv = ['cinder-volume-usage-audit']
+ CONF(sys.argv[1:], project='cinder', version=version.version_string())
+
+ def tearDown(self):
+ super(TestCinderVolumeUsageAuditCmd, self).tearDown()
+
+ @mock.patch('cinder.utils.last_completed_audit_period')
+ @mock.patch('cinder.rpc.init')
+ @mock.patch('cinder.version.version_string')
+ @mock.patch('cinder.openstack.common.log.getLogger')
+ @mock.patch('cinder.openstack.common.log.setup')
+ @mock.patch('cinder.context.get_admin_context')
+ def test_main_time_error(self, get_admin_context, log_setup, get_logger,
+ version_string, rpc_init,
+ last_completed_audit_period):
+ CONF.set_override('start_time', '2014-01-01 01:00:00')
+ CONF.set_override('end_time', '2013-01-01 01:00:00')
+ last_completed_audit_period.return_value = (mock.sentinel.begin,
+ mock.sentinel.end)
+
+ exit = self.assertRaises(SystemExit, volume_usage_audit.main)
+
+ get_admin_context.assert_called_once_with()
+ self.assertEqual(CONF.project, 'cinder')
+ self.assertEqual(CONF.version, version.version_string())
+ log_setup.assert_called_once_with("cinder")
+ get_logger.assert_called_once_with('cinder')
+ self.assertEqual(exit.code, -1)
+ rpc_init.assert_called_once_with(CONF)
+ last_completed_audit_period.assert_called_once_with()
+
+ @mock.patch('cinder.volume.utils.notify_about_volume_usage')
+ @mock.patch('cinder.db.volume_get_active_by_window')
+ @mock.patch('cinder.utils.last_completed_audit_period')
+ @mock.patch('cinder.rpc.init')
+ @mock.patch('cinder.version.version_string')
+ @mock.patch('cinder.openstack.common.log.getLogger')
+ @mock.patch('cinder.openstack.common.log.setup')
+ @mock.patch('cinder.context.get_admin_context')
+ def test_main_send_create_volume_error(self, get_admin_context, log_setup,
+ get_logger, version_string,
+ rpc_init,
+ last_completed_audit_period,
+ volume_get_active_by_window,
+ notify_about_volume_usage):
+ CONF.set_override('send_actions', True)
+ CONF.set_override('start_time', '2014-01-01 01:00:00')
+ CONF.set_override('end_time', '2014-02-02 02:00:00')
+ begin = datetime.datetime(2014, 1, 1, 1, 0)
+ end = datetime.datetime(2014, 2, 2, 2, 0)
+ ctxt = context.RequestContext('fake-user', 'fake-project')
+ get_admin_context.return_value = ctxt
+ last_completed_audit_period.return_value = (begin, end)
+ volume1_created = datetime.datetime(2014, 1, 1, 2, 0)
+ volume1_deleted = datetime.datetime(2014, 1, 1, 3, 0)
+ volume1 = mock.MagicMock(id='1', project_id='fake-project',
+ created_at=volume1_created,
+ deleted_at=volume1_deleted)
+ volume_get_active_by_window.return_value = [volume1]
+ extra_info = {
+ 'audit_period_beginning': str(begin),
+ 'audit_period_ending': str(end),
+ }
+ local_extra_info = {
+ 'audit_period_beginning': str(volume1.created_at),
+ 'audit_period_ending': str(volume1.created_at),
+ }
+
+ def _notify_about_volume_usage(*args, **kwargs):
+ if 'create.end' in args:
+ raise Exception()
+ else:
+ pass
+
+ notify_about_volume_usage.side_effect = _notify_about_volume_usage
+
+ volume_usage_audit.main()
+
+ get_admin_context.assert_called_once_with()
+ self.assertEqual(CONF.project, 'cinder')
+ self.assertEqual(CONF.version, version.version_string())
+ log_setup.assert_called_once_with("cinder")
+ get_logger.assert_called_once_with('cinder')
+ rpc_init.assert_called_once_with(CONF)
+ last_completed_audit_period.assert_called_once_with()
+ volume_get_active_by_window.assert_called_once_with(ctxt, begin, end)
+ notify_about_volume_usage.assert_any_call(ctxt, volume1, 'exists',
+ extra_usage_info=extra_info)
+ notify_about_volume_usage.assert_any_call(
+ ctxt, volume1, 'create.start', extra_usage_info=local_extra_info)
+ notify_about_volume_usage.assert_any_call(
+ ctxt, volume1, 'create.end', extra_usage_info=local_extra_info)
+
+ @mock.patch('cinder.volume.utils.notify_about_volume_usage')
+ @mock.patch('cinder.db.volume_get_active_by_window')
+ @mock.patch('cinder.utils.last_completed_audit_period')
+ @mock.patch('cinder.rpc.init')
+ @mock.patch('cinder.version.version_string')
+ @mock.patch('cinder.openstack.common.log.getLogger')
+ @mock.patch('cinder.openstack.common.log.setup')
+ @mock.patch('cinder.context.get_admin_context')
+ def test_main_send_delete_volume_error(self, get_admin_context, log_setup,
+ get_logger, version_string,
+ rpc_init,
+ last_completed_audit_period,
+ volume_get_active_by_window,
+ notify_about_volume_usage):
+ CONF.set_override('send_actions', True)
+ CONF.set_override('start_time', '2014-01-01 01:00:00')
+ CONF.set_override('end_time', '2014-02-02 02:00:00')
+ begin = datetime.datetime(2014, 1, 1, 1, 0)
+ end = datetime.datetime(2014, 2, 2, 2, 0)
+ ctxt = context.RequestContext('fake-user', 'fake-project')
+ get_admin_context.return_value = ctxt
+ last_completed_audit_period.return_value = (begin, end)
+ volume1_created = datetime.datetime(2014, 1, 1, 2, 0)
+ volume1_deleted = datetime.datetime(2014, 1, 1, 3, 0)
+ volume1 = mock.MagicMock(id='1', project_id='fake-project',
+ created_at=volume1_created,
+ deleted_at=volume1_deleted)
+ volume_get_active_by_window.return_value = [volume1]
+ extra_info = {
+ 'audit_period_beginning': str(begin),
+ 'audit_period_ending': str(end),
+ }
+ local_extra_info_create = {
+ 'audit_period_beginning': str(volume1.created_at),
+ 'audit_period_ending': str(volume1.created_at),
+ }
+ local_extra_info_delete = {
+ 'audit_period_beginning': str(volume1.deleted_at),
+ 'audit_period_ending': str(volume1.deleted_at),
+ }
+
+ def _notify_about_volume_usage(*args, **kwargs):
+ if 'delete.end' in args:
+ raise Exception()
+ else:
+ pass
+
+ notify_about_volume_usage.side_effect = _notify_about_volume_usage
+
+ volume_usage_audit.main()
+
+ get_admin_context.assert_called_once_with()
+ self.assertEqual(CONF.project, 'cinder')
+ self.assertEqual(CONF.version, version.version_string())
+ log_setup.assert_called_once_with("cinder")
+ get_logger.assert_called_once_with('cinder')
+ rpc_init.assert_called_once_with(CONF)
+ last_completed_audit_period.assert_called_once_with()
+ volume_get_active_by_window.assert_called_once_with(ctxt, begin, end)
+ notify_about_volume_usage.assert_any_call(
+ ctxt, volume1, 'exists', extra_usage_info=extra_info)
+ notify_about_volume_usage.assert_any_call(
+ ctxt, volume1, 'create.start',
+ extra_usage_info=local_extra_info_create)
+ notify_about_volume_usage.assert_any_call(
+ ctxt, volume1, 'create.end',
+ extra_usage_info=local_extra_info_create)
+ notify_about_volume_usage.assert_any_call(
+ ctxt, volume1, 'delete.start',
+ extra_usage_info=local_extra_info_delete)
+ notify_about_volume_usage.assert_any_call(
+ ctxt, volume1, 'delete.end',
+ extra_usage_info=local_extra_info_delete)
+
+ @mock.patch('cinder.volume.utils.notify_about_snapshot_usage')
+ @mock.patch('cinder.db.snapshot_get_active_by_window')
+ @mock.patch('cinder.volume.utils.notify_about_volume_usage')
+ @mock.patch('cinder.db.volume_get_active_by_window')
+ @mock.patch('cinder.utils.last_completed_audit_period')
+ @mock.patch('cinder.rpc.init')
+ @mock.patch('cinder.version.version_string')
+ @mock.patch('cinder.openstack.common.log.getLogger')
+ @mock.patch('cinder.openstack.common.log.setup')
+ @mock.patch('cinder.context.get_admin_context')
+ def test_main_send_snapshot_error(self, get_admin_context,
+ log_setup, get_logger,
+ version_string, rpc_init,
+ last_completed_audit_period,
+ volume_get_active_by_window,
+ notify_about_volume_usage,
+ snapshot_get_active_by_window,
+ notify_about_snapshot_usage):
+ CONF.set_override('send_actions', True)
+ CONF.set_override('start_time', '2014-01-01 01:00:00')
+ CONF.set_override('end_time', '2014-02-02 02:00:00')
+ begin = datetime.datetime(2014, 1, 1, 1, 0)
+ end = datetime.datetime(2014, 2, 2, 2, 0)
+ ctxt = context.RequestContext('fake-user', 'fake-project')
+ get_admin_context.return_value = ctxt
+ last_completed_audit_period.return_value = (begin, end)
+ snapshot1_created = datetime.datetime(2014, 1, 1, 2, 0)
+ snapshot1_deleted = datetime.datetime(2014, 1, 1, 3, 0)
+ snapshot1 = mock.MagicMock(id='1', project_id='fake-project',
+ created_at=snapshot1_created,
+ deleted_at=snapshot1_deleted)
+ volume_get_active_by_window.return_value = []
+ snapshot_get_active_by_window.return_value = [snapshot1]
+ extra_info = {
+ 'audit_period_beginning': str(begin),
+ 'audit_period_ending': str(end),
+ }
+ local_extra_info_create = {
+ 'audit_period_beginning': str(snapshot1.created_at),
+ 'audit_period_ending': str(snapshot1.created_at),
+ }
+ local_extra_info_delete = {
+ 'audit_period_beginning': str(snapshot1.deleted_at),
+ 'audit_period_ending': str(snapshot1.deleted_at),
+ }
+
+ def _notify_about_snapshot_usage(*args, **kwargs):
+ # notify_about_snapshot_usage raises an exception, but does not
+ # block
+ raise Exception()
+
+ notify_about_snapshot_usage.side_effect = _notify_about_snapshot_usage
+
+ volume_usage_audit.main()
+
+ get_admin_context.assert_called_once_with()
+ self.assertEqual(CONF.project, 'cinder')
+ self.assertEqual(CONF.version, version.version_string())
+ log_setup.assert_called_once_with("cinder")
+ get_logger.assert_called_once_with('cinder')
+ rpc_init.assert_called_once_with(CONF)
+ last_completed_audit_period.assert_called_once_with()
+ volume_get_active_by_window.assert_called_once_with(ctxt, begin, end)
+ self.assertFalse(notify_about_volume_usage.called)
+ notify_about_snapshot_usage.assert_any_call(ctxt, snapshot1, 'exists',
+ extra_info)
+ notify_about_snapshot_usage.assert_any_call(
+ ctxt, snapshot1, 'create.start',
+ extra_usage_info=local_extra_info_create)
+ notify_about_snapshot_usage.assert_any_call(
+ ctxt, snapshot1, 'delete.start',
+ extra_usage_info=local_extra_info_delete)
+
+ @mock.patch('cinder.volume.utils.notify_about_snapshot_usage')
+ @mock.patch('cinder.db.snapshot_get_active_by_window')
+ @mock.patch('cinder.volume.utils.notify_about_volume_usage')
+ @mock.patch('cinder.db.volume_get_active_by_window')
+ @mock.patch('cinder.utils.last_completed_audit_period')
+ @mock.patch('cinder.rpc.init')
+ @mock.patch('cinder.version.version_string')
+ @mock.patch('cinder.openstack.common.log.getLogger')
+ @mock.patch('cinder.openstack.common.log.setup')
+ @mock.patch('cinder.context.get_admin_context')
+ def test_main(self, get_admin_context, log_setup, get_logger,
+ version_string, rpc_init, last_completed_audit_period,
+ volume_get_active_by_window, notify_about_volume_usage,
+ snapshot_get_active_by_window, notify_about_snapshot_usage):
+ CONF.set_override('send_actions', True)
+ CONF.set_override('start_time', '2014-01-01 01:00:00')
+ CONF.set_override('end_time', '2014-02-02 02:00:00')
+ begin = datetime.datetime(2014, 1, 1, 1, 0)
+ end = datetime.datetime(2014, 2, 2, 2, 0)
+ ctxt = context.RequestContext('fake-user', 'fake-project')
+ get_admin_context.return_value = ctxt
+ last_completed_audit_period.return_value = (begin, end)
+
+ volume1_created = datetime.datetime(2014, 1, 1, 2, 0)
+ volume1_deleted = datetime.datetime(2014, 1, 1, 3, 0)
+ volume1 = mock.MagicMock(id='1', project_id='fake-project',
+ created_at=volume1_created,
+ deleted_at=volume1_deleted)
+ volume_get_active_by_window.return_value = [volume1]
+ extra_info = {
+ 'audit_period_beginning': str(begin),
+ 'audit_period_ending': str(end),
+ }
+ extra_info_volume_create = {
+ 'audit_period_beginning': str(volume1.created_at),
+ 'audit_period_ending': str(volume1.created_at),
+ }
+ extra_info_volume_delete = {
+ 'audit_period_beginning': str(volume1.deleted_at),
+ 'audit_period_ending': str(volume1.deleted_at),
+ }
+
+ snapshot1_created = datetime.datetime(2014, 1, 1, 2, 0)
+ snapshot1_deleted = datetime.datetime(2014, 1, 1, 3, 0)
+ snapshot1 = mock.MagicMock(id='1', project_id='fake-project',
+ created_at=snapshot1_created,
+ deleted_at=snapshot1_deleted)
+ snapshot_get_active_by_window.return_value = [snapshot1]
+ extra_info_snapshot_create = {
+ 'audit_period_beginning': str(snapshot1.created_at),
+ 'audit_period_ending': str(snapshot1.created_at),
+ }
+ extra_info_snapshot_delete = {
+ 'audit_period_beginning': str(snapshot1.deleted_at),
+ 'audit_period_ending': str(snapshot1.deleted_at),
+ }
+
+ volume_usage_audit.main()
+
+ get_admin_context.assert_called_once_with()
+ self.assertEqual(CONF.project, 'cinder')
+ self.assertEqual(CONF.version, version.version_string())
+ log_setup.assert_called_once_with("cinder")
+ get_logger.assert_called_once_with('cinder')
+ rpc_init.assert_called_once_with(CONF)
+ last_completed_audit_period.assert_called_once_with()
+ volume_get_active_by_window.assert_called_once_with(ctxt, begin, end)
+ notify_about_volume_usage.assert_any_call(
+ ctxt, volume1, 'exists', extra_usage_info=extra_info)
+ notify_about_volume_usage.assert_any_call(
+ ctxt, volume1, 'create.start',
+ extra_usage_info=extra_info_volume_create)
+ notify_about_volume_usage.assert_any_call(
+ ctxt, volume1, 'create.end',
+ extra_usage_info=extra_info_volume_create)
+ notify_about_volume_usage.assert_any_call(
+ ctxt, volume1, 'delete.start',
+ extra_usage_info=extra_info_volume_delete)
+ notify_about_volume_usage.assert_any_call(
+ ctxt, volume1, 'delete.end',
+ extra_usage_info=extra_info_volume_delete)
+
+ notify_about_snapshot_usage.assert_any_call(ctxt, snapshot1,
+ 'exists', extra_info)
+ notify_about_snapshot_usage.assert_any_call(
+ ctxt, snapshot1, 'create.start',
+ extra_usage_info=extra_info_snapshot_create)
+ notify_about_snapshot_usage.assert_any_call(
+ ctxt, snapshot1, 'create.end',
+ extra_usage_info=extra_info_snapshot_create)
+ notify_about_snapshot_usage.assert_any_call(
+ ctxt, snapshot1, 'delete.start',
+ extra_usage_info=extra_info_snapshot_delete)
+ notify_about_snapshot_usage.assert_any_call(
+ ctxt, snapshot1, 'delete.end',
+ extra_usage_info=extra_info_snapshot_delete)
Code Review / openstack-build / cinder-build.git / commitdiff