<commit_before>#   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 webob

from nova.api.openstack import common
from nova.api.openstack.compute.contrib import admin_actions
from nova.compute import vm_states
import nova.context
from nova import exception
from nova.objects import instance as instance_obj
from nova.openstack.common import jsonutils
from nova.openstack.common import timeutils
from nova.openstack.common import uuidutils
from nova import test
from nova.tests.api.openstack import fakes
from nova.tests import fake_instance


class CommonMixin(object):
    def setUp(self):
        super(CommonMixin, self).setUp()
        self.controller = admin_actions.AdminActionsController()
        self.compute_api = self.controller.compute_api
        self.context = nova.context.RequestContext('fake', 'fake')

        def _fake_controller(*args, **kwargs):
            return self.controller

        self.stubs.Set(admin_actions, 'AdminActionsController',
                       _fake_controller)

        self.flags(
            osapi_compute_extension=[
                'nova.api.openstack.compute.contrib.select_extensions'],
            osapi_compute_ext_list=['Admin_actions'])

        self.app = fakes.wsgi_app(init_only=('servers',),
                                  fake_auth_context=self.context)
        self.mox.StubOutWithMock(self.compute_api, 'get')

    def _make_request(self, url, body):
        req = webob.Request.blank('/v2/fake' + url)
        req.method = 'POST'
        req.body = jsonutils.dumps(body)
        req.content_type = 'application/json'
        return req.get_response(self.app)

    def _stub_instance_get(self, uuid=None):
        if uuid is None:
            uuid = uuidutils.generate_uuid()
        instance = fake_instance.fake_db_instance(
                id=1, uuid=uuid, vm_state=vm_states.ACTIVE,
                task_state=None, launched_at=timeutils.utcnow())
        instance = instance_obj.Instance._from_db_object(
                self.context, instance_obj.Instance(), instance)
        self.compute_api.get(self.context, uuid,
                             want_objects=True).AndReturn(instance)
        return instance

    def _stub_instance_get_failure(self, exc_info, uuid=None):
        if uuid is None:
            uuid = uuidutils.generate_uuid()
        self.compute_api.get(self.context, uuid,
                             want_objects=True).AndRaise(exc_info)
        return uuid

    def _test_non_existing_instance(self, action, body_map=None):
        uuid = uuidutils.generate_uuid()
        self._stub_instance_get_failure(
                exception.InstanceNotFound(instance_id=uuid), uuid=uuid)

        self.mox.ReplayAll()

        res = self._make_request('/servers/%s/action' % uuid,
                                 {action: body_map.get(action)})
        self.assertEqual(404, res.status_int)
        # Do these here instead of tearDown because this method is called
        # more than once for the same test case
        self.mox.VerifyAll()
        self.mox.UnsetStubs()

    def _test_action(self, action, body=None, method=None):
        if method is None:
            method = action

        instance = self._stub_instance_get()
        getattr(self.compute_api, method)(self.context, instance)

        self.mox.ReplayAll()

        res = self._make_request('/servers/%s/action' % instance['uuid'],
                                 {action: None})
        self.assertEqual(202, res.status_int)
        # Do these here instead of tearDown because this method is called
        # more than once for the same test case
        self.mox.VerifyAll()
        self.mox.UnsetStubs()

    def _test_invalid_state(self, action, method=None, body_map=None,
                            compute_api_args_map=None):
        if method is None:
            method = action
        if body_map is None:
            body_map = {}
        if compute_api_args_map is None:
            compute_api_args_map = {}

        instance = self._stub_instance_get()

        args, kwargs = compute_api_args_map.get(action, ((), {}))

        getattr(self.compute_api, method)(self.context, instance,
                                          *args, **kwargs).AndRaise(
                exception.InstanceInvalidState(
                    attr='vm_state', instance_uuid=instance['uuid'],
                    state='foo', method=method))

        self.mox.ReplayAll()

        res = self._make_request('/servers/%s/action' % instance['uuid'],
                                 {action: body_map.get(action)})
        self.assertEqual(409, res.status_int)
        self.assertIn("Cannot \'%s\' while instance" % action, res.body)
        # Do these here instead of tearDown because this method is called
        # more than once for the same test case
        self.mox.VerifyAll()
        self.mox.UnsetStubs()


