← Back to branch summary

~thedac/charm-helpers/volume-fixes

« back to all changes in this revision

Viewing changes to tests/contrib/storage/test_linux_storage_loopback.py

  • Committer: James Page
  • Date: 2013-06-25 07:42:19 UTC
  • mfrom: (34.1.2 charm-helpers)
  • Revision ID: james.page@canonical.com-20130625074219-bwl8dr3msmhkcjki
[gandelman-a]

This adds some small linux-specific utility helpers useful when working with block devices and storage. They are mostly derived from the Openstack helpers but renamed and tests added.

Show diffs side-by-side

added added

removed removed

Lines of Context:
tests/contrib/storage/test_linux_storage_loopback.py
 
1
import unittest
 
2
 
 
3
from mock import patch
 
4
 
 
5
import charmhelpers.contrib.storage.linux.loopback as loopback
 
6
 
 
7
LOOPBACK_DEVICES = """
 
8
/dev/loop0: [0805]:2244465 (/tmp/foo.img)
 
9
/dev/loop1: [0805]:2244466 (/tmp/bar.img)
 
10
/dev/loop2: [0805]:2244467 (/tmp/baz.img)
 
11
"""
 
12
 
 
13
# It's a mouthful.
 
14
STORAGE_LINUX_LOOPBACK = 'charmhelpers.contrib.storage.linux.loopback'
 
15
 
 
16
 
 
17
class LoopbackStorageUtilsTests(unittest.TestCase):
 
18
    @patch(STORAGE_LINUX_LOOPBACK + '.check_output')
 
19
    def test_loopback_devices(self, output):
 
20
        '''It translates current loopback mapping to a dict'''
 
21
        output.return_value = LOOPBACK_DEVICES
 
22
        ex = {
 
23
            '/dev/loop1': '/tmp/bar.img',
 
24
            '/dev/loop0': '/tmp/foo.img',
 
25
            '/dev/loop2': '/tmp/baz.img'
 
26
        }
 
27
        self.assertEquals(loopback.loopback_devices(), ex)
 
28
 
 
29
    @patch(STORAGE_LINUX_LOOPBACK + '.create_loopback')
 
30
    @patch('subprocess.check_call')
 
31
    @patch(STORAGE_LINUX_LOOPBACK + '.loopback_devices')
 
32
    def test_loopback_create_already_exists(self, loopbacks, check_call,
 
33
                                            create):
 
34
        '''It finds existing loopback device for requested file'''
 
35
        loopbacks.return_value = {'/dev/loop1': '/tmp/bar.img'}
 
36
        res = loopback.ensure_loopback_device('/tmp/bar.img', '5G')
 
37
        self.assertEquals(res, '/dev/loop1')
 
38
        self.assertFalse(create.called)
 
39
        self.assertFalse(check_call.called)
 
40
 
 
41
    @patch(STORAGE_LINUX_LOOPBACK + '.loopback_devices')
 
42
    @patch(STORAGE_LINUX_LOOPBACK + '.create_loopback')
 
43
    @patch('os.path.exists')
 
44
    def test_loop_creation_no_truncate(self, path_exists, create_loopback,
 
45
                                       loopbacks):
 
46
        '''It does not create a new sparse image for loopback if one exists'''
 
47
        loopbacks.return_value = {}
 
48
        path_exists.return_value = True
 
49
        with patch('subprocess.check_call') as check_call:
 
50
            loopback.ensure_loopback_device('/tmp/foo.img', '15G')
 
51
            self.assertFalse(check_call.called)
 
52
 
 
53
    @patch(STORAGE_LINUX_LOOPBACK + '.loopback_devices')
 
54
    @patch(STORAGE_LINUX_LOOPBACK + '.create_loopback')
 
55
    @patch('os.path.exists')
 
56
    def test_ensure_loopback_creation(self, path_exists, create_loopback,
 
57
                                      loopbacks):
 
58
        '''It creates a new sparse image for loopback if one does not exists'''
 
59
        loopbacks.return_value = {}
 
60
        path_exists.return_value = False
 
61
        create_loopback.return_value = '/dev/loop0'
 
62
        with patch(STORAGE_LINUX_LOOPBACK + '.check_call') as check_call:
 
63
            loopback.ensure_loopback_device('/tmp/foo.img', '15G')
 
64
            check_call.assert_called_with(['truncate', '--size', '15G',
 
65
                                           '/tmp/foo.img'])
 
66
 
 
67
    def test_create_loopback(self):
 
68
        '''It corectly calls losetup to create a loopback device'''
 
69
        with patch(STORAGE_LINUX_LOOPBACK + '.check_output') as check_output:
 
70
            check_output.return_value = '/dev/loop0'
 
71
            result = loopback.create_loopback('/tmp/foo')
 
72
            check_output.assert_called_with(['losetup', '--find', '/tmp/foo'])
 
73
            self.assertEquals(result, '/dev/loop0')