openstack/nova
Code Issues Proposed changes
c50a39f8c8
nova/nova/tests/unit/fake_network.py
Chris Dent def4b17934 Use nova.db.api directly 
nova/db/__init__.py was importing * from nova.db.api. This meant that
any time any code anywhere within the nova.db package was imported
then nova.db.api was too, leading to a cascade of imports that may
not have been desired. Also, in general, code in __init__.py is a pain.

Therefore, this change adjusts code that so that either:

* nova.db.api is used directly
* nova.db.api is imported as 'db'

In either case, the functionality remains the same.

The primary goal of this change was to make it possible to import the
model files without having to import the db api. Moving the model files
to a different place in the directory hierarchy was considered, but
given that "code in __init__.py is a pain" this mode was chosen.

This looks like a very large change, but it is essentially adjusting
package names, many in mocks.

Change-Id: Ic1fd7c87ceda05eeb96735da2a415ef37060bb1a
2018-07-10 14:56:27 +00:00

478 lines
17 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 six.moves import range
from nova.compute import api as compute_api
from nova.compute import manager as compute_manager
import nova.conf
import nova.context
from nova.db import api as db
from nova import exception
from nova.network import manager as network_manager
from nova.network import model as network_model
from nova.network import rpcapi as network_rpcapi
from nova import objects
from nova.objects import base as obj_base
from nova.objects import network as network_obj
from nova.objects import virtual_interface as vif_obj
from nova.tests.unit.objects import test_fixed_ip
from nova.tests.unit.objects import test_instance_info_cache
from nova.tests.unit.objects import test_pci_device
from nova.tests.unit import utils
from nova.tests import uuidsentinel as uuids
HOST = "testhost"
CONF = nova.conf.CONF
class FakeModel(dict):
"""Represent a model from the db."""
def __init__(self, *args, **kwargs):
self.update(kwargs)
class FakeNetworkManager(network_manager.NetworkManager):
"""This NetworkManager doesn't call the base class so we can bypass all
inherited service cruft and just perform unit tests.
"""
class FakeDB(object):
vifs = [{'id': 0,
'created_at': None,
'updated_at': None,
'deleted_at': None,
'deleted': 0,
'instance_uuid': uuids.instance_1,
'network_id': 1,
'uuid': uuids.vifs_1,
'address': 'DC:AD:BE:FF:EF:01',
'tag': 'fake-tag1'},
{'id': 1,
'created_at': None,
'updated_at': None,
'deleted_at': None,
'deleted': 0,
'instance_uuid': uuids.instance_2,
'network_id': 21,
'uuid': uuids.vifs_2,
'address': 'DC:AD:BE:FF:EF:02',
'tag': 'fake-tag2'},
{'id': 2,
'created_at': None,
'updated_at': None,
'deleted_at': None,
'deleted': 0,
'instance_uuid': uuids.instance_1,
'network_id': 31,
'uuid': uuids.vifs_3,
'address': 'DC:AD:BE:FF:EF:03',
'tag': None}]
floating_ips = [dict(address='172.16.1.1',
fixed_ip_id=100),
dict(address='172.16.1.2',
fixed_ip_id=200),
dict(address='173.16.1.2',
fixed_ip_id=210)]
fixed_ips = [dict(test_fixed_ip.fake_fixed_ip,
id=100,
address='172.16.0.1',
virtual_interface_id=0),
dict(test_fixed_ip.fake_fixed_ip,
id=200,
address='172.16.0.2',
virtual_interface_id=1),
dict(test_fixed_ip.fake_fixed_ip,
id=210,
address='173.16.0.2',
virtual_interface_id=2)]
def fixed_ip_get_by_instance(self, context, instance_uuid):
return [dict(address='10.0.0.0'), dict(address='10.0.0.1'),
dict(address='10.0.0.2')]
def network_get_by_cidr(self, context, cidr):
raise exception.NetworkNotFoundForCidr(cidr=cidr)
def network_create_safe(self, context, net):
fakenet = dict(net)
fakenet['id'] = 999
return fakenet
def network_get(self, context, network_id, project_only="allow_none"):
return {'cidr_v6': '2001:db8:69:%x::/64' % network_id}
def network_get_by_uuid(self, context, network_uuid):
raise exception.NetworkNotFoundForUUID(uuid=network_uuid)
def network_get_all(self, context):
raise exception.NoNetworksFound()
def network_get_all_by_uuids(self, context, project_only="allow_none"):
raise exception.NoNetworksFound()
def network_disassociate(self, context, network_id):
return True
def virtual_interface_get_all(self, context):
return self.vifs
def fixed_ips_by_virtual_interface(self, context, vif_id):
return [ip for ip in self.fixed_ips
if ip['virtual_interface_id'] == vif_id]
def fixed_ip_disassociate(self, context, address):
return True
def __init__(self, stubs=None):
self.db = self.FakeDB()
if stubs:
stubs.Set(vif_obj, 'db', self.db)
self.deallocate_called = None
self.deallocate_fixed_ip_calls = []
self.network_rpcapi = network_rpcapi.NetworkAPI()
# TODO(matelakat) method signature should align with the faked one's
def deallocate_fixed_ip(self, context, address=None, host=None,
instance=None):
self.deallocate_fixed_ip_calls.append((context, address, host))
# TODO(matelakat) use the deallocate_fixed_ip_calls instead
self.deallocate_called = address
def _create_fixed_ips(self, context, network_id, fixed_cidr=None,
extra_reserved=None, bottom_reserved=0,
top_reserved=0):
pass
def get_instance_nw_info(context, instance_id, rxtx_factor,
host, instance_uuid=None, **kwargs):
pass
def fake_network(network_id, ipv6=None):
if ipv6 is None:
ipv6 = CONF.use_ipv6
fake_network = {'id': network_id,
'uuid': getattr(uuids, 'network%i' % network_id),
'label': 'test%d' % network_id,
'injected': False,
'multi_host': False,
'cidr': '192.168.%d.0/24' % network_id,
'cidr_v6': None,
'netmask': '255.255.255.0',
'netmask_v6': None,
'bridge': 'fake_br%d' % network_id,
'bridge_interface': 'fake_eth%d' % network_id,
'gateway': '192.168.%d.1' % network_id,
'gateway_v6': None,
'broadcast': '192.168.%d.255' % network_id,
'dns1': '192.168.%d.3' % network_id,
'dns2': '192.168.%d.4' % network_id,
'dns3': '192.168.%d.3' % network_id,
'vlan': None,
'host': None,
'project_id': uuids.project,
'vpn_public_address': '192.168.%d.2' % network_id,
'vpn_public_port': None,
'vpn_private_address': None,
'dhcp_start': None,
'rxtx_base': network_id * 10,
'priority': None,
'deleted': False,
'created_at': None,
'updated_at': None,
'deleted_at': None,
'mtu': None,
'dhcp_server': '192.168.%d.1' % network_id,
'enable_dhcp': True,
'share_address': False}
if ipv6:
fake_network['cidr_v6'] = '2001:db8:0:%x::/64' % network_id
fake_network['gateway_v6'] = '2001:db8:0:%x::1' % network_id
fake_network['netmask_v6'] = '64'
if CONF.flat_injected:
fake_network['injected'] = True
return fake_network
def fake_network_obj(context, network_id=1, ipv6=None):
return network_obj.Network._from_db_object(
context, network_obj.Network(), fake_network(network_id, ipv6))
def fake_vif(x):
return {'id': x,
'created_at': None,
'updated_at': None,
'deleted_at': None,
'deleted': 0,
'address': 'DE:AD:BE:EF:00:%02x' % x,
'uuid': getattr(uuids, 'vif%i' % x),
'network_id': x,
'instance_uuid': uuids.vifs_1,
'tag': 'fake-tag'}
def floating_ip_ids():
for i in range(1, 100):
yield i
def fixed_ip_ids():
for i in range(1, 100):
yield i
floating_ip_id = floating_ip_ids()
fixed_ip_id = fixed_ip_ids()
def next_fixed_ip(network_id, num_floating_ips=0):
next_id = next(fixed_ip_id)
f_ips = [FakeModel(**next_floating_ip(next_id))
for i in range(num_floating_ips)]
return {'id': next_id,
'network_id': network_id,
'address': '192.168.%d.%03d' % (network_id, (next_id + 99)),
'instance_uuid': uuids.fixed_ip,
'allocated': False,
'reserved': False,
'created_at': None,
'updated_at': None,
'deleted_at': None,
'leased': True,
'host': HOST,
'deleted': 0,
'network': fake_network(network_id),
'virtual_interface': fake_vif(network_id),
# and since network_id and vif_id happen to be equivalent
'virtual_interface_id': network_id,
'floating_ips': f_ips}
def next_floating_ip(fixed_ip_id):
next_id = next(floating_ip_id)
return {'id': next_id,
'address': '10.10.10.%03d' % (next_id + 99),
'fixed_ip_id': fixed_ip_id,
'project_id': None,
'auto_assigned': False}
def ipv4_like(ip, match_string):
ip = ip.split('.')
match_octets = match_string.split('.')
for i, octet in enumerate(match_octets):
if octet == '*':
continue
if octet != ip[i]:
return False
return True
def fake_get_instance_nw_info(test, num_networks=1, ips_per_vif=2,
floating_ips_per_fixed_ip=0):
# test is an instance of nova.test.TestCase
# ips_per_vif is the number of ips each vif will have
# num_floating_ips is number of float ips for each fixed ip
network = network_manager.FlatManager(host=HOST)
network.db = db
# reset the fixed and floating ip generators
global floating_ip_id, fixed_ip_id, fixed_ips
floating_ip_id = floating_ip_ids()
fixed_ip_id = fixed_ip_ids()
fixed_ips = []
def fixed_ips_fake(*args, **kwargs):
global fixed_ips
ips = [next_fixed_ip(i, floating_ips_per_fixed_ip)
for i in range(1, num_networks + 1)
for j in range(ips_per_vif)]
fixed_ips = ips
return ips
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.fixed_ip_get_by_instance', fixed_ips_fake)
test.stub_out('nova.db.api.instance_info_cache_update', update_cache_fake)
class FakeContext(nova.context.RequestContext):
def is_admin(self):
return True
nw_model = network.get_instance_nw_info(
FakeContext('fakeuser', 'fake_project'),
0, 3, None)
return nw_model
def stub_out_nw_api_get_instance_nw_info(test, func=None,
num_networks=1,
ips_per_vif=1,
floating_ips_per_fixed_ip=0):
def get_instance_nw_info(self, context, instance, conductor_api=None):
return fake_get_instance_nw_info(test, num_networks=num_networks,
ips_per_vif=ips_per_vif,
floating_ips_per_fixed_ip=floating_ips_per_fixed_ip)
if func is None:
func = get_instance_nw_info
test.stub_out('nova.network.api.API.get_instance_nw_info', func)
def stub_out_network_cleanup(test):
test.stub_out('nova.network.api.API.deallocate_for_instance',
lambda *args, **kwargs: None)
_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 stub_compute_with_ips(test):
orig_get = compute_api.API.get
orig_get_all = compute_api.API.get_all
orig_create = compute_api.API.create
def fake_get(*args, **kwargs):
return _get_instances_with_cached_ips(orig_get, *args, **kwargs)
def fake_get_all(*args, **kwargs):
return _get_instances_with_cached_ips(orig_get_all, *args, **kwargs)
def fake_create(*args, **kwargs):
return _create_instances_with_cached_ips(orig_create, *args, **kwargs)
def fake_pci_device_get_by_addr(context, node_id, dev_addr):
return test_pci_device.fake_db_dev
test.stub_out('nova.db.api.pci_device_get_by_addr',
fake_pci_device_get_by_addr)
test.stub_out('nova.compute.api.API.get', fake_get)
test.stub_out('nova.compute.api.API.get_all', fake_get_all)
test.stub_out('nova.compute.api.API.create', fake_create)
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')]}]}}]
if CONF.use_ipv6:
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