"Fossies" - the Fresh Open Source Software Archive

Member "cinder-13.0.7/cinder/tests/unit/volume/drivers/disco/test_create_snapshot.py" (4 Oct 2019, 5947 Bytes) of package /linux/misc/openstack/cinder-13.0.7.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 # (c) Copyright 2015 Industrial Technology Research Institute.
    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 """Test case for the function create snapshot."""
   16 
   17 
   18 import copy
   19 import mock
   20 import time
   21 
   22 from cinder import db
   23 from cinder import exception
   24 from cinder.tests.unit import fake_snapshot
   25 from cinder.tests.unit import utils
   26 from cinder.tests.unit.volume.drivers import disco
   27 
   28 
   29 class CreateSnapshotTestCase(disco.TestDISCODriver):
   30     """Test cases for DISCO connector."""
   31 
   32     def get_fake_volume(self, ctx, id):
   33         """Return fake volume from db calls."""
   34         return self.volume
   35 
   36     def setUp(self):
   37         """Initialise variables and mock functions."""
   38         super(CreateSnapshotTestCase, self).setUp()
   39 
   40         self.snapshot = fake_snapshot.fake_snapshot_obj(
   41             self.ctx, **{'volume': self.volume})
   42 
   43         # Mock db call in the cinder driver
   44         self.mock_object(db.sqlalchemy.api, 'volume_get',
   45                          self.get_fake_volume)
   46 
   47         mock.patch.object(self.requester,
   48                           'snapshotCreate',
   49                           self.snapshot_request).start()
   50 
   51         mock.patch.object(self.requester,
   52                           'snapshotDetail',
   53                           self.snapshot_detail_request).start()
   54 
   55         snapshot_detail_response = {
   56             'status': 0,
   57             'snapshotInfoResult':
   58                 {'snapshotId': 1234,
   59                  'description': 'a description',
   60                  'createTime': '',
   61                  'expireTime': '',
   62                  'isDeleted': False,
   63                  'status': 0}
   64         }
   65 
   66         snap_success = copy.deepcopy(snapshot_detail_response)
   67         snap_pending = copy.deepcopy(snapshot_detail_response)
   68         snap_fail = copy.deepcopy(snapshot_detail_response)
   69         snap_response_fail = copy.deepcopy(snapshot_detail_response)
   70         snap_success['snapshotInfoResult']['status'] = (
   71             self.DETAIL_OPTIONS['success'])
   72         snap_pending['snapshotInfoResult']['status'] = (
   73             self.DETAIL_OPTIONS['pending'])
   74         snap_fail['snapshotInfoResult']['status'] = (
   75             self.DETAIL_OPTIONS['failure'])
   76         snap_response_fail['status'] = 1
   77 
   78         self.FAKE_RESPONSE['snapshot_detail'] = {
   79             'success': snap_success,
   80             'fail': snap_fail,
   81             'pending': snap_pending,
   82             'request_fail': snap_response_fail}
   83 
   84         self.response = (
   85             self.FAKE_RESPONSE['standard']['success'])
   86         self.response['result'] = 1234
   87 
   88         self.response_detail = (
   89             self.FAKE_RESPONSE['snapshot_detail']['success'])
   90         self.test_pending = False
   91 
   92         self.test_pending_count = 0
   93 
   94     def snapshot_request(self, *cmd, **kwargs):
   95         """Mock function for the createSnapshot call."""
   96         return self.response
   97 
   98     def snapshot_detail_request(self, *cmd, **kwargs):
   99         """Mock function for the snapshotDetail call."""
  100         if self.test_pending:
  101             if self.test_pending_count == 0:
  102                 self.test_pending_count += 1
  103                 return self.FAKE_RESPONSE['snapshot_detail']['pending']
  104             else:
  105                 return self.FAKE_RESPONSE['snapshot_detail']['success']
  106         else:
  107             return self.response_detail
  108 
  109     def test_create_snapshot(self):
  110         """Normal test case."""
  111         expected = 1234
  112         actual = self.driver.create_snapshot(self.volume)
  113         self.assertEqual(expected, actual['provider_location'])
  114 
  115     def test_create_snapshot_fail(self):
  116         """Request to DISCO failed."""
  117         self.response = self.FAKE_RESPONSE['standard']['fail']
  118         self.assertRaises(exception.VolumeBackendAPIException,
  119                           self.test_create_snapshot)
  120 
  121     def test_create_snapshot_fail_not_immediate(self):
  122         """Request to DISCO failed when monitoring the snapshot details."""
  123         self.response = self.FAKE_RESPONSE['standard']['success']
  124         self.response_detail = (
  125             self.FAKE_RESPONSE['snapshot_detail']['fail'])
  126         self.assertRaises(exception.VolumeBackendAPIException,
  127                           self.test_create_snapshot)
  128 
  129     def test_create_snapshot_fail_not_immediate_response_fail(self):
  130         """Request to get the snapshot details returns a failure."""
  131         self.response = self.FAKE_RESPONSE['standard']['success']
  132         self.response_detail = (
  133             self.FAKE_RESPONSE['snapshot_detail']['request_fail'])
  134         self.assertRaises(exception.VolumeBackendAPIException,
  135                           self.test_create_snapshot)
  136 
  137     def test_create_snapshot_detail_pending(self):
  138         """Request to get the snapshot detail return pending then success."""
  139         self.response = self.FAKE_RESPONSE['standard']['success']
  140         self.test_pending = True
  141         self.test_create_snapshot()
  142 
  143     @mock.patch.object(time, 'time')
  144     def test_create_snapshot_timeout(self, mock_time):
  145         """Snapshot request timeout."""
  146         timeout = 3
  147         mock_time.side_effect = utils.generate_timeout_series(timeout)
  148         self.driver.configuration.disco_snapshot_check_timeout = timeout
  149         self.response = self.FAKE_RESPONSE['standard']['success']
  150         self.response_detail = (
  151             self.FAKE_RESPONSE['snapshot_detail']['pending'])
  152         self.assertRaises(exception.VolumeBackendAPIException,
  153                           self.test_create_snapshot)