class AdminActionsTest(CommonMixin, test.NoDBTestCase):
    def test_actions(self):
        actions = ['pause', 'unpause', 'suspend', 'resume', 'migrate',
                   'resetNetwork', 'injectNetworkInfo', 'lock',
                   'unlock']
        method_translations = {'migrate': 'resize',
                               'resetNetwork': 'reset_network',
                               'injectNetworkInfo': 'inject_network_info'}

        for action in actions:
            method = method_translations.get(action)
            self.mox.StubOutWithMock(self.compute_api, method or action)
            self._test_action(action, method=method)
            # Re-mock this.
            self.mox.StubOutWithMock(self.compute_api, 'get')

    def test_actions_raise_conflict_on_invalid_state(self):
        actions = ['pause', 'unpause', 'suspend', 'resume', 'migrate']
        method_translations = {'migrate': 'resize'}

        for action in actions:
            method = method_translations.get(action)
            self.mox.StubOutWithMock(self.compute_api, method or action)
            self._test_invalid_state(action, method=method)
            # Re-mock this.
            self.mox.StubOutWithMock(self.compute_api, 'get')

    def test_actions_with_non_existed_instance(self):
        actions = ['pause', 'unpause', 'suspend', 'resume',
                   'resetNetwork', 'injectNetworkInfo', 'lock',
                   'unlock', 'os-resetState', 'migrate']
        body_map = {'os-resetState': {'state': 'active'},
                    'os-migrateLive':
                                  {'host': 'hostname',
                                   'block_migration': False,
                                   'disk_over_commit': False}}
        for action in actions:
            self._test_non_existing_instance(action,
                                             body_map=body_map)
            # Re-mock this.<commit_msg>Supplement 'os-migrateLive' in actions list<commit_after>#   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 webob

from nova.api.openstack import common
from nova.api.openstack.compute.contrib import admin_actions
from nova.compute import vm_states
import nova.context
from nova import exception
from nova.objects import instance as instance_obj
from nova.openstack.common import jsonutils
from nova.openstack.common import timeutils
from nova.openstack.common import uuidutils
from nova import test
from nova.tests.api.openstack import fakes
from nova.tests import fake_instance


