"Fossies" - the Fresh Open Source Software Archive

Member "cinder-14.0.2/cinder/tests/unit/backup/drivers/test_backup_posix.py" (4 Oct 2019, 6846 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 (c) 2015 Red Hat, Inc.
    2 # All Rights Reserved.
    3 #
    4 #    Licensed under the Apache License, Version 2.0 (the "License"); you may
    5 #    not use this file except in compliance with the License. You may obtain
    6 #    a copy of the License at
    7 #
    8 #         http://www.apache.org/licenses/LICENSE-2.0
    9 #
   10 #    Unless required by applicable law or agreed to in writing, software
   11 #    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
   12 #    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
   13 #    License for the specific language governing permissions and limitations
   14 #    under the License.
   15 """
   16 Tests for Posix backup driver.
   17 
   18 """
   19 
   20 import os
   21 
   22 import mock
   23 from six.moves import builtins
   24 
   25 from cinder.backup.drivers import posix
   26 from cinder import context
   27 from cinder import objects
   28 from cinder import test
   29 from cinder.tests.unit import fake_constants as fake
   30 
   31 
   32 FAKE_FILE_SIZE = 52428800
   33 FAKE_SHA_BLOCK_SIZE_BYTES = 1024
   34 FAKE_BACKUP_ENABLE_PROGRESS_TIMER = True
   35 
   36 FAKE_CONTAINER = 'fake/container'
   37 FAKE_BACKUP_ID = fake.BACKUP_ID
   38 FAKE_BACKUP_ID_PART1 = fake.BACKUP_ID[:2]
   39 FAKE_BACKUP_ID_PART2 = fake.BACKUP_ID[2:4]
   40 FAKE_BACKUP_ID_REST = fake.BACKUP_ID[4:]
   41 FAKE_BACKUP = {'id': FAKE_BACKUP_ID, 'container': None}
   42 
   43 UPDATED_CONTAINER_NAME = os.path.join(FAKE_BACKUP_ID_PART1,
   44                                       FAKE_BACKUP_ID_PART2,
   45                                       FAKE_BACKUP_ID)
   46 
   47 FAKE_BACKUP_MOUNT_POINT_BASE = '/fake/mount-point-base'
   48 FAKE_EXPORT_PATH = 'fake/export/path'
   49 
   50 FAKE_BACKUP_POSIX_PATH = os.path.join(FAKE_BACKUP_MOUNT_POINT_BASE,
   51                                       FAKE_EXPORT_PATH)
   52 
   53 FAKE_PREFIX = 'prefix-'
   54 FAKE_CONTAINER_ENTRIES = [FAKE_PREFIX + 'one', FAKE_PREFIX + 'two', 'three']
   55 EXPECTED_CONTAINER_ENTRIES = [FAKE_PREFIX + 'one', FAKE_PREFIX + 'two']
   56 FAKE_OBJECT_NAME = 'fake-object-name'
   57 FAKE_OBJECT_PATH = os.path.join(FAKE_BACKUP_POSIX_PATH, FAKE_CONTAINER,
   58                                 FAKE_OBJECT_NAME)
   59 
   60 
   61 class PosixBackupDriverTestCase(test.TestCase):
   62 
   63     def setUp(self):
   64         super(PosixBackupDriverTestCase, self).setUp()
   65         self.ctxt = context.get_admin_context()
   66 
   67         self.override_config('backup_file_size',
   68                              FAKE_FILE_SIZE)
   69         self.override_config('backup_sha_block_size_bytes',
   70                              FAKE_SHA_BLOCK_SIZE_BYTES)
   71         self.override_config('backup_enable_progress_timer',
   72                              FAKE_BACKUP_ENABLE_PROGRESS_TIMER)
   73         self.override_config('backup_posix_path',
   74                              FAKE_BACKUP_POSIX_PATH)
   75         self.mock_object(posix, 'LOG')
   76 
   77         self.driver = posix.PosixBackupDriver(self.ctxt)
   78 
   79     def test_init(self):
   80         drv = posix.PosixBackupDriver(self.ctxt)
   81         self.assertEqual(FAKE_BACKUP_POSIX_PATH,
   82                          drv.backup_path)
   83 
   84     def test_update_container_name_container_passed(self):
   85         result = self.driver.update_container_name(FAKE_BACKUP, FAKE_CONTAINER)
   86 
   87         self.assertEqual(FAKE_CONTAINER, result)
   88 
   89     def test_update_container_na_container_passed(self):
   90         result = self.driver.update_container_name(FAKE_BACKUP, None)
   91 
   92         self.assertEqual(UPDATED_CONTAINER_NAME, result)
   93 
   94     def test_put_container(self):
   95         self.mock_object(os.path, 'exists', return_value=False)
   96         self.mock_object(os, 'makedirs')
   97         self.mock_object(os, 'chmod')
   98         path = os.path.join(self.driver.backup_path, FAKE_CONTAINER)
   99 
  100         self.driver.put_container(FAKE_CONTAINER)
  101 
  102         os.path.exists.assert_called_once_with(path)
  103         os.makedirs.assert_called_once_with(path)
  104         os.chmod.assert_called_once_with(path, 0o770)
  105 
  106     def test_put_container_already_exists(self):
  107         self.mock_object(os.path, 'exists', return_value=True)
  108         self.mock_object(os, 'makedirs')
  109         self.mock_object(os, 'chmod')
  110         path = os.path.join(self.driver.backup_path, FAKE_CONTAINER)
  111 
  112         self.driver.put_container(FAKE_CONTAINER)
  113 
  114         os.path.exists.assert_called_once_with(path)
  115         self.assertEqual(0, os.makedirs.call_count)
  116         self.assertEqual(0, os.chmod.call_count)
  117 
  118     def test_put_container_exception(self):
  119         self.mock_object(os.path, 'exists', return_value=False)
  120         self.mock_object(os, 'makedirs', side_effect=OSError)
  121         self.mock_object(os, 'chmod')
  122         path = os.path.join(self.driver.backup_path, FAKE_CONTAINER)
  123 
  124         self.assertRaises(OSError, self.driver.put_container,
  125                           FAKE_CONTAINER)
  126         os.path.exists.assert_called_once_with(path)
  127         os.makedirs.assert_called_once_with(path)
  128         self.assertEqual(0, os.chmod.call_count)
  129 
  130     def test_get_container_entries(self):
  131         self.mock_object(os, 'listdir', return_value=FAKE_CONTAINER_ENTRIES)
  132 
  133         result = self.driver.get_container_entries(FAKE_CONTAINER, FAKE_PREFIX)
  134 
  135         self.assertEqual(EXPECTED_CONTAINER_ENTRIES, result)
  136 
  137     def test_get_container_entries_no_list(self):
  138         self.mock_object(os, 'listdir', return_value=[])
  139 
  140         result = self.driver.get_container_entries(FAKE_CONTAINER, FAKE_PREFIX)
  141 
  142         self.assertEqual([], result)
  143 
  144     def test_get_container_entries_no_match(self):
  145         self.mock_object(os, 'listdir', return_value=FAKE_CONTAINER_ENTRIES)
  146 
  147         result = self.driver.get_container_entries(FAKE_CONTAINER,
  148                                                    FAKE_PREFIX + 'garbage')
  149 
  150         self.assertEqual([], result)
  151 
  152     def test_get_object_writer(self):
  153         self.mock_object(builtins, 'open', mock.mock_open())
  154         self.mock_object(os, 'chmod')
  155 
  156         self.driver.get_object_writer(FAKE_CONTAINER, FAKE_OBJECT_NAME)
  157 
  158         os.chmod.assert_called_once_with(FAKE_OBJECT_PATH, 0o660)
  159         builtins.open.assert_called_once_with(FAKE_OBJECT_PATH, 'wb')
  160 
  161     def test_get_object_reader(self):
  162         self.mock_object(builtins, 'open', mock.mock_open())
  163 
  164         self.driver.get_object_reader(FAKE_CONTAINER, FAKE_OBJECT_NAME)
  165 
  166         builtins.open.assert_called_once_with(FAKE_OBJECT_PATH, 'rb')
  167 
  168     def test_delete_object(self):
  169         self.mock_object(os, 'remove')
  170 
  171         self.driver.delete_object(FAKE_CONTAINER, FAKE_OBJECT_NAME)
  172 
  173     @mock.patch.object(posix.timeutils, 'utcnow')
  174     def test_generate_object_name_prefix(self, utcnow_mock):
  175         timestamp = '20170518102205'
  176         utcnow_mock.return_value.strftime.return_value = timestamp
  177         backup = objects.Backup(self.ctxt, volume_id=fake.VOLUME_ID,
  178                                 id=fake.BACKUP_ID)
  179         res = self.driver._generate_object_name_prefix(backup)
  180         expected = 'volume_%s_%s_backup_%s' % (backup.volume_id,
  181                                                timestamp,
  182                                                backup.id)
  183         self.assertEqual(expected, res)