# 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 mock from oslo_utils import uuidutils from webob import exc from karbor.api.v1 import plans as plan_api from karbor.api.v1 import scheduled_operations as operation_api from karbor.api.v1 import triggers as trigger_api from karbor import context from karbor import exception from karbor.tests import base from karbor.tests.unit.api import fakes from karbor.tests.unit.api.v1 import test_triggers class FakeRemoteOperationApi(object): def __init__(self): super(FakeRemoteOperationApi, self).__init__() self._create_operation_exception = None self._delete_operation_exception = None def create_scheduled_operation(self, context, operation): if self._create_operation_exception: raise self._create_operation_exception def delete_scheduled_operation(self, context, operation_id, trigger_id): if self._delete_operation_exception: raise self._delete_operation_exception class ScheduledOperationApiTest(base.TestCase): def setUp(self): super(ScheduledOperationApiTest, self).setUp() self.remote_operation_api = FakeRemoteOperationApi() self.controller = operation_api.ScheduledOperationController() self.controller.operationengine_api = self.remote_operation_api self.ctxt = context.RequestContext('demo', 'fakeproject', True) self.req = fakes.HTTPRequest.blank('/v1/scheduled_operations') trigger = self._create_trigger() self._plan = self._create_plan(uuidutils.generate_uuid()) self.default_create_operation_param = { "name": "123", "description": "123", "operation_type": "protect", "trigger_id": trigger['trigger_info']['id'], "operation_definition": { "plan_id": self._plan['id'], "provider_id": self._plan['provider_id'] }, } def test_create_operation_InvalidBody(self): self.assertRaises(exception.ValidationError, self.controller.create, self.req, body={}) def test_create_operation_InvalidName(self): body = self._get_create_operation_request_body() self.assertRaises(exception.ValidationError, self.controller.create, self.req, body=body) def test_create_operation_invalid_trigger(self): param = self.default_create_operation_param.copy() param['trigger_id'] = 123 body = self._get_create_operation_request_body(param) self.assertRaises(exception.ValidationError, self.controller.create, self.req, body=body) def test_create_operation_receive_invalid_except(self): self.remote_operation_api._create_operation_exception =\ exception.TriggerIsInvalid(trigger_id=None) param = self.default_create_operation_param.copy() body = self._get_create_operation_request_body(param) self.assertRaises(exc.HTTPBadRequest, self.controller.create, self.req, body=body) self.remote_operation_api._create_operation_exception = None def test_create_operation_receive_unknown_except(self): self.remote_operation_api._create_operation_exception =\ exception.TriggerNotFound(id=None) param = self.default_create_operation_param.copy() body = self._get_create_operation_request_body(param) self.assertRaises(exc.HTTPInternalServerError, self.controller.create, self.req, body=body) self.remote_operation_api._create_operation_exception = None def test_create_operation(self): name = 'my protect' param = self.default_create_operation_param.copy() param['name'] = name body = self._get_create_operation_request_body(param) operation = self.controller.create(self.req, body=body) self.assertEqual(name, operation['scheduled_operation']['name']) def test_create_retention_scheduled_operation(self): name = 'my retention protect' param = self.default_create_operation_param.copy() param['name'] = name param['operation_definition']['retention_duration'] = 100 param['operation_definition']['max_backups'] = 3 body = self._get_create_operation_request_body(param) operation = self.controller.create(self.req, body=body) self.assertEqual(name, operation['scheduled_operation']['name']) def test_delete_operation_receive_NotFound_except(self): self.remote_operation_api._delete_operation_exception =\ exception.ScheduledOperationStateNotFound(op_id=None) operation = self._create_one_operation() self.assertRaises(exc.HTTPInternalServerError, self.controller.delete, self.req, operation['scheduled_operation']['id']) self.remote_operation_api._delete_operation_exception = None def test_delete_operation(self): operation = self._create_one_operation() self.controller.delete(self.req, operation['scheduled_operation']['id']) self.assertRaises(exc.HTTPNotFound, self.controller.show, self.req, operation['scheduled_operation']['id']) def test_show_operation_not_exist(self): self.assertRaises(exc.HTTPNotFound, self.controller.show, self.req, '2a9ce1f3-cc1a-4516-9435-0ebb13caa398') def test_show_operation_invalid_id(self): self.assertRaises(exc.HTTPBadRequest, self.controller.show, self.req, 1) def test_show_operation(self): operation = self._create_one_operation() operation1 = self.controller.show( self.req, operation['scheduled_operation']['id']) self.assertEqual(operation['scheduled_operation']['id'], operation1['scheduled_operation']['id']) def test_list_operation(self): operation = self._create_one_operation() operations = self.controller.index(self.req) for item in operations['operations']: if item['id'] == operation['scheduled_operation']['id']: self.assertTrue(1) self.assertFalse(0) def _create_one_operation(self): param = self.default_create_operation_param.copy() body = self._get_create_operation_request_body(param) return self.controller.create(self.req, body=body) def _get_create_operation_request_body(self, param={}): return {"scheduled_operation": param} def _create_trigger(self): create_trigger_param = { "trigger_info": { "name": "123", "type": "time", "properties": { "format": "crontab", "pattern": "* * * * *" }, } } controller = trigger_api.TriggersController() controller.operationengine_api = test_triggers.FakeRemoteOperationApi() req = fakes.HTTPRequest.blank('/v1/triggers') return controller.create(req, body=create_trigger_param) @mock.patch( 'karbor.context.RequestContext.can') @mock.patch( 'karbor.services.protection.rpcapi.ProtectionAPI.show_provider') def _create_plan(self, provider_id, mock_provider, mock_policy): create_plan_param = { 'plan': { 'name': '123', 'provider_id': provider_id, 'resources': [ {'id': '39bb894794b741e982bd26144d2949f6', 'type': 'OS::Cinder::Volume', 'name': '123'} ], 'parameters': {"OS::Cinder::Volume": {"backup_name": "test"}}, } } controller = plan_api.PlansController() mock_provider.return_value = fakes.PROVIDER_OS req = fakes.HTTPRequest.blank('/v1/plans') plan = controller.create(req, body=create_plan_param) return plan['plan']