1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
|
# Copyright (c) 2017 OpenStack Foundation
# All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
import datetime
from keystoneauth1 import exceptions as ks_exc
import mock
from oslo_utils.fixture import uuidsentinel as uuids
from nova import context as nova_context
from nova.notifications import base
from nova import test
from nova.tests.unit import fake_instance
from nova import utils
class TestNullSafeUtils(test.NoDBTestCase):
def test_null_safe_isotime(self):
dt = None
self.assertEqual('', base.null_safe_isotime(dt))
dt = datetime.datetime(second=1,
minute=1,
hour=1,
day=1,
month=1,
year=2017)
self.assertEqual(utils.strtime(dt), base.null_safe_isotime(dt))
def test_null_safe_str(self):
line = None
self.assertEqual('', base.null_safe_str(line))
line = 'test'
self.assertEqual(line, base.null_safe_str(line))
class TestSendInstanceUpdateNotification(test.NoDBTestCase):
@mock.patch('nova.notifications.objects.base.NotificationBase.emit',
new_callable=mock.NonCallableMock) # asserts not called
# TODO(mriedem): Rather than mock is_enabled, it would be better to
# configure oslo_messaging_notifications.driver=['noop']
@mock.patch('nova.rpc.NOTIFIER.is_enabled', return_value=False)
def test_send_versioned_instance_update_notification_disabled(self,
mock_enabled,
mock_info):
"""Tests the case that versioned notifications are disabled which makes
_send_versioned_instance_update_notification a noop.
"""
base._send_versioned_instance_update(mock.sentinel.ctxt,
mock.sentinel.instance,
mock.sentinel.payload,
mock.sentinel.host,
mock.sentinel.service)
@mock.patch.object(base, '_compute_states_payload')
@mock.patch('nova.rpc.get_notifier')
@mock.patch.object(base, 'info_from_instance')
def test_send_legacy_instance_update_notification(
self, mock_info, mock_get_notifier, mock_states,
):
"""Tests the case that versioned notifications are disabled and
assert that this does not prevent sending the unversioned
instance.update notification.
"""
self.flags(notification_format='unversioned', group='notifications')
base.send_instance_update_notification(mock.sentinel.ctxt,
mock.sentinel.instance)
mock_get_notifier.return_value.info.assert_called_once_with(
mock.sentinel.ctxt, 'compute.instance.update', mock.ANY)
mock_info.assert_called_once_with(
mock.sentinel.ctxt, mock.sentinel.instance, None,
populate_image_ref_url=True)
@mock.patch('nova.image.glance.API.generate_image_url',
side_effect=ks_exc.EndpointNotFound)
def test_info_from_instance_image_api_endpoint_not_found_no_token(
self, mock_gen_image_url):
"""Tests the case that we fail to generate the image ref url because
CONF.glance.api_servers isn't set and we have a context without an
auth token, like in the case of a periodic task using an admin context.
In this case, we expect the payload field 'image_ref_url' to just be
the instance.image_ref (image ID for a non-volume-backed server).
"""
ctxt = nova_context.get_admin_context()
instance = fake_instance.fake_instance_obj(ctxt, image_ref=uuids.image)
instance.system_metadata = {}
instance.metadata = {}
payload = base.info_from_instance(ctxt, instance, network_info=None,
populate_image_ref_url=True)
self.assertEqual(instance.image_ref, payload['image_ref_url'])
mock_gen_image_url.assert_called_once_with(instance.image_ref, ctxt)
@mock.patch('nova.image.glance.API.generate_image_url',
side_effect=ks_exc.EndpointNotFound)
def test_info_from_instance_image_api_endpoint_not_found_with_token(
self, mock_gen_image_url):
"""Tests the case that we fail to generate the image ref url because
an EndpointNotFound error is raised up from the image API but the
context does have a token so we pass the error through.
"""
ctxt = nova_context.RequestContext(
'fake-user', 'fake-project', auth_token='fake-token')
instance = fake_instance.fake_instance_obj(ctxt, image_ref=uuids.image)
self.assertRaises(ks_exc.EndpointNotFound, base.info_from_instance,
ctxt, instance, network_info=None,
populate_image_ref_url=True)
mock_gen_image_url.assert_called_once_with(instance.image_ref, ctxt)
@mock.patch('nova.image.glance.API.generate_image_url')
def test_info_from_instance_not_call_generate_image_url(
self, mock_gen_image_url):
ctxt = nova_context.get_admin_context()
instance = fake_instance.fake_instance_obj(ctxt, image_ref=uuids.image)
instance.system_metadata = {}
instance.metadata = {}
base.info_from_instance(ctxt, instance, network_info=None,
populate_image_ref_url=False)
mock_gen_image_url.assert_not_called()
|