class CommonMixin(object):
    def setUp(self):
        super(CommonMixin, self).setUp()
        self.controller = admin_actions.AdminActionsController()
        self.compute_api = self.controller.compute_api
        self.context = nova.context.RequestContext('fake', 'fake')

        def _fake_controller(*args, **kwargs):
            return self.controller

        self.stubs.Set(admin_actions, 'AdminActionsController',
                       _fake_controller)

        self.flags(
            osapi_compute_extension=[
                'nova.api.openstack.compute.contrib.select_extensions'],
            osapi_compute_ext_list=['Admin_actions'])

        self.app = fakes.wsgi_app(init_only=('servers',),
                                  fake_auth_context=self.context)
        self.mox.StubOutWithMock(self.compute_api, 'get')

    def _make_request(self, url, body):
        req = webob.Request.blank('/v2/fake' + url)
        req.method = 'POST'
        req.body = jsonutils.dumps(body)
        req.content_type = 'application/json'
        return req.get_response(self.app)

    def _stub_instance_get(self, uuid=None):
        if uuid is None:
            uuid = uuidutils.generate_uuid()
        instance = fake_instance.fake_db_instance(
                id=1, uuid=uuid, vm_state=vm_states.ACTIVE,
                task_state=None, launched_at=timeutils.utcnow())
        instance = instance_obj.Instance._from_db_object(
                self.context, instance_obj.Instance(), instance)
        self.compute_api.get(self.context, uuid,
                             want_objects=True).AndReturn(instance)
        return instance

    def _stub_instance_get_failure(self, exc_info, uuid=None):
        if uuid is None:
            uuid = uuidutils.generate_uuid()
        self.compute_api.get(self.context, uuid,
                             want_objects=True).AndRaise(exc_info)
        return uuid

    def _test_non_existing_instance(self, action, body_map=None):
        uuid = uuidutils.generate_uuid()
        self._stub_instance_get_failure(
                exception.InstanceNotFound(instance_id=uuid), uuid=uuid)

        self.mox.ReplayAll()

        res = self._make_request('/servers/%s/action' % uuid,
                                 {action: body_map.get(action)})
        self.assertEqual(404, res.status_int)
        # Do these here instead of tearDown because this method is called
        # more than once for the same test case
        self.mox.VerifyAll()
        self.mox.UnsetStubs()

    def _test_action(self, action, body=None, method=None):
        if method is None:
            method = action

        instance = self._stub_instance_get()
        getattr(self.compute_api, method)(self.context, instance)

        self.mox.ReplayAll()

        res = self._make_request('/servers/%s/action' % instance['uuid'],
                                 {action: None})
        self.assertEqual(202, res.status_int)
        # Do these here instead of tearDown because this method is called
        # more than once for the same test case
        self.mox.VerifyAll()
        self.mox.UnsetStubs()

    def _test_invalid_state(self, action, method=None, body_map=None,
                            compute_api_args_map=None):
        if method is None:
            method = action
        if body_map is None:
            body_map = {}
        if compute_api_args_map is None:
            compute_api_args_map = {}

        instance = self._stub_instance_get()

        args, kwargs = compute_api_args_map.get(action, ((), {}))

        getattr(self.compute_api, method)(self.context, instance,
                                          *args, **kwargs).AndRaise(
                exception.InstanceInvalidState(
                    attr='vm_state', instance_uuid=instance['uuid'],
                    state='foo', method=method))

        self.mox.ReplayAll()

        res = self._make_request('/servers/%s/action' % instance['uuid'],
                                 {action: body_map.get(action)})
        self.assertEqual(409, res.status_int)
        self.assertIn("Cannot \'%s\' while instance" % action, res.body)
        # Do these here instead of tearDown because this method is called
        # more than once for the same test case
        self.mox.VerifyAll()
        self.mox.UnsetStubs()


class AdminActionsTest(CommonMixin, test.NoDBTestCase):
    def test_actions(self):
        actions = ['pause', 'unpause', 'suspend', 'resume', 'migrate',
                   'resetNetwork', 'injectNetworkInfo', 'lock',
                   'unlock']
        method_translations = {'migrate': 'resize',
                               'resetNetwork': 'reset_network',
                               'injectNetworkInfo': 'inject_network_info'}

        for action in actions:
            method = method_translations.get(action)
            self.mox.StubOutWithMock(self.compute_api, method or action)
            self._test_action(action, method=method)
            # Re-mock this.
            self.mox.StubOutWithMock(self.compute_api, 'get')

    def test_actions_raise_conflict_on_invalid_state(self):
        actions = ['pause', 'unpause', 'suspend', 'resume', 'migrate']
        method_translations = {'migrate': 'resize'}

        for action in actions:
            method = method_translations.get(action)
            self.mox.StubOutWithMock(self.compute_api, method or action)
            self._test_invalid_state(action, method=method)
            # Re-mock this.
            self.mox.StubOutWithMock(self.compute_api, 'get')

    def test_actions_with_non_existed_instance(self):
        actions = ['pause', 'unpause', 'suspend', 'resume',
                   'resetNetwork', 'injectNetworkInfo', 'lock',
                   'unlock', 'os-resetState', 'migrate', 'os-migrateLive']
        body_map = {'os-resetState': {'state': 'active'},
                    'os-migrateLive':
                                  {'host': 'hostname',
                                   'block_migration': False,
                                   'disk_over_commit': False}}
        for action in actions:
            self._test_non_existing_instance(action,
                                             body_map=body_map)
            # Re-mock this.
{'fraction_non_alphanumeric': 0.09676327193932828, 'fraction_numerical': 0.0025731310942578548, 'mean_word_length': 2.233143607705779, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 22, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4668846', 'n_tokens_mistral': 4078, 'n_tokens_neox': 4059, 'n_words': 946}