← Back to branch summary

~ubuntu-branches/ubuntu/quantal/nova/quantal-security

« back to all changes in this revision

Viewing changes to nova/tests/test_xensm.py

  • Committer: Package Import Robot
  • Author(s): Chuck Short, Chuck Short, Adam Gandelman
  • Date: 2012-06-22 12:39:57 UTC
  • mfrom: (1.1.57)
  • Revision ID: package-import@ubuntu.com-20120622123957-hbzwg84nt9rqwg8r
Tags: 2012.2~f2~20120621.14517-0ubuntu1
[ Chuck Short ]
* New upstream version.

[ Adam Gandelman ]
* debian/rules: Temporarily disable test suite while blocking
  tests are investigated. 
* debian/patches/kombu_tests_timeout.patch: Dropped.

Show diffs side-by-side

added added

removed removed

Lines of Context:
nova/tests/test_xensm.py
 
1
# vim: tabstop=4 shiftwidth=4 softtabstop=4
 
2
 
 
3
#    Copyright (c) 2010 Citrix Systems, Inc.
 
4
#
 
5
#    Licensed under the Apache License, Version 2.0 (the "License"); you may
 
6
#    not use this file except in compliance with the License. You may obtain
 
7
#    a copy of the License at
 
8
#
 
9
#         http://www.apache.org/licenses/LICENSE-2.0
 
10
#
 
11
#    Unless required by applicable law or agreed to in writing, software
 
12
#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
 
13
#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
 
14
#    License for the specific language governing permissions and limitations
 
15
#    under the License.
 
16
 
 
17
"""Test suite for Xen Storage Manager Volume Driver."""
 
18
 
 
19
import os
 
20
 
 
21
from nova import context
 
22
from nova import db
 
23
from nova import exception
 
24
from nova import flags
 
25
from nova import log as logging
 
26
from nova import test
 
27
from nova.tests.xenapi import stubs
 
28
from nova.virt.xenapi import connection as xenapi_conn
 
29
from nova.virt.xenapi import fake as xenapi_fake
 
30
from nova.virt.xenapi import volume_utils
 
31
from nova.volume import xensm
 
32
 
 
33
LOG = logging.getLogger(__name__)
 
34
 
 
35
FLAGS = flags.FLAGS
 
36
 
 
37
 
 
38
class XenSMTestCase(test.TestCase):
 
39
    """Unit tests for Xen Storage Manager Volume operations."""
 
40
 
 
41
    def _get_sm_backend_params(self):
 
42
        config_params = ("name_label=testsmbackend "
 
43
                         "server=localhost "
 
44
                         "serverpath=/tmp/nfspath")
 
45
        params = dict(flavor_id=1,
 
46
                      sr_uuid=None,
 
47
                      sr_type='nfs',
 
48
                      config_params=config_params)
 
49
        return params
 
50
 
 
51
    def setUp(self):
 
52
        super(XenSMTestCase, self).setUp()
 
53
        self.user_id = 'fake'
 
54
        self.project_id = 'fake'
 
55
        self.context = context.RequestContext(self.user_id, self.project_id)
 
56
        self.flags(connection_type='xenapi',
 
57
                   xenapi_connection_url='http://test_url',
 
58
                   xenapi_connection_username='test_user',
 
59
                   xenapi_connection_password='test_pass')
 
60
        stubs.stubout_session(self.stubs, xenapi_fake.SessionBase)
 
61
        xenapi_fake.reset()
 
62
        self.driver = xensm.XenSMDriver()
 
63
        self.driver.db = db
 
64
 
 
65
    def _setup_step(self, ctxt):
 
66
        # Create a fake backend conf
 
67
        params = self._get_sm_backend_params()
 
68
        beconf = db.sm_backend_conf_create(ctxt,
 
69
                                           params)
 
70
        # Call setup, the one time operation that will create a backend SR
 
71
        self.driver.do_setup(ctxt)
 
72
        return beconf
 
73
 
 
74
    def test_do_setup(self):
 
75
        ctxt = context.get_admin_context()
 
76
        beconf = self._setup_step(ctxt)
 
77
        beconf = db.sm_backend_conf_get(ctxt, beconf['id'])
 
78
        self.assertIsInstance(beconf['sr_uuid'], basestring)
 
79
 
 
80
    def _create_volume(self, size=0):
 
81
        """Create a volume object."""
 
82
        vol = {}
 
83
        vol['size'] = size
 
84
        vol['user_id'] = 'fake'
 
85
        vol['project_id'] = 'fake'
 
86
        vol['host'] = 'localhost'
 
87
        vol['availability_zone'] = FLAGS.storage_availability_zone
 
88
        vol['status'] = "creating"
 
89
        vol['attach_status'] = "detached"
 
90
        return db.volume_create(self.context, vol)
 
91
 
 
92
    def test_create_volume(self):
 
93
        ctxt = context.get_admin_context()
 
94
        beconf = self._setup_step(ctxt)
 
95
        volume = self._create_volume()
 
96
        self.assertNotRaises(None, self.driver.create_volume, volume)
 
97
        self.assertNotRaises(None,
 
98
                             db.sm_volume_get,
 
99
                             ctxt,
 
100
                             volume['id'])
 
101
 
 
102
    def test_local_path(self):
 
103
        ctxt = context.get_admin_context()
 
104
        volume = self._create_volume()
 
105
        val = self.driver.local_path(volume)
 
106
        self.assertIsInstance(val, basestring)
 
107
 
 
108
    def test_delete_volume(self):
 
109
        ctxt = context.get_admin_context()
 
110
        beconf = self._setup_step(ctxt)
 
111
        volume = self._create_volume()
 
112
        self.driver.create_volume(volume)
 
113
        self.assertNotRaises(None, self.driver.delete_volume, volume)
 
114
        self.assertRaises(exception.NotFound,
 
115
                          db.sm_volume_get,
 
116
                          ctxt,
 
117
                          volume['id'])
 
118
 
 
119
    def test_delete_volume_raises_notfound(self):
 
120
        ctxt = context.get_admin_context()
 
121
        beconf = self._setup_step(ctxt)
 
122
        self.assertRaises(exception.NotFound,
 
123
                          self.driver.delete_volume,
 
124
                          {'id': "FA15E-1D"})
 
125
 
 
126
    def _get_expected_conn(self, beconf, vol):
 
127
        expected = {}
 
128
        expected['volume_id'] = unicode(vol['id'])
 
129
        expected['flavor_id'] = beconf['flavor_id']
 
130
        expected['sr_uuid'] = unicode(beconf['sr_uuid'])
 
131
        expected['sr_type'] = unicode(beconf['sr_type'])
 
132
        return expected
 
133
 
 
134
    def test_initialize_connection(self):
 
135
        ctxt = context.get_admin_context()
 
136
        beconf = self._setup_step(ctxt)
 
137
        beconf = db.sm_backend_conf_get(ctxt, beconf['id'])
 
138
        volume = self._create_volume()
 
139
        self.driver.create_volume(volume)
 
140
        expected = self._get_expected_conn(beconf, volume)
 
141
        conn = self.driver.initialize_connection(volume, 'fakeConnector')
 
142
        res = {}
 
143
        for key in ['volume_id', 'flavor_id', 'sr_uuid', 'sr_type']:
 
144
            res[key] = conn['data'][key]
 
145
        self.assertDictMatch(expected, res)
 
146
        self.assertEqual(set(conn['data']['introduce_sr_keys']),
 
147
                         set([u'sr_type', u'server', u'serverpath']))