"Fossies" - the Fresh Open Source Software Archive

Member "cinder-14.0.2/cinder/tests/unit/api/contrib/test_snapshot_actions.py" (4 Oct 2019, 5761 Bytes) of package /linux/misc/openstack/cinder-14.0.2.tar.gz:


As a special service "Fossies" has tried to format the requested source page into HTML format using (guessed) Python source code syntax highlighting (style: standard) with prefixed line numbers. Alternatively you can here view or download the uninterpreted source code file.

    1 #   Copyright 2013, Red Hat, Inc.
    2 #
    3 #   Licensed under the Apache License, Version 2.0 (the "License"); you may
    4 #   not use this file except in compliance with the License. You may obtain
    5 #   a copy of the License at
    6 #
    7 #       http://www.apache.org/licenses/LICENSE-2.0
    8 #
    9 #   Unless required by applicable law or agreed to in writing, software
   10 #   distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
   11 #   WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
   12 #   License for the specific language governing permissions and limitations
   13 #   under the License.
   14 
   15 import ddt
   16 import mock
   17 from oslo_serialization import jsonutils
   18 from six.moves import http_client
   19 import webob
   20 
   21 from cinder.api.contrib import snapshot_actions
   22 from cinder.api import microversions as mv
   23 from cinder import context
   24 from cinder import db
   25 from cinder import exception
   26 from cinder.objects import fields
   27 from cinder import test
   28 from cinder.tests.unit.api import fakes
   29 from cinder.tests.unit.api.v2 import fakes as v2_fakes
   30 from cinder.tests.unit import fake_constants as fake
   31 
   32 
   33 def fake_snapshot_get(context, snapshot_id):
   34     snapshot = v2_fakes.fake_snapshot(snapshot_id)
   35 
   36     if snapshot_id == fake.SNAPSHOT_ID:
   37         snapshot['status'] = fields.SnapshotStatus.CREATING
   38     else:
   39         snapshot['status'] = fields.SnapshotStatus.ERROR
   40     return snapshot
   41 
   42 
   43 @ddt.ddt
   44 class SnapshotActionsTest(test.TestCase):
   45 
   46     def setUp(self):
   47         super(SnapshotActionsTest, self).setUp()
   48         self.user_ctxt = context.RequestContext(
   49             fake.USER_ID, fake.PROJECT_ID, auth_token=True)
   50         self.controller = snapshot_actions.SnapshotActionsController()
   51 
   52     @mock.patch('cinder.db.snapshot_update', autospec=True)
   53     @mock.patch('cinder.db.sqlalchemy.api._snapshot_get',
   54                 side_effect=fake_snapshot_get)
   55     @mock.patch('cinder.db.snapshot_metadata_get', return_value=dict())
   56     def test_update_snapshot_status(self, metadata_get, *args):
   57 
   58         body = {'os-update_snapshot_status':
   59                 {'status': fields.SnapshotStatus.AVAILABLE}}
   60         req = webob.Request.blank('/v2/%s/snapshots/%s/action' % (
   61             fake.PROJECT_ID, fake.SNAPSHOT_ID))
   62         req.method = "POST"
   63         req.body = jsonutils.dump_as_bytes(body)
   64         req.headers["content-type"] = "application/json"
   65 
   66         res = req.get_response(fakes.wsgi_app(
   67             fake_auth_context=self.user_ctxt))
   68         self.assertEqual(http_client.ACCEPTED, res.status_int)
   69 
   70     @mock.patch('cinder.db.sqlalchemy.api._snapshot_get',
   71                 side_effect=fake_snapshot_get)
   72     @mock.patch('cinder.db.snapshot_metadata_get', return_value=dict())
   73     def test_update_snapshot_status_invalid_status(self, metadata_get, *args):
   74         body = {'os-update_snapshot_status': {'status': 'in-use'}}
   75         req = webob.Request.blank('/v2/%s/snapshots/%s/action' % (
   76             fake.PROJECT_ID, fake.SNAPSHOT_ID))
   77         req.method = "POST"
   78         req.body = jsonutils.dump_as_bytes(body)
   79         req.headers["content-type"] = "application/json"
   80 
   81         res = req.get_response(fakes.wsgi_app(
   82             fake_auth_context=self.user_ctxt))
   83         self.assertEqual(http_client.BAD_REQUEST, res.status_int)
   84 
   85     def test_update_snapshot_status_without_status(self):
   86         self.mock_object(db, 'snapshot_get', fake_snapshot_get)
   87         body = {'os-update_snapshot_status': {}}
   88         req = webob.Request.blank('/v2/%s/snapshots/%s/action' % (
   89             fake.PROJECT_ID, fake.SNAPSHOT_ID))
   90         req.method = "POST"
   91         req.body = jsonutils.dump_as_bytes(body)
   92         req.headers["content-type"] = "application/json"
   93 
   94         res = req.get_response(fakes.wsgi_app(
   95             fake_auth_context=self.user_ctxt))
   96         self.assertEqual(http_client.BAD_REQUEST, res.status_int)
   97 
   98     @mock.patch('cinder.db.snapshot_update', autospec=True)
   99     @mock.patch('cinder.db.sqlalchemy.api._snapshot_get',
  100                 side_effect=fake_snapshot_get)
  101     @mock.patch('cinder.db.snapshot_metadata_get', return_value=dict())
  102     def test_update_snapshot_valid_progress(self, metadata_get, *args):
  103         body = {'os-update_snapshot_status':
  104                 {'status': fields.SnapshotStatus.AVAILABLE,
  105                  'progress': '50%'}}
  106         req = webob.Request.blank('/v2/%s/snapshots/%s/action' % (
  107             fake.PROJECT_ID, fake.SNAPSHOT_ID))
  108         req.method = "POST"
  109         req.body = jsonutils.dump_as_bytes(body)
  110         req.headers["content-type"] = "application/json"
  111 
  112         res = req.get_response(fakes.wsgi_app(
  113             fake_auth_context=self.user_ctxt))
  114         self.assertEqual(http_client.ACCEPTED, res.status_int)
  115 
  116     @ddt.data(({'os-update_snapshot_status':
  117                {'status': fields.SnapshotStatus.AVAILABLE,
  118                 'progress': '50'}}, exception.InvalidInput),
  119               ({'os-update_snapshot_status':
  120                {'status': fields.SnapshotStatus.AVAILABLE,
  121                 'progress': '103%'}}, exception.InvalidInput),
  122               ({'os-update_snapshot_status':
  123                {'status': fields.SnapshotStatus.AVAILABLE,
  124                 'progress': "   "}}, exception.InvalidInput),
  125               ({'os-update_snapshot_status':
  126                {'status': fields.SnapshotStatus.AVAILABLE,
  127                 'progress': 50}}, exception.ValidationError))
  128     @ddt.unpack
  129     def test_update_snapshot_invalid_progress(self, body, exception_class):
  130         req = webob.Request.blank('/v3/%s/snapshots/%s/action' % (
  131             fake.PROJECT_ID, fake.SNAPSHOT_ID))
  132         req.api_version_request = mv.get_api_version(mv.BASE_VERSION)
  133         self.assertRaises(exception_class,
  134                           self.controller._update_snapshot_status,
  135                           req, body=body)