# License for the specific language governing permissions and limitations
# under the License.
+import contextlib
+import os
+import tempfile
+
from cinder import db
from cinder import exception
from cinder.openstack.common import log as logging
LOG = logging.getLogger(__name__)
+class FakeImageService:
+ def download(self, context, image_id, path):
+ pass
+
+
class RBDTestCase(test.TestCase):
def setUp(self):
location = 'rbd://abc/pool/image/snap'
self.assertFalse(self.driver._is_cloneable(location))
+ def _copy_image(self):
+ @contextlib.contextmanager
+ def fake_temp_file(dir):
+ class FakeTmp:
+ def __init__(self, name):
+ self.name = name
+ yield FakeTmp('test')
+ self.stubs.Set(tempfile, 'NamedTemporaryFile', fake_temp_file)
+ self.stubs.Set(os.path, 'exists', lambda x: True)
+ self.driver.copy_image_to_volume(None, {'name': 'test'},
+ FakeImageService(), None)
+
+ def test_copy_image_no_volume_tmp(self):
+ self.flags(volume_tmp_dir=None)
+ self._copy_image()
+
+ def test_copy_image_volume_tmp(self):
+ self.flags(volume_tmp_dir='/var/run/cinder/tmp')
+ self._copy_image()
+
class FakeRBDDriver(RBDDriver):
# TODO(jdurgin): replace with librbd
# this is a temporary hack, since rewriting this driver
# to use librbd would take too long
- if FLAGS.volume_tmp_dir and not os.exists(FLAGS.volume_tmp_dir):
+ if FLAGS.volume_tmp_dir and not os.path.exists(FLAGS.volume_tmp_dir):
os.makedirs(FLAGS.volume_tmp_dir)
with tempfile.NamedTemporaryFile(dir=FLAGS.volume_tmp_dir) as tmp: