~ubuntu-branches/ubuntu/raring/cinder/raring-updates

« back to all changes in this revision

Viewing changes to cinder/tests/api/contrib/test_admin_actions.py

• browse files at revision 1.1.10
• compare with another revision
• compare with revision 1.1.10
• download diff
• download tarball
• view history from revision 1.1.10

Show diffs side-by-side

added

removed

cinder/tests/api/contrib/test_admin_actions.py

 

1

import shutil

 

2

import tempfile

1

3

import webob

2

4

 

3

5

from cinder import context

5

7

from cinder import exception

6

8

from cinder.openstack.common import jsonutils

7

9

from cinder import test

8

 

from cinder.tests.api.openstack import fakes

 

10

from cinder.tests.api import fakes

 

11

from cinder.tests.api.v2 import stubs

9

12

from cinder.volume import api as volume_api

10

13

 

11

14

 

13

16

    # no auth, just let environ['cinder.context'] pass through

14

17

    api = fakes.router.APIRouter()

15

18

    mapper = fakes.urlmap.URLMap()

16

 

    mapper['/v1'] = api

 

19

    mapper['/v2'] = api

17

20

    return mapper

18

21

 

19

22

 

20

23

class AdminActionsTest(test.TestCase):

21

24

 

22

25

    def setUp(self):

 

26

        self.tempdir = tempfile.mkdtemp()

23

27

        super(AdminActionsTest, self).setUp()

24

28

        self.flags(rpc_backend='cinder.openstack.common.rpc.impl_fake')

 

29

        self.flags(lock_path=self.tempdir)

25

30

        self.volume_api = volume_api.API()

26

31

 

 

32

    def tearDown(self):

 

33

        shutil.rmtree(self.tempdir)

 

34

 

27

35

    def test_reset_status_as_admin(self):

28

36

        # admin context

29

37

        ctx = context.RequestContext('admin', 'fake', True)

30

38

        # current status is available

31

39

        volume = db.volume_create(ctx, {'status': 'available'})

32

 

        req = webob.Request.blank('/v1/fake/volumes/%s/action' % volume['id'])

 

40

        req = webob.Request.blank('/v2/fake/volumes/%s/action' % volume['id'])

33

41

        req.method = 'POST'

34

42

        req.headers['content-type'] = 'application/json'

35

43

        # request status of 'error'

47

55

        # current status is 'error'

48

56

        volume = db.volume_create(context.get_admin_context(),

49

57

                                  {'status': 'error'})

50

 

        req = webob.Request.blank('/v1/fake/volumes/%s/action' % volume['id'])

 

58

        req = webob.Request.blank('/v2/fake/volumes/%s/action' % volume['id'])

51

59

        req.method = 'POST'

52

60

        req.headers['content-type'] = 'application/json'

53

61

        # request changing status to available

67

75

        ctx = context.RequestContext('admin', 'fake', True)

68

76

        # current status is available

69

77

        volume = db.volume_create(ctx, {'status': 'available'})

70

 

        req = webob.Request.blank('/v1/fake/volumes/%s/action' % volume['id'])

 

78

        req = webob.Request.blank('/v2/fake/volumes/%s/action' % volume['id'])

71

79

        req.method = 'POST'

72

80

        req.headers['content-type'] = 'application/json'

73

81

        # malformed request body

86

94

        ctx = context.RequestContext('admin', 'fake', True)

87

95

        # current status is available

88

96

        volume = db.volume_create(ctx, {'status': 'available'})

89

 

        req = webob.Request.blank('/v1/fake/volumes/%s/action' % volume['id'])

 

97

        req = webob.Request.blank('/v2/fake/volumes/%s/action' % volume['id'])

90

98

        req.method = 'POST'

91

99

        req.headers['content-type'] = 'application/json'

92

100

        # 'invalid' is not a valid status

104

112

        # admin context

105

113

        ctx = context.RequestContext('admin', 'fake', True)

106

114

        # missing-volume-id

