← Back to branch summary

~ubuntu-branches/ubuntu/quantal/nova/quantal-proposed

« back to all changes in this revision

Viewing changes to nova/tests/test_virt.py

  • Committer: Package Import Robot
  • Author(s): Chuck Short
  • Date: 2012-08-16 14:04:11 UTC
  • mto: This revision was merged to the branch mainline in revision 84.
  • Revision ID: package-import@ubuntu.com-20120816140411-0mr4n241wmk30t9l
Tags: upstream-2012.2~f3
Import upstream version 2012.2~f3

Show diffs side-by-side

added added

removed removed

Lines of Context:
nova/tests/test_virt.py
15
15
#    License for the specific language governing permissions and limitations
16
16
#    under the License.
17
17
 
 
18
import os
 
19
 
18
20
from nova import exception
19
21
from nova import flags
20
22
from nova import test
 
23
from nova import utils
21
24
from nova.virt.disk import api as disk_api
22
25
from nova.virt import driver
23
26
 
 
27
from nova.openstack.common import jsonutils
 
28
 
24
29
FLAGS = flags.FLAGS
25
30
 
26
31
 
86
91
 
87
92
 
88
93
class TestVirtDisk(test.TestCase):
 
94
    def setUp(self):
 
95
        super(TestVirtDisk, self).setUp()
 
96
        self.executes = []
 
97
 
 
98
        def fake_execute(*cmd, **kwargs):
 
99
            self.executes.append(cmd)
 
100
            return None, None
 
101
 
 
102
        self.stubs.Set(utils, 'execute', fake_execute)
 
103
 
 
104
    def test_lxc_destroy_container(self):
 
105
 
 
106
        def proc_mounts(self, mount_point):
 
107
            mount_points = {
 
108
                '/mnt/loop/nopart': '/dev/loop0',
 
109
                '/mnt/loop/part': '/dev/mapper/loop0p1',
 
110
                '/mnt/nbd/nopart': '/dev/nbd15',
 
111
                '/mnt/nbd/part': '/dev/mapper/nbd15p1',
 
112
                '/mnt/guestfs': 'guestmount',
 
113
            }
 
114
            return mount_points[mount_point]
 
115
 
 
116
        self.stubs.Set(os.path, 'exists', lambda _: True)
 
117
        self.stubs.Set(disk_api._DiskImage, '_device_for_path', proc_mounts)
 
118
        expected_commands = []
 
119
 
 
120
        disk_api.destroy_container('/mnt/loop/nopart')
 
121
        expected_commands += [
 
122
                              ('umount', '/dev/loop0'),
 
123
                              ('losetup', '--detach', '/dev/loop0'),
 
124
                             ]
 
125
 
 
126
        disk_api.destroy_container('/mnt/loop/part')
 
127
        expected_commands += [
 
128
                              ('umount', '/dev/mapper/loop0p1'),
 
129
                              ('kpartx', '-d', '/dev/loop0'),
 
130
                              ('losetup', '--detach', '/dev/loop0'),
 
131
                             ]
 
132
 
 
133
        disk_api.destroy_container('/mnt/nbd/nopart')
 
134
        expected_commands += [
 
135
                              ('umount', '/dev/nbd15'),
 
136
                              ('qemu-nbd', '-d', '/dev/nbd15'),
 
137
                             ]
 
138
 
 
139
        disk_api.destroy_container('/mnt/nbd/part')
 
140
        expected_commands += [
 
141
                              ('umount', '/dev/mapper/nbd15p1'),
 
142
                              ('kpartx', '-d', '/dev/nbd15'),
 
143
                              ('qemu-nbd', '-d', '/dev/nbd15'),
 
144
                             ]
 
145
 
 
146
        disk_api.destroy_container('/mnt/guestfs')
 
147
        expected_commands += [
 
148
                              ('fusermount', '-u', '/mnt/guestfs'),
 
149
                             ]
 
150
        # It's not worth trying to match the last timeout command
 
151
        self.executes.pop()
 
152
 
 
153
        self.assertEqual(self.executes, expected_commands)
 
154
 
 
155
 
 
156
class TestVirtDiskPaths(test.TestCase):
 
157
    def setUp(self):
 
158
        super(TestVirtDiskPaths, self).setUp()
 
159
 
 
160
        real_execute = utils.execute
 
161
 
 
162
        def nonroot_execute(*cmd_parts, **kwargs):
 
163
            kwargs.pop('run_as_root', None)
 
164
            return real_execute(*cmd_parts, **kwargs)
 
165
 
 
166
        self.stubs.Set(utils, 'execute', nonroot_execute)
 
167
 
89
168
    def test_check_safe_path(self):
90
169
        ret = disk_api._join_and_check_path_within_fs('/foo', 'etc',
91
170
                                                      'something.conf')
101
180
                          disk_api._inject_file_into_fs,
102
181
                          '/tmp', '/etc/../../../../etc/passwd',
103
182
                          'hax')
 
183
 
 
184
    def test_inject_metadata(self):
 
185
        with utils.tempdir() as tmpdir:
 
186
            meta_objs = [{"key": "foo", "value": "bar"}]
 
187
            metadata = {"foo": "bar"}
 
188
            disk_api._inject_metadata_into_fs(meta_objs, tmpdir)
 
189
            json_file = os.path.join(tmpdir, 'meta.js')
 
190
            json_data = jsonutils.loads(open(json_file).read())
 
191
            self.assertEqual(metadata, json_data)