openstack
/
nova
Code Issues Proposed changes
nova/nova/tests/unit/fake_network.py

227 lines
8.1 KiB
Python
Raw Blame History

# Copyright 2011 Rackspace
# 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.
from oslo_serialization import jsonutils
from oslo_utils.fixture import uuidsentinel as uuids
from nova.compute import manager as compute_manager
from nova.db import api as db
from nova.network import model as network_model
from nova import objects
from nova.objects import base as obj_base
from nova.tests.unit.objects import test_instance_info_cache
from nova.tests.unit import utils
def fake_get_instance_nw_info(test, num_networks=1):
def update_cache_fake(*args, **kwargs):
fake_info_cache = {
'created_at': None,
'updated_at': None,
'deleted_at': None,
'deleted': False,
'instance_uuid': uuids.vifs_1,
'network_info': '[]',
}
return fake_info_cache
test.stub_out('nova.db.api.instance_info_cache_update', update_cache_fake)
# TODO(stephenfin): This doesn't match the kind of object we would receive
# from '_build_vif_model' and callers of same. We should fix that.
nw_model = network_model.NetworkInfo()
for network_id in range(1, num_networks + 1):
network = network_model.Network(
id=getattr(uuids, 'network%i' % network_id),
bridge='fake_br%d' % network_id,
label='test%d' % network_id,
subnets=[
network_model.Subnet(
cidr='192.168.%d.0/24' % network_id,
dns=[
network_model.IP(
address='192.168.%d.3' % network_id,
type='dns',
version=4,
meta={},
),
network_model.IP(
address='192.168.%d.4' % network_id,
type='dns',
version=4,
meta={},
),
],
gateway=network_model.IP(
address='192.168.%d.1' % network_id,
type='gateway',
version=4,
meta={},
),
ips=[
network_model.FixedIP(
address='192.168.%d.100' % network_id,
version=4,
meta={},
),
],
routes=[],
version=4,
meta={},
),
network_model.Subnet(
cidr='2001:db8:0:%x::/64' % network_id,
dns=[],
gateway=network_model.IP(
address='2001:db8:0:%x::1' % network_id,
type='gateway',
version=6,
meta={},
),
ips=[
network_model.FixedIP(
address='2001:db8:0:%x:dcad:beff:feef:1' % (
network_id),
version=6,
meta={},
),
],
routes=[],
version=6,
meta={}
),
],
meta={
"tenant_id": "806e1f03-b36f-4fc6-be29-11a366f150eb"
},
)
vif = network_model.VIF(
id=getattr(uuids, 'vif%i' % network_id),
address='DE:AD:BE:EF:00:%02x' % network_id,
network=network,
type='bridge',
details={},
devname=None,
ovs_interfaceid=None,
qbh_params=None,
qbg_params=None,
active=False,
vnic_type='normal',
profile=None,
preserve_on_delete=False,
meta={'rxtx_cap': 30},
)
nw_model.append(vif)
return nw_model
_real_functions = {}
def set_stub_network_methods(test):
global _real_functions
cm = compute_manager.ComputeManager
if not _real_functions:
_real_functions = {
'_allocate_network': cm._allocate_network,
'_deallocate_network': cm._deallocate_network}
def fake_networkinfo(*args, **kwargs):
return network_model.NetworkInfo()
def fake_async_networkinfo(*args, **kwargs):
return network_model.NetworkInfoAsyncWrapper(fake_networkinfo)
test.stub_out('nova.compute.manager.ComputeManager._allocate_network',
fake_async_networkinfo)
test.stub_out('nova.compute.manager.ComputeManager._deallocate_network',
lambda *args, **kwargs: None)
def unset_stub_network_methods(test):
global _real_functions
if _real_functions:
for name in _real_functions:
test.stub_out('nova.compute.manager.ComputeManager.' + name,
_real_functions[name])
def _get_fake_cache():
def _ip(ip, fixed=True, floats=None):
ip_dict = {'address': ip, 'type': 'fixed'}
if not fixed:
ip_dict['type'] = 'floating'
if fixed and floats:
ip_dict['floating_ips'] = [_ip(f, fixed=False) for f in floats]
return ip_dict
info = [{'address': 'aa:bb:cc:dd:ee:ff',
'id': utils.FAKE_NETWORK_UUID,
'network': {'bridge': 'br0',
'id': 1,
'label': 'private',
'subnets': [{'cidr': '192.168.0.0/24',
'ips': [_ip('192.168.0.3')]}]}}]
ipv6_addr = 'fe80:b33f::a8bb:ccff:fedd:eeff'
info[0]['network']['subnets'].append({'cidr': 'fe80:b33f::/64',
'ips': [_ip(ipv6_addr)]})
return jsonutils.dumps(info)
def _get_instances_with_cached_ips(orig_func, *args, **kwargs):
"""Kludge the cache into instance(s) without having to create DB
entries
"""
instances = orig_func(*args, **kwargs)
context = args[0]
fake_device = objects.PciDevice.get_by_dev_addr(context, 1, 'a')
def _info_cache_for(instance):
info_cache = dict(test_instance_info_cache.fake_info_cache,
network_info=_get_fake_cache(),
instance_uuid=instance['uuid'])
if isinstance(instance, obj_base.NovaObject):
_info_cache = objects.InstanceInfoCache(context)
objects.InstanceInfoCache._from_db_object(context, _info_cache,
info_cache)
info_cache = _info_cache
instance['info_cache'] = info_cache
if isinstance(instances, (list, obj_base.ObjectListBase)):
for instance in instances:
_info_cache_for(instance)
fake_device.claim(instance.uuid)
fake_device.allocate(instance)
else:
_info_cache_for(instances)
fake_device.claim(instances.uuid)
fake_device.allocate(instances)
return instances
def _create_instances_with_cached_ips(orig_func, *args, **kwargs):
"""Kludge the above kludge so that the database doesn't get out
of sync with the actual instance.
"""
instances, reservation_id = orig_func(*args, **kwargs)
fake_cache = _get_fake_cache()
for instance in instances:
instance['info_cache'].network_info = fake_cache
db.instance_info_cache_update(args[1], instance['uuid'],
{'network_info': fake_cache})
return instances, reservation_id
View Git Blame Copy Permalink