107

 

        req = webob.Request.blank('/v1/fake/volumes/%s/action' %

 

115

        req = webob.Request.blank('/v2/fake/volumes/%s/action' %

108

116

                                  'missing-volume-id')

109

117

        req.method = 'POST'

110

118

        req.headers['content-type'] = 'application/json'

125

133

        # current status is available

126

134

        volume = db.volume_create(ctx, {'status': 'available',

127

135

                                        'attach_status': 'attached'})

128

 

        req = webob.Request.blank('/v1/fake/volumes/%s/action' % volume['id'])

 

136

        req = webob.Request.blank('/v2/fake/volumes/%s/action' % volume['id'])

129

137

        req.method = 'POST'

130

138

        req.headers['content-type'] = 'application/json'

131

139

        # request update attach_status to detached

149

157

        # current status is available

150

158

        volume = db.volume_create(ctx, {'status': 'available',

151

159

                                        'attach_status': 'detached'})

152

 

        req = webob.Request.blank('/v1/fake/volumes/%s/action' % volume['id'])

 

160

        req = webob.Request.blank('/v2/fake/volumes/%s/action' % volume['id'])

153

161

        req.method = 'POST'

154

162

        req.headers['content-type'] = 'application/json'

155

163

        # 'invalid' is not a valid attach_status

173

181

        volume = db.volume_create(ctx, {})

174

182

        snapshot = db.snapshot_create(ctx, {'status': 'error_deleting',

175

183

                                            'volume_id': volume['id']})

176

 

        req = webob.Request.blank('/v1/fake/snapshots/%s/action' %

 

184

        req = webob.Request.blank('/v2/fake/snapshots/%s/action' %

177

185

                                  snapshot['id'])

178

186

        req.method = 'POST'

179

187

        req.headers['content-type'] = 'application/json'

195

203

        volume = db.volume_create(ctx, {})

196

204

        snapshot = db.snapshot_create(ctx, {'status': 'available',

197

205

                                            'volume_id': volume['id']})

198

 

        req = webob.Request.blank('/v1/fake/snapshots/%s/action' %

 

206

        req = webob.Request.blank('/v2/fake/snapshots/%s/action' %

199

207

                                  snapshot['id'])

200

208

        req.method = 'POST'

201

209

        req.headers['content-type'] = 'application/json'

216

224

        ctx = context.RequestContext('admin', 'fake', True)

217

225

        # current status is creating

218

226

        volume = db.volume_create(ctx, {'status': 'creating'})

219

 

        req = webob.Request.blank('/v1/fake/volumes/%s/action' % volume['id'])

 

227

        req = webob.Request.blank('/v2/fake/volumes/%s/action' % volume['id'])

220

228

        req.method = 'POST'

221

229

        req.headers['content-type'] = 'application/json'

222

230

        req.body = jsonutils.dumps({'os-force_delete': {}})

236

244

        snapshot = db.snapshot_create(ctx, {'status': 'creating',

237

245

                                            'volume_size': 1,

238

246

                                            'volume_id': volume['id']})

239

 

        path = '/v1/fake/snapshots/%s/action' % snapshot['id']

 

247

        path = '/v2/fake/snapshots/%s/action' % snapshot['id']

240

248

        req = webob.Request.blank(path)

241

249

        req.method = 'POST'

242

250

        req.headers['content-type'] = 'application/json'

264

272

        self.volume_api.reserve_volume(ctx, volume)

265

273

        self.volume_api.initialize_connection(ctx, volume, {})

266

274

        mountpoint = '/dev/vbd'

267

 

        self.volume_api.attach(ctx, volume, fakes.FAKE_UUID, mountpoint)

 

275

        self.volume_api.attach(ctx, volume, stubs.FAKE_UUID, mountpoint)

268

276

        # volume is attached

269

277

        volume = db.volume_get(ctx, volume['id'])

270

278

        self.assertEquals(volume['status'], 'in-use')

271

 

        self.assertEquals(volume['instance_uuid'], fakes.FAKE_UUID)

 

279

        self.assertEquals(volume['instance_uuid'], stubs.FAKE_UUID)

272

280

        self.assertEquals(volume['mountpoint'], mountpoint)

273

281

        self.assertEquals(volume['attach_status'], 'attached')

274

282

        # build request to force detach

275

 

        req = webob.Request.blank('/v1/fake/volumes/%s/action' % volume['id'])

 

283

        req = webob.Request.blank('/v2/fake/volumes/%s/action' % volume['id'])

276

284

        req.method = 'POST'

277

285

        req.headers['content-type'] = 'application/json'

278

286

        # request status of 'error'

289

297

        self.assertEquals(volume['instance_uuid'], None)

290

298

        self.assertEquals(volume['mountpoint'], None)

291

299

        self.assertEquals(volume['attach_status'], 'detached')

 

300

 

 

301

    def test_attach_in_use_volume(self):

 

302

        """Test that attaching to an in-use volume fails."""

 

303

        # admin context

 

304

        ctx = context.RequestContext('admin', 'fake', True)

 

305

        # current status is available

 

306

        volume = db.volume_create(ctx, {'status': 'available', 'host': 'test',

 

307

                                        'provider_location': ''})

 

308

        # start service to handle rpc messages for attach requests

 

309

        self.start_service('volume', host='test')

 

310

        self.volume_api.reserve_volume(ctx, volume)

 

311

        self.volume_api.initialize_connection(ctx, volume, {})

 

312

        mountpoint = '/dev/vbd'

 

313

        self.volume_api.attach(ctx, volume, stubs.FAKE_UUID, mountpoint)

 

314

        self.assertRaises(exception.InvalidVolume,

 

315

                          self.volume_api.attach,

 

316

                          ctx,

 

317

                          volume,

 

318

                          fakes.get_fake_uuid(),

 

319

                          mountpoint)

 

320

 

 

321

    def test_attach_attaching_volume_with_different_instance(self):

 

322

        """Test that attaching volume reserved for another instance fails."""

 

323

        # admin context

 

324

        ctx = context.RequestContext('admin', 'fake', True)

 

325

        # current status is available

 

326

        volume = db.volume_create(ctx, {'status': 'available', 'host': 'test',

 

327

                                        'provider_location': ''})

 

328

        # start service to handle rpc messages for attach requests

 

329

        self.start_service('volume', host='test')

 

330

        self.volume_api.initialize_connection(ctx, volume, {})

 

331

        values = {'status': 'attaching',

 

332

                  'instance_uuid': fakes.get_fake_uuid()}

 

333

        db.volume_update(ctx, volume['id'], values)

 

334

        mountpoint = '/dev/vbd'

 

335

        self.assertRaises(exception.InvalidVolume,

 

336

                          self.volume_api.attach,

 

337

                          ctx,

 

338

                          volume,

 

339

                          stubs.FAKE_UUID,

 

340

                          mountpoint)

• Older »

Loggerhead is a web-based interface for Breezy
Version: 2.0.1