1
# vim: tabstop=4 shiftwidth=4 softtabstop=4
3
# Copyright 2011 OpenStack LLC.
6
# Licensed under the Apache License, Version 2.0 (the "License"); you may
7
# not use this file except in compliance with the License. You may obtain
8
# a copy of the License at
10
# http://www.apache.org/licenses/LICENSE-2.0
12
# Unless required by applicable law or agreed to in writing, software
13
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
14
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
15
# License for the specific language governing permissions and limitations
18
"""Tests For miscellaneous util methods used with volume."""
20
from nova import context
22
from nova import flags
23
from nova.openstack.common import importutils
24
from nova.openstack.common import log as logging
25
from nova.openstack.common.notifier import api as notifier_api
26
from nova.openstack.common.notifier import test_notifier
28
from nova.tests import fake_network
29
from nova.volume import utils as volume_utils
32
LOG = logging.getLogger(__name__)
36
class UsageInfoTestCase(test.TestCase):
39
super(UsageInfoTestCase, self).setUp()
40
self.flags(compute_driver='nova.virt.fake.FakeDriver',
42
notification_driver=[test_notifier.__name__])
43
fake_network.set_stub_network_methods(self.stubs)
45
self.volume = importutils.import_object(FLAGS.volume_manager)
47
self.project_id = 'fake'
48
self.snapshot_id = 'fake'
50
self.context = context.RequestContext(self.user_id, self.project_id)
51
test_notifier.NOTIFICATIONS = []
54
notifier_api._reset_drivers()
55
super(UsageInfoTestCase, self).tearDown()
57
def _create_volume(self, params={}):
58
"""Create a test volume"""
60
vol['snapshot_id'] = self.snapshot_id
61
vol['user_id'] = self.user_id
62
vol['project_id'] = self.project_id
63
vol['host'] = FLAGS.host
64
vol['availability_zone'] = FLAGS.storage_availability_zone
65
vol['status'] = "creating"
66
vol['attach_status'] = "detached"
67
vol['size'] = self.volume_size
69
return db.volume_create(self.context, vol)['id']
71
def test_notify_usage_exists(self):
72
"""Ensure 'exists' notification generates appropriate usage data."""
73
volume_id = self._create_volume()
74
volume = db.volume_get(self.context, volume_id)
75
volume_utils.notify_usage_exists(self.context, volume)
76
self.assertEquals(len(test_notifier.NOTIFICATIONS), 1)
77
msg = test_notifier.NOTIFICATIONS[0]
78
self.assertEquals(msg['priority'], 'INFO')
79
self.assertEquals(msg['event_type'], 'volume.exists')
80
payload = msg['payload']
81
self.assertEquals(payload['tenant_id'], self.project_id)
82
self.assertEquals(payload['user_id'], self.user_id)
83
self.assertEquals(payload['snapshot_id'], self.snapshot_id)
84
self.assertEquals(payload['volume_id'], volume.id)
85
self.assertEquals(payload['size'], self.volume_size)
86
for attr in ('display_name', 'created_at', 'launched_at',
87
'status', 'audit_period_beginning',
88
'audit_period_ending'):
89
self.assertTrue(attr in payload,
90
msg="Key %s not in payload" % attr)
91
db.volume_destroy(context.get_admin_context(), volume['id'])