text: # Copyright 2015 Hewlett-Packard Development Company, L.P.
  #
  #    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_service import fixture as service_fixture
  
  from nova import test
  from nova.virt.disk.mount import api
  from nova.virt.disk.mount import block
  from nova.virt.disk.mount import loop
  from nova.virt.disk.mount import nbd
  from nova.virt.image import model as imgmodel
  
  
  PARTITION = 77
  ORIG_DEVICE = "/dev/null"
  AUTOMAP_PARTITION = "/dev/nullp77"
  MAP_PARTITION = "/dev/mapper/nullp77"
  
  
  class MountTestCase(test.NoDBTestCase):
      def setUp(self):
          super(MountTestCase, self).setUp()
          # Make RetryDecorator not actually sleep on retries
          self.useFixture(service_fixture.SleepFixture())
  
      def _test_map_dev(self, partition):
          mount = api.Mount(mock.sentinel.image, mock.sentinel.mount_dir)
          mount.device = ORIG_DEVICE
          mount.partition = partition
          mount.map_dev()
          return mount
  
      def _exists_effect(self, data):
          def exists_effect(filename):
              try:
                  v = data[filename]
                  if isinstance(v, list):
                      if len(v) > 0:
                          return v.pop(0)
                      self.fail("Out of items for: %s" % filename)
                  return v
              except KeyError:
                  self.fail("Unexpected call with: %s" % filename)
          return exists_effect
  
      def _check_calls(self, exists, filenames, trailing=0):
          self.assertEqual([mock.call(x) for x in filenames],
                  exists.call_args_list[:len(filenames)])
          self.assertEqual([mock.call(MAP_PARTITION)] * trailing,
              exists.call_args_list[len(filenames):])
  
      @mock.patch('os.path.exists')
      def test_map_dev_partition_search(self, exists):
          exists.side_effect = self._exists_effect({
              ORIG_DEVICE: True})
          mount = self._test_map_dev(-1)
          self._check_calls(exists, [ORIG_DEVICE])
          self.assertNotEqual("", mount.error)
          self.assertFalse(mount.mapped)
  
      @mock.patch('os.path.exists')
      @mock.patch('nova.privsep.fs.create_device_maps',
                  return_value=(None, None))
      def test_map_dev_good(self, mock_create_maps, mock_exists):
          mock_exists.side_effect = self._exists_effect({
              ORIG_DEVICE: True,
              AUTOMAP_PARTITION: False,
              MAP_PARTITION: [False, True]})
          mount = self._test_map_dev(PARTITION)
          self._check_calls(mock_exists, [ORIG_DEVICE, AUTOMAP_PARTITION], 2)
          self.assertEqual("", mount.error)
          self.assertTrue(mount.mapped)
  
      @mock.patch('os.path.exists')
      @mock.patch('nova.privsep.fs.create_device_maps',
                  return_value=(None, None))
      def test_map_dev_error(self, mock_create_maps, mock_exists):
          mock_exists.side_effect = self._exists_effect({
              ORIG_DEVICE: True,
              AUTOMAP_PARTITION: False,
              MAP_PARTITION: False})
          mount = self._test_map_dev(PARTITION)
          self._check_calls(mock_exists, [ORIG_DEVICE, AUTOMAP_PARTITION],
                  api.MAX_FILE_CHECKS + 1)
          self.assertNotEqual("", mount.error)
          self.assertFalse(mount.mapped)
  
      @mock.patch('os.path.exists')
      @mock.patch('nova.privsep.fs.create_device_maps',
                  return_value=(None, None))
      def test_map_dev_error_then_pass(self, mock_create_maps, mock_exists):
          mock_exists.side_effect = self._exists_effect({
              ORIG_DEVICE: True,
              AUTOMAP_PARTITION: False,
              MAP_PARTITION: [False, False, True]})
          mount = self._test_map_dev(PARTITION)
          self._check_calls(mock_exists, [ORIG_DEVICE, AUTOMAP_PARTITION], 3)
          self.assertEqual("", mount.error)
          self.assertTrue(mount.mapped)
  
      @mock.patch('os.path.exists')
      def test_map_dev_automap(self, exists):
          exists.side_effect = self._exists_effect({
              ORIG_DEVICE: True,
              AUTOMAP_PARTITION: True})
          mount = self._test_map_dev(PARTITION)
          self._check_calls(exists,
              [ORIG_DEVICE, AUTOMAP_PARTITION, AUTOMAP_PARTITION])
          self.assertEqual(AUTOMAP_PARTITION, mount.mapped_device)
          self.assertTrue(mount.automapped)
          self.assertTrue(mount.mapped)
  
      @mock.patch('os.path.exists')
      def test_map_dev_else(self, exists):
          exists.side_effect = self._exists_effect({
              ORIG_DEVICE: True,
              AUTOMAP_PARTITION: True})
          mount = self._test_map_dev(None)
          self._check_calls(exists, [ORIG_DEVICE])
          self.assertEqual(ORIG_DEVICE, mount.mapped_device)
          self.assertFalse(mount.automapped)
          self.assertTrue(mount.mapped)
  
      def test_instance_for_format_raw(self):
          image = imgmodel.LocalFileImage("/some/file.raw",
                                          imgmodel.FORMAT_RAW)
          mount_dir = '/mount/dir'
          partition = -1
          inst = api.Mount.instance_for_format(image, mount_dir, partition)
          self.assertIsInstance(inst, loop.LoopMount)
  
      def test_instance_for_format_qcow2(self):
          image = imgmodel.LocalFileImage("/some/file.qcows",
                                          imgmodel.FORMAT_QCOW2)
          mount_dir = '/mount/dir'
          partition = -1
          inst = api.Mount.instance_for_format(image, mount_dir, partition)
          self.assertIsInstance(inst, nbd.NbdMount)
  
      def test_instance_for_format_block(self):
          image = imgmodel.LocalBlockImage(
              "/dev/mapper/instances--instance-0000001_disk",)
          mount_dir = '/mount/dir'
          partition = -1
          inst = api.Mount.instance_for_format(image, mount_dir, partition)
          self.assertIsInstance(inst, block.BlockMount)
  
      def test_instance_for_device_loop(self):
          image = mock.MagicMock()
          mount_dir = '/mount/dir'
          partition = -1
          device = '/dev/loop0'
          inst = api.Mount.instance_for_device(image, mount_dir, partition,
                                                 device)
          self.assertIsInstance(inst, loop.LoopMount)
  
      def test_instance_for_device_loop_partition(self):
          image = mock.MagicMock()
          mount_dir = '/mount/dir'
          partition = 1
          device = '/dev/mapper/loop0p1'
          inst = api.Mount.instance_for_device(image, mount_dir, partition,
                                                 device)
          self.assertIsInstance(inst, loop.LoopMount)
  
      def test_instance_for_device_nbd(self):
          image = mock.MagicMock()
          mount_dir = '/mount/dir'
          partition = -1
          device = '/dev/nbd0'
          inst = api.Mount.instance_for_device(image, mount_dir, partition,
                                                 device)
          self.assertIsInstance(inst, nbd.NbdMount)
  
      def test_instance_for_device_nbd_partition(self):
          image = mock.MagicMock()
          mount_dir = '/mount/dir'
          partition = 1
          device = '/dev/mapper/nbd0p1'
          inst = api.Mount.instance_for_device(image, mount_dir, partition,
                                                 device)
          self.assertIsInstance(inst, nbd.NbdMount)
  
      def test_instance_for_device_block(self):
          image = mock.MagicMock()
          mount_dir = '/mount/dir'
          partition = -1
          device = '/dev/mapper/instances--instance-0000001_disk'
          inst = api.Mount.instance_for_device(image, mount_dir, partition,
                                                 device)
          self.assertIsInstance(inst, block.BlockMount)
  
      def test_instance_for_device_block_partiton(self,):
          image = mock.MagicMock()
          mount_dir = '/mount/dir'
          partition = 1
          device = '/dev/mapper/instances--instance-0000001_diskp1'
          inst = api.Mount.instance_for_device(image, mount_dir, partition,
                                                 device)
          self.assertIsInstance(inst, block.BlockMount)
  