text: #   Copyright 2011 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.
  
  import datetime
  import uuid
  
  from oslo.config import cfg
  import webob
  
  from nova.api.openstack import compute
  from nova.api.openstack.compute.contrib import admin_actions
  from nova.compute import api as compute_api
  from nova.compute import vm_states
  from nova import context
  from nova import exception
  from nova.openstack.common import jsonutils
  from nova.scheduler import rpcapi as scheduler_rpcapi
  from nova import test
  from nova.tests.api.openstack import fakes
  
  
  CONF = cfg.CONF
  
  INSTANCE = {
               "id": 1,
               "name": "fake",
               "display_name": "test_server",
               "uuid": "abcd",
               "user_id": 'fake_user_id',
               "tenant_id": 'fake_tenant_id',
               "created_at": datetime.datetime(2010, 10, 10, 12, 0, 0),
               "updated_at": datetime.datetime(2010, 11, 11, 11, 0, 0),
               "security_groups": [{"id": 1, "name": "test"}],
               "progress": 0,
               "image_ref": 'http://foo.com/123',
               "fixed_ips": [],
               "instance_type": {"flavorid": '124'},
          }
  
  
  def fake_compute_api(*args, **kwargs):
      return True
  
  
  def fake_compute_api_raises_invalid_state(*args, **kwargs):
      raise exception.InstanceInvalidState(attr='fake_attr',
              state='fake_state', method='fake_method',
              instance_uuid='fake')
  
  
  def fake_compute_api_get(self, context, instance_id):
      return {'id': 1, 'uuid': instance_id, 'vm_state': vm_states.ACTIVE,
              'task_state': None}
  
  
  class AdminActionsTest(test.TestCase):
  
      _actions = ('pause', 'unpause', 'suspend', 'resume', 'migrate',
                  'resetNetwork', 'injectNetworkInfo', 'lock', 'unlock')
  
      _methods = ('pause', 'unpause', 'suspend', 'resume', 'resize',
                  'reset_network', 'inject_network_info', 'lock', 'unlock')
  
      _actions_that_check_state = (
              # action, method
              ('pause', 'pause'),
              ('unpause', 'unpause'),
              ('suspend', 'suspend'),
              ('resume', 'resume'),
              ('migrate', 'resize'))
  
      def setUp(self):
          super(AdminActionsTest, self).setUp()
          self.stubs.Set(compute_api.API, 'get', fake_compute_api_get)
          self.UUID = uuid.uuid4()
          for _method in self._methods:
              self.stubs.Set(compute_api.API, _method, fake_compute_api)
          self.flags(
              osapi_compute_extension=[
                  'nova.api.openstack.compute.contrib.select_extensions'],
              osapi_compute_ext_list=['Admin_actions'])
  
      def test_admin_api_actions(self):
          app = fakes.wsgi_app(init_only=('servers',))
          for _action in self._actions:
              req = webob.Request.blank('/v2/fake/servers/%s/action' %
                      self.UUID)
              req.method = 'POST'
              req.body = jsonutils.dumps({_action: None})
              req.content_type = 'application/json'
              res = req.get_response(app)
              self.assertEqual(res.status_int, 202)
  
      def test_admin_api_actions_raise_conflict_on_invalid_state(self):
          app = fakes.wsgi_app(init_only=('servers',))
  
          for _action, _method in self._actions_that_check_state:
              self.stubs.Set(compute_api.API, _method,
                  fake_compute_api_raises_invalid_state)
  
              req = webob.Request.blank('/v2/fake/servers/%s/action' %
                      self.UUID)
              req.method = 'POST'
              req.body = jsonutils.dumps({_action: None})
              req.content_type = 'application/json'
              res = req.get_response(app)
              self.assertEqual(res.status_int, 409)
              self.assertIn("Cannot \'%(_action)s\' while instance" % locals(),
                      res.body)
  
      def test_migrate_live_enabled(self):
          ctxt = context.get_admin_context()
          ctxt.user_id = 'fake'
          ctxt.project_id = 'fake'
          ctxt.is_admin = True
          app = fakes.wsgi_app(fake_auth_context=ctxt, init_only=('servers',))
          req = webob.Request.blank('/v2/fake/servers/%s/action' % self.UUID)
          req.method = 'POST'
          req.body = jsonutils.dumps({
              'os-migrateLive': {
                  'host': 'hostname',
                  'block_migration': False,
                  'disk_over_commit': False,
              }
          })
          req.content_type = 'application/json'
  
          def fake_update(inst, context, instance,
                          task_state, expected_task_state):
              return None
  
          def fake_scheduler_api_live_migration(self, context, dest,
                                          block_migration=False,
                                          disk_over_commit=False, instance=None,
                                          instance_id=None, topic=None):
              return None
  
          self.stubs.Set(compute_api.API, 'update', fake_update)
          self.stubs.Set(scheduler_rpcapi.SchedulerAPI,
                         'live_migration',
                         fake_scheduler_api_live_migration)
  
          res = req.get_response(app)
          self.assertEqual(res.status_int, 202)
  
      def test_migrate_live_missing_dict_param(self):
          ctxt = context.get_admin_context()
          ctxt.user_id = 'fake'
          ctxt.project_id = 'fake'
          ctxt.is_admin = True
          app = fakes.wsgi_app(fake_auth_context=ctxt, init_only=('servers',))
          req = webob.Request.blank('/v2/fake/servers/%s/action' % self.UUID)
          req.method = 'POST'
          req.body = jsonutils.dumps({
              'os-migrateLive': {
                  'dummy': 'hostname',
                  'block_migration': False,
                  'disk_over_commit': False,
              }
          })
          req.content_type = 'application/json'
          res = req.get_response(app)
          self.assertEqual(res.status_int, 400)
  
      def test_migrate_live_compute_service_unavailable(self):
          ctxt = context.get_admin_context()
          ctxt.user_id = 'fake'
          ctxt.project_id = 'fake'
          ctxt.is_admin = True
          app = fakes.wsgi_app(fake_auth_context=ctxt, init_only=('servers',))
          req = webob.Request.blank('/v2/fake/servers/%s/action' % self.UUID)
          req.method = 'POST'
          req.body = jsonutils.dumps({
              'os-migrateLive': {
                  'host': 'hostname',
                  'block_migration': False,
                  'disk_over_commit': False,
              }
          })
          req.content_type = 'application/json'
  
          def fake_update(inst, context, instance,
                          task_state, expected_task_state):
              return None
  
          def fake_scheduler_api_live_migration(context, dest,
                                          block_migration=False,
                                          disk_over_commit=False, instance=None,
                                          instance_id=None, topic=None):
              raise exception.ComputeServiceUnavailable(host='host')
  
          self.stubs.Set(compute_api.API, 'update', fake_update)
          self.stubs.Set(scheduler_rpcapi.SchedulerAPI,
                         'live_migration',
                         fake_scheduler_api_live_migration)
  
          res = req.get_response(app)
          self.assertEqual(res.status_int, 400)
          self.assertIn(
              unicode(exception.ComputeServiceUnavailable(host='host')),
              res.body)
  
      def test_migrate_live_invalid_hypervisor_type(self):
          ctxt = context.get_admin_context()
          ctxt.user_id = 'fake'
          ctxt.project_id = 'fake'
          ctxt.is_admin = True
          app = fakes.wsgi_app(fake_auth_context=ctxt, init_only=('servers',))
          req = webob.Request.blank('/v2/fake/servers/%s/action' % self.UUID)
          req.method = 'POST'
          req.body = jsonutils.dumps({
              'os-migrateLive': {
                  'host': 'hostname',
                  'block_migration': False,
                  'disk_over_commit': False,
              }
          })
          req.content_type = 'application/json'
  
          def fake_update(inst, context, instance,
                          task_state, expected_task_state):
              return None
  
          def fake_scheduler_api_live_migration(context, dest,
                                          block_migration=False,
                                          disk_over_commit=False, instance=None,
                                          instance_id=None, topic=None):
              raise exception.InvalidHypervisorType()
  
          self.stubs.Set(compute_api.API, 'update', fake_update)
          self.stubs.Set(scheduler_rpcapi.SchedulerAPI,
                         'live_migration',
                         fake_scheduler_api_live_migration)
  
          res = req.get_response(app)
          self.assertEqual(res.status_int, 400)
          self.assertIn(
              unicode(exception.InvalidHypervisorType()),
              res.body)
  
      def test_migrate_live_unable_to_migrate_to_self(self):
          ctxt = context.get_admin_context()
          ctxt.user_id = 'fake'
          ctxt.project_id = 'fake'
          ctxt.is_admin = True
          app = fakes.wsgi_app(fake_auth_context=ctxt, init_only=('servers',))
          req = webob.Request.blank('/v2/fake/servers/%s/action' % self.UUID)
          req.method = 'POST'
          req.body = jsonutils.dumps({
              'os-migrateLive': {
                  'host': 'hostname',
                  'block_migration': False,
                  'disk_over_commit': False,
              }
          })
          req.content_type = 'application/json'
  
          def fake_update(inst, context, instance,
                          task_state, expected_task_state):
              return None
  
          def fake_scheduler_api_live_migration(context, dest,
                                          block_migration=False,
                                          disk_over_commit=False, instance=None,
                                          instance_id=None, topic=None):
              raise exception.UnableToMigrateToSelf(self.UUID, host='host')
  
          self.stubs.Set(compute_api.API, 'update', fake_update)
          self.stubs.Set(scheduler_rpcapi.SchedulerAPI,
                         'live_migration',
                         fake_scheduler_api_live_migration)
  
          res = req.get_response(app)
          self.assertEqual(res.status_int, 400)
          self.assertIn(
              unicode(exception.UnableToMigrateToSelf(self.UUID, host='host')),
              res.body)
  
      def test_migrate_live_destination_hypervisor_too_old(self):
          ctxt = context.get_admin_context()
          ctxt.user_id = 'fake'
          ctxt.project_id = 'fake'
          ctxt.is_admin = True
          app = fakes.wsgi_app(fake_auth_context=ctxt, init_only=('servers',))
          req = webob.Request.blank('/v2/fake/servers/%s/action' % self.UUID)
          req.method = 'POST'
          req.body = jsonutils.dumps({
              'os-migrateLive': {
                  'host': 'hostname',
                  'block_migration': False,
                  'disk_over_commit': False,
              }
          })
          req.content_type = 'application/json'
  
          def fake_update(inst, context, instance,
                          task_state, expected_task_state):
              return None
  
          def fake_scheduler_api_live_migration(context, dest,
                                          block_migration=False,
                                          disk_over_commit=False, instance=None,
                                          instance_id=None, topic=None):
              raise exception.DestinationHypervisorTooOld()
  
          self.stubs.Set(compute_api.API, 'update', fake_update)
          self.stubs.Set(scheduler_rpcapi.SchedulerAPI,
                         'live_migration',
                         fake_scheduler_api_live_migration)
  
          res = req.get_response(app)
          self.assertEqual(res.status_int, 400)
          self.assertIn(
              unicode(exception.DestinationHypervisorTooOld()),
              res.body)
  
  
  class CreateBackupTests(test.TestCase):
  
      def setUp(self):
          super(CreateBackupTests, self).setUp()
  
          self.stubs.Set(compute_api.API, 'get', fake_compute_api_get)
          self.backup_stubs = fakes.stub_out_compute_api_backup(self.stubs)
          self.app = compute.APIRouter(init_only=('servers',))
          self.uuid = uuid.uuid4()
  
      def _get_request(self, body):
          url = '/fake/servers/%s/action' % self.uuid
          req = fakes.HTTPRequest.blank(url)
          req.method = 'POST'
          req.content_type = 'application/json'
          req.body = jsonutils.dumps(body)
          return req
  
      def test_create_backup_with_metadata(self):
          body = {
              'createBackup': {
                  'name': 'Backup 1',
                  'backup_type': 'daily',
                  'rotation': 1,
                  'metadata': {'123': 'asdf'},
              },
          }
  
          request = self._get_request(body)
          response = request.get_response(self.app)
  
          self.assertEqual(response.status_int, 202)
          self.assertTrue(response.headers['Location'])
  
      def test_create_backup_with_too_much_metadata(self):
          body = {
              'createBackup': {
                  'name': 'Backup 1',
                  'backup_type': 'daily',
                  'rotation': 1,
                  'metadata': {'123': 'asdf'},
              },
          }
          for num in range(CONF.quota_metadata_items + 1):
              body['createBackup']['metadata']['foo%i' % num] = "bar"
  
          request = self._get_request(body)
          response = request.get_response(self.app)
          self.assertEqual(response.status_int, 413)
  
      def test_create_backup_no_name(self):
          # Name is required for backups.
          body = {
              'createBackup': {
                  'backup_type': 'daily',
                  'rotation': 1,
              },
          }
  
          request = self._get_request(body)
          response = request.get_response(self.app)
          self.assertEqual(response.status_int, 400)
  
      def test_create_backup_no_rotation(self):
          # Rotation is required for backup requests.
          body = {
              'createBackup': {
                  'name': 'Backup 1',
                  'backup_type': 'daily',
              },
          }
  
          request = self._get_request(body)
          response = request.get_response(self.app)
          self.assertEqual(response.status_int, 400)
  
      def test_create_backup_negative_rotation(self):
          """Rotation must be greater than or equal to zero
          for backup requests
          """
          body = {
              'createBackup': {
                  'name': 'Backup 1',
                  'backup_type': 'daily',
                  'rotation': -1,
              },
          }
  
          request = self._get_request(body)
          response = request.get_response(self.app)
          self.assertEqual(response.status_int, 400)
  
      def test_create_backup_no_backup_type(self):
          # Backup Type (daily or weekly) is required for backup requests.
          body = {
              'createBackup': {
                  'name': 'Backup 1',
                  'rotation': 1,
              },
          }
  
          request = self._get_request(body)
          response = request.get_response(self.app)
          self.assertEqual(response.status_int, 400)
  
      def test_create_backup_bad_entity(self):
          body = {'createBackup': 'go'}
  
          request = self._get_request(body)
          response = request.get_response(self.app)
          self.assertEqual(response.status_int, 400)
  
      def test_create_backup_rotation_is_zero(self):
          # The happy path for creating backups if rotation is zero.
          body = {
              'createBackup': {
                  'name': 'Backup 1',
                  'backup_type': 'daily',
                  'rotation': 0,
              },
          }
  
          request = self._get_request(body)
          response = request.get_response(self.app)
  
          self.assertEqual(response.status_int, 202)
          self.assertFalse('Location' in response.headers)
  
      def test_create_backup_rotation_is_positive(self):
          # The happy path for creating backups if rotation is positive.
          body = {
              'createBackup': {
                  'name': 'Backup 1',
                  'backup_type': 'daily',
                  'rotation': 1,
              },
          }
  
          request = self._get_request(body)
          response = request.get_response(self.app)
  
          self.assertEqual(response.status_int, 202)
          self.assertTrue(response.headers['Location'])
  
      def test_create_backup_raises_conflict_on_invalid_state(self):
          body = {
              'createBackup': {
                  'name': 'Backup 1',
                  'backup_type': 'daily',
                  'rotation': 1,
              },
          }
  
          self.stubs.Set(compute_api.API, 'backup',
                  fake_compute_api_raises_invalid_state)
  
          request = self._get_request(body)
          response = request.get_response(self.app)
          self.assertEqual(response.status_int, 409)
  
  
  class ResetStateTests(test.TestCase):
      def setUp(self):
          super(ResetStateTests, self).setUp()
  
          self.exists = True
          self.kwargs = None
          self.uuid = uuid.uuid4()
  
          def fake_get(inst, context, instance_id):
              if self.exists:
                  return dict(id=1, uuid=instance_id, vm_state=vm_states.ACTIVE)
              raise exception.InstanceNotFound(instance_id=instance_id)
  
          def fake_update(inst, context, instance, **kwargs):
              self.kwargs = kwargs
  
          self.stubs.Set(compute_api.API, 'get', fake_get)
          self.stubs.Set(compute_api.API, 'update', fake_update)
          self.admin_api = admin_actions.AdminActionsController()
  
          url = '/fake/servers/%s/action' % self.uuid
          self.request = fakes.HTTPRequest.blank(url)
  
      def test_no_state(self):
          self.assertRaises(webob.exc.HTTPBadRequest,
                            self.admin_api._reset_state,
                            self.request, 'inst_id',
                            {"os-resetState": None})
  
      def test_bad_state(self):
          self.assertRaises(webob.exc.HTTPBadRequest,
                            self.admin_api._reset_state,
                            self.request, 'inst_id',
                            {"os-resetState": {"state": "spam"}})
  
      def test_no_instance(self):
          self.exists = False
          self.assertRaises(webob.exc.HTTPNotFound,
                            self.admin_api._reset_state,
                            self.request, 'inst_id',
                            {"os-resetState": {"state": "active"}})
  
      def test_reset_active(self):
          body = {"os-resetState": {"state": "active"}}
          result = self.admin_api._reset_state(self.request, 'inst_id', body)
  
          self.assertEqual(result.status_int, 202)
          self.assertEqual(self.kwargs, dict(vm_state=vm_states.ACTIVE,
                                             task_state=None))
  
      def test_reset_error(self):
          body = {"os-resetState": {"state": "error"}}
          result = self.admin_api._reset_state(self.request, 'inst_id', body)
  
          self.assertEqual(result.status_int, 202)
          self.assertEqual(self.kwargs, dict(vm_state=vm_states.ERROR,
                                             task_state=None))
  