openstack
/
nova
Code Issues Proposed changes

Fixed review comments

This commit is contained in:
Tushar Patil 2011-08-19 14:25:41 -07:00
parent 10fdf23403
commit d27a4d4a59
6 changed files with 108 additions and 172 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

21
nova/db/api.py
View File

@ -323,13 +323,13 @@ def migration_get_by_instance_and_status(context, instance_uuid, status):
#################### ####################
def fixed_ip_associate(context, address, instance_id): def fixed_ip_associate(context, address, instance_id, network_id=None):
"""Associate fixed ip to instance. """Associate fixed ip to instance.
Raises if fixed ip is not available. Raises if fixed ip is not available.
""" """
return IMPL.fixed_ip_associate(context, address, instance_id) return IMPL.fixed_ip_associate(context, address, instance_id, network_id)
def fixed_ip_associate_pool(context, network_id, instance_id=None, host=None): def fixed_ip_associate_pool(context, network_id, instance_id=None, host=None):
@ -342,14 +342,6 @@ def fixed_ip_associate_pool(context, network_id, instance_id=None, host=None):
instance_id, host) instance_id, host)
def fixed_ip_associate_by_address(context, network_id, instance_id, address):
"""check if the address is free and is in the network
and it is not associated to any instance.
"""
return IMPL.fixed_ip_associate_by_address(context, network_id,
instance_id, address)
def fixed_ip_create(context, values): def fixed_ip_create(context, values):
"""Create a fixed ip from the values dictionary.""" """Create a fixed ip from the values dictionary."""
return IMPL.fixed_ip_create(context, values) return IMPL.fixed_ip_create(context, values)
@ -687,9 +679,9 @@ def network_get_all(context):
return IMPL.network_get_all(context) return IMPL.network_get_all(context)
def network_get_networks_by_uuids(context, network_uuids): def network_get_all_by_uuids(context, network_uuids, project_id=None):
"""Return networks by ids.""" """Return networks by ids."""
return IMPL.network_get_networks_by_uuids(context, network_uuids) return IMPL.network_get_all_by_uuids(context, network_uuids, project_id)
# pylint: disable=C0103 # pylint: disable=C0103
@ -1244,11 +1236,6 @@ def project_get_networks(context, project_id, associate=True):
return IMPL.project_get_networks(context, project_id, associate) return IMPL.project_get_networks(context, project_id, associate)
def project_get_networks_by_uuids(context, network_uuids):
"""Return the networks by uuids associated with the project."""
return IMPL.project_get_networks_by_uuids(context, network_uuids)
def project_get_networks_v6(context, project_id): def project_get_networks_v6(context, project_id):
return IMPL.project_get_networks_v6(context, project_id) return IMPL.project_get_networks_v6(context, project_id)

104
nova/db/sqlalchemy/api.py
View File

@ -266,7 +266,7 @@ def service_get_all_network_sorted(context):
session = get_session() session = get_session()
with session.begin(): with session.begin():
topic = 'network' topic = 'network'
label = 'network_' label = 'network_count'
subq = session.query(models.Network.host, subq = session.query(models.Network.host,
func.count(models.Network.id).label(label)).\ func.count(models.Network.id).label(label)).\
filter_by(deleted=False).\ filter_by(deleted=False).\
@ -652,23 +652,36 @@ def floating_ip_update(context, address, values):
################### ###################
@require_context @require_admin_context
def fixed_ip_associate(context, address, instance_id): def fixed_ip_associate(context, address, instance_id, network_id=None):
session = get_session() session = get_session()
with session.begin(): with session.begin():
instance = instance_get(context, instance_id, session=session) network_or_none = or_(models.FixedIp.network_id == network_id,
models.FixedIp.network_id == None)
fixed_ip_ref = session.query(models.FixedIp).\ fixed_ip_ref = session.query(models.FixedIp).\
filter_by(address=address).\ filter(network_or_none).\
filter_by(reserved=False).\
filter_by(deleted=False).\ filter_by(deleted=False).\
filter_by(instance=None).\ filter_by(address=address).\
with_lockmode('update').\ with_lockmode('update').\
first() first()
# NOTE(vish): if with_lockmode isn't supported, as in sqlite, # NOTE(vish): if with_lockmode isn't supported, as in sqlite,
# then this has concurrency issues # then this has concurrency issues
if not fixed_ip_ref: if fixed_ip_ref is None:
raise exception.NoMoreFixedIps() raise exception.FixedIpNotFoundForNetwork(address=address,
fixed_ip_ref.instance = instance network_id=network_id)
if fixed_ip_ref.instance is not None:
raise exception.FixedIpAlreadyInUse(address=address)
if not fixed_ip_ref.network:
fixed_ip_ref.network = network_get(context,
network_id,
session=session)
fixed_ip_ref.instance = instance_get(context,
instance_id,
session=session)
session.add(fixed_ip_ref) session.add(fixed_ip_ref)
return fixed_ip_ref['address']
@require_admin_context @require_admin_context
@ -703,40 +716,6 @@ def fixed_ip_associate_pool(context, network_id, instance_id=None, host=None):
return fixed_ip_ref['address'] return fixed_ip_ref['address']
@require_admin_context
def fixed_ip_associate_by_address(context, network_id, instance_id,
address):
if address is None:
return fixed_ip_associate_pool(context, network_id, instance_id)
session = get_session()
with session.begin():
fixed_ip_ref = session.query(models.FixedIp).\
filter_by(reserved=False).\
filter_by(deleted=False).\
filter_by(network_id=network_id).\
filter_by(address=address).\
with_lockmode('update').\
first()
# NOTE(vish): if with_lockmode isn't supported, as in sqlite,
# then this has concurrency issues
if fixed_ip_ref is None:
raise exception.FixedIpNotFoundForNetwork(address=address,
network_id=network_id)
if fixed_ip_ref.instance is not None:
raise exception.FixedIpAlreadyInUse(address=address)
if not fixed_ip_ref.network:
fixed_ip_ref.network = network_get(context,
network_id,
session=session)
fixed_ip_ref.instance = instance_get(context,
instance_id,
session=session)
session.add(fixed_ip_ref)
return fixed_ip_ref['address']
@require_context @require_context
def fixed_ip_create(_context, values): def fixed_ip_create(_context, values):
fixed_ip_ref = models.FixedIp() fixed_ip_ref = models.FixedIp()
@ -1777,10 +1756,13 @@ def network_get_all(context):
@require_admin_context @require_admin_context
def network_get_networks_by_uuids(context, network_uuids): def network_get_all_by_uuids(context, network_uuids, project_id=None):
session = get_session() session = get_session()
project_or_none = or_(models.Network.project_id == project_id,
models.Network.project_id == None)
result = session.query(models.Network).\ result = session.query(models.Network).\
filter(models.Network.uuid.in_(network_uuids)).\ filter(models.Network.uuid.in_(network_uuids)).\
filter(project_or_none).\
filter_by(deleted=False).all() filter_by(deleted=False).all()
if not result: if not result:
raise exception.NoNetworksFound() raise exception.NoNetworksFound()
@ -1800,6 +1782,9 @@ def network_get_networks_by_uuids(context, network_uuids):
found = True found = True
break break
if not found: if not found:
if project_id:
raise exception.NetworkNotFoundForProject(network_uuid=uuid,
project_id=context.project_id)
raise exception.NetworkNotFound(network_id=network_uuid) raise exception.NetworkNotFound(network_id=network_uuid)
return result return result
@ -2961,37 +2946,6 @@ def project_get_networks(context, project_id, associate=True):
return result return result
@require_context
def project_get_networks_by_uuids(context, network_uuids):
session = get_session()
result = session.query(models.Network).\
filter(models.Network.uuid.in_(network_uuids)).\
filter_by(deleted=False).\
filter_by(project_id=context.project_id).all()
if not result:
raise exception.NoNetworksFound()
#check if host is set to all of the networks
# returned in the result
for network in result:
if network['host'] is None:
raise exception.NetworkHostNotSet(network_id=network['id'])
#check if the result contains all the networks
#we are looking for
for uuid in network_uuids:
found = False
for network in result:
if network['uuid'] == uuid:
found = True
break
if not found:
raise exception.NetworkNotFoundForProject(network_uuid=uuid,
project_id=context.project_id)
return result
@require_context @require_context
def project_get_networks_v6(context, project_id): def project_get_networks_v6(context, project_id):
return project_get_networks(context, project_id) return project_get_networks(context, project_id)

0
nova/db/sqlalchemy/migrate_repo/versions/037_add_uuid_to_networks.py → nova/db/sqlalchemy/migrate_repo/versions/039_add_uuid_to_networks.py
View File

6
nova/exception.py
View File

@ -185,10 +185,6 @@ class Invalid(NovaException):
message = _("Unacceptable parameters.") message = _("Unacceptable parameters.")
class AlreadyInUse(NovaException):
message = _("Already is in use.")
class InvalidSignature(Invalid): class InvalidSignature(Invalid):
message = _("Invalid signature %(signature)s for user %(user)s.") message = _("Invalid signature %(signature)s for user %(user)s.")
@ -474,7 +470,7 @@ class FixedIpNotFoundForNetwork(FixedIpNotFound):
"network (%(network_uuid)s).") "network (%(network_uuid)s).")
class FixedIpAlreadyInUse(AlreadyInUse): class FixedIpAlreadyInUse(NovaException):
message = _("Fixed IP address %(address)s is already in use.") message = _("Fixed IP address %(address)s is already in use.")

68
nova/network/manager.py
View File

@ -399,8 +399,8 @@ class NetworkManager(manager.SchedulerDependentManager):
# a non-vlan instance should connect to # a non-vlan instance should connect to
if requested_networks is not None and len(requested_networks) != 0: if requested_networks is not None and len(requested_networks) != 0:
network_uuids = [uuid for (uuid, fixed_ip) in requested_networks] network_uuids = [uuid for (uuid, fixed_ip) in requested_networks]
networks = self.db.network_get_networks_by_uuids(context, networks = self.db.network_get_all_by_uuids(context,
network_uuids) network_uuids)
else: else:
try: try:
networks = self.db.network_get_all(context) networks = self.db.network_get_all(context)
@ -588,10 +588,15 @@ class NetworkManager(manager.SchedulerDependentManager):
# network_get_by_compute_host # network_get_by_compute_host
address = None address = None
if network['cidr']: if network['cidr']:
address = self.db.fixed_ip_associate_by_address(context.elevated(), address = kwargs.get('address', None)
network['id'], if address:
instance_id, address = self.db.fixed_ip_associate(context,
kwargs.get('address', None)) address, instance_id,
network['id'])
else:
address = self.db.fixed_ip_associate_pool(context.elevated(),
network['id'],
instance_id)
self._do_trigger_security_group_members_refresh_for_instance( self._do_trigger_security_group_members_refresh_for_instance(
instance_id) instance_id)
get_vif = self.db.virtual_interface_get_by_instance_and_network get_vif = self.db.virtual_interface_get_by_instance_and_network
@ -826,7 +831,7 @@ class NetworkManager(manager.SchedulerDependentManager):
network_uuids = [uuid for (uuid, fixed_ip) in networks] network_uuids = [uuid for (uuid, fixed_ip) in networks]
self.db.network_get_networks_by_uuids(context, network_uuids) self._get_networks_by_uuids(context, network_uuids)
for network_uuid, address in networks: for network_uuid, address in networks:
# check if the fixed IP address is valid and # check if the fixed IP address is valid and
@ -843,6 +848,9 @@ class NetworkManager(manager.SchedulerDependentManager):
if fixed_ip_ref['instance'] is not None: if fixed_ip_ref['instance'] is not None:
raise exception.FixedIpAlreadyInUse(address=address) raise exception.FixedIpAlreadyInUse(address=address)
def _get_networks_by_uuids(self, context, network_uuids):
return self.db.network_get_all_by_uuids(context, network_uuids)
class FlatManager(NetworkManager): class FlatManager(NetworkManager):
"""Basic network where no vlans are used. """Basic network where no vlans are used.
@ -980,10 +988,15 @@ class VlanManager(RPCAllocateFixedIP, FloatingIP, NetworkManager):
address, address,
instance_id) instance_id)
else: else:
address = self.db.fixed_ip_associate_by_address(context, address = kwargs.get('address', None)
network['id'], if address:
instance_id, address = self.db.fixed_ip_associate(context, address,
kwargs.get('address', None)) instance_id,
network['id'])
else:
address = self.db.fixed_ip_associate_pool(context,
network['id'],
instance_id)
self._do_trigger_security_group_members_refresh_for_instance( self._do_trigger_security_group_members_refresh_for_instance(
instance_id) instance_id)
vif = self.db.virtual_interface_get_by_instance_and_network(context, vif = self.db.virtual_interface_get_by_instance_and_network(context,
@ -1005,8 +1018,9 @@ class VlanManager(RPCAllocateFixedIP, FloatingIP, NetworkManager):
# get networks associated with project # get networks associated with project
if requested_networks is not None and len(requested_networks) != 0: if requested_networks is not None and len(requested_networks) != 0:
network_uuids = [uuid for (uuid, fixed_ip) in requested_networks] network_uuids = [uuid for (uuid, fixed_ip) in requested_networks]
networks = self.db.project_get_networks_by_uuids(context, networks = self.db.network_get_all_by_uuids(context,
network_uuids) network_uuids,
project_id)
else: else:
networks = self.db.project_get_networks(context, project_id) networks = self.db.project_get_networks(context, project_id)
return networks return networks
@ -1058,31 +1072,9 @@ class VlanManager(RPCAllocateFixedIP, FloatingIP, NetworkManager):
self.db.network_update(context, network_ref['id'], self.db.network_update(context, network_ref['id'],
{'gateway_v6': gateway}) {'gateway_v6': gateway})
def validate_networks(self, context, networks): def _get_networks_by_uuids(self, context, network_uuids):
"""check if the networks exists and host return self.db.network_get_all_by_uuids(context, network_uuids,
is set to each network. context.project_id)
"""
if networks is None or len(networks) == 0:
return
network_uuids = [uuid for (uuid, fixed_ip) in networks]
self.db.project_get_networks_by_uuids(context, network_uuids)
for network_uuid, address in networks:
# check if the fixed IP address is valid and
# it actually belongs to the network
if fixed_ip is not None:
if not utils.is_valid_ipv4(address):
raise exception.FixedIpInvalid(address=address)
fixed_ip_ref = self.db.fixed_ip_get_by_address(context,
address)
if fixed_ip_ref['network']['uuid'] != network_uuid:
raise exception.FixedIpNotFoundForNetwork(address=address,
network_uuid=network_uuid)
if fixed_ip_ref['instance'] is not None:
raise exception.FixedIpAlreadyInUse(address=address)
@property @property
def _bottom_reserved_ips(self): def _bottom_reserved_ips(self):

81
nova/tests/test_network.py
View File

@ -15,6 +15,7 @@
# License for the specific language governing permissions and limitations # License for the specific language governing permissions and limitations
# under the License. # under the License.
from nova import context
from nova import db from nova import db
from nova import exception from nova import exception
from nova import log as logging from nova import log as logging
@ -128,6 +129,8 @@ class FlatNetworkTestCase(test.TestCase):
super(FlatNetworkTestCase, self).setUp() super(FlatNetworkTestCase, self).setUp()
self.network = network_manager.FlatManager(host=HOST) self.network = network_manager.FlatManager(host=HOST)
self.network.db = db self.network.db = db
self.context = context.RequestContext('testuser', 'testproject',
is_admin=False)
def test_get_instance_nw_info(self): def test_get_instance_nw_info(self):
self.mox.StubOutWithMock(db, 'fixed_ip_get_by_instance') self.mox.StubOutWithMock(db, 'fixed_ip_get_by_instance')
@ -186,13 +189,13 @@ class FlatNetworkTestCase(test.TestCase):
self.assertDictListMatch(nw[1]['ips'], check) self.assertDictListMatch(nw[1]['ips'], check)
def test_validate_networks(self): def test_validate_networks(self):
self.mox.StubOutWithMock(db, 'network_get_networks_by_uuids') self.mox.StubOutWithMock(db, 'network_get_all_by_uuids')
self.mox.StubOutWithMock(db, "fixed_ip_get_by_address") self.mox.StubOutWithMock(db, "fixed_ip_get_by_address")
requested_networks = [("bbbbbbbb-bbbb-bbbb-bbbb-bbbbbbbbbbbb", requested_networks = [("bbbbbbbb-bbbb-bbbb-bbbb-bbbbbbbbbbbb",
"192.168.1.100")] "192.168.1.100")]
db.network_get_networks_by_uuids(mox.IgnoreArg(), db.network_get_all_by_uuids(mox.IgnoreArg(),
mox.IgnoreArg()).AndReturn(networks) mox.IgnoreArg()).AndReturn(networks)
fixed_ips[1]['network'] = FakeModel(**networks[1]) fixed_ips[1]['network'] = FakeModel(**networks[1])
fixed_ips[1]['instance'] = None fixed_ips[1]['instance'] = None
@ -200,22 +203,22 @@ class FlatNetworkTestCase(test.TestCase):
mox.IgnoreArg()).AndReturn(fixed_ips[1]) mox.IgnoreArg()).AndReturn(fixed_ips[1])
self.mox.ReplayAll() self.mox.ReplayAll()
self.network.validate_networks(None, requested_networks) self.network.validate_networks(self.context, requested_networks)
def test_validate_networks_none_requested_networks(self): def test_validate_networks_none_requested_networks(self):
self.network.validate_networks(None, None) self.network.validate_networks(self.context, None)
def test_validate_networks_empty_requested_networks(self): def test_validate_networks_empty_requested_networks(self):
requested_networks = [] requested_networks = []
self.mox.ReplayAll() self.mox.ReplayAll()
self.network.validate_networks(None, requested_networks) self.network.validate_networks(self.context, requested_networks)
def test_validate_networks_invalid_fixed_ip(self): def test_validate_networks_invalid_fixed_ip(self):
self.mox.StubOutWithMock(db, 'network_get_networks_by_uuids') self.mox.StubOutWithMock(db, 'network_get_all_by_uuids')
requested_networks = [(1, "192.168.0.100.1")] requested_networks = [(1, "192.168.0.100.1")]
db.network_get_networks_by_uuids(mox.IgnoreArg(), db.network_get_all_by_uuids(mox.IgnoreArg(),
mox.IgnoreArg()).AndReturn(networks) mox.IgnoreArg()).AndReturn(networks)
self.mox.ReplayAll() self.mox.ReplayAll()
self.assertRaises(exception.FixedIpInvalid, self.assertRaises(exception.FixedIpInvalid,
@ -223,11 +226,11 @@ class FlatNetworkTestCase(test.TestCase):
requested_networks) requested_networks)
def test_validate_networks_empty_fixed_ip(self): def test_validate_networks_empty_fixed_ip(self):
self.mox.StubOutWithMock(db, 'network_get_networks_by_uuids') self.mox.StubOutWithMock(db, 'network_get_all_by_uuids')
requested_networks = [(1, "")] requested_networks = [(1, "")]
db.network_get_networks_by_uuids(mox.IgnoreArg(), db.network_get_all_by_uuids(mox.IgnoreArg(),
mox.IgnoreArg()).AndReturn(networks) mox.IgnoreArg()).AndReturn(networks)
self.mox.ReplayAll() self.mox.ReplayAll()
self.assertRaises(exception.FixedIpInvalid, self.assertRaises(exception.FixedIpInvalid,
@ -235,10 +238,10 @@ class FlatNetworkTestCase(test.TestCase):
None, requested_networks) None, requested_networks)
def test_validate_networks_none_fixed_ip(self): def test_validate_networks_none_fixed_ip(self):
self.mox.StubOutWithMock(db, 'network_get_networks_by_uuids') self.mox.StubOutWithMock(db, 'network_get_all_by_uuids')
requested_networks = [(1, None)] requested_networks = [(1, None)]
db.network_get_networks_by_uuids(mox.IgnoreArg(), db.network_get_all_by_uuids(mox.IgnoreArg(),
mox.IgnoreArg()).AndReturn(networks) mox.IgnoreArg()).AndReturn(networks)
self.mox.ReplayAll() self.mox.ReplayAll()
@ -250,6 +253,8 @@ class VlanNetworkTestCase(test.TestCase):
super(VlanNetworkTestCase, self).setUp() super(VlanNetworkTestCase, self).setUp()
self.network = network_manager.VlanManager(host=HOST) self.network = network_manager.VlanManager(host=HOST)
self.network.db = db self.network.db = db
self.context = context.RequestContext('testuser', 'testproject',
is_admin=False)
def test_vpn_allocate_fixed_ip(self): def test_vpn_allocate_fixed_ip(self):
self.mox.StubOutWithMock(db, 'fixed_ip_associate') self.mox.StubOutWithMock(db, 'fixed_ip_associate')
@ -272,7 +277,7 @@ class VlanNetworkTestCase(test.TestCase):
self.network.allocate_fixed_ip(None, 0, network, vpn=True) self.network.allocate_fixed_ip(None, 0, network, vpn=True)
def test_allocate_fixed_ip(self): def test_allocate_fixed_ip(self):
self.mox.StubOutWithMock(db, 'fixed_ip_associate_by_address') self.mox.StubOutWithMock(db, 'fixed_ip_associate_pool')
self.mox.StubOutWithMock(db, 'fixed_ip_update') self.mox.StubOutWithMock(db, 'fixed_ip_update')
self.mox.StubOutWithMock(db, self.mox.StubOutWithMock(db,
'virtual_interface_get_by_instance_and_network') 'virtual_interface_get_by_instance_and_network')
@ -281,8 +286,7 @@ class VlanNetworkTestCase(test.TestCase):
db.instance_get(mox.IgnoreArg(), db.instance_get(mox.IgnoreArg(),
mox.IgnoreArg()).AndReturn({'security_groups': mox.IgnoreArg()).AndReturn({'security_groups':
[{'id': 0}]}) [{'id': 0}]})
db.fixed_ip_associate_by_address(mox.IgnoreArg(), db.fixed_ip_associate_pool(mox.IgnoreArg(),
mox.IgnoreArg(),
mox.IgnoreArg(), mox.IgnoreArg(),
mox.IgnoreArg()).AndReturn('192.168.0.1') mox.IgnoreArg()).AndReturn('192.168.0.1')
db.fixed_ip_update(mox.IgnoreArg(), db.fixed_ip_update(mox.IgnoreArg(),
@ -294,7 +298,7 @@ class VlanNetworkTestCase(test.TestCase):
network = dict(networks[0]) network = dict(networks[0])
network['vpn_private_address'] = '192.168.0.2' network['vpn_private_address'] = '192.168.0.2'
self.network.allocate_fixed_ip(None, 0, network) self.network.allocate_fixed_ip(self.context, 0, network)
def test_create_networks_too_big(self): def test_create_networks_too_big(self):
self.assertRaises(ValueError, self.network.create_networks, None, self.assertRaises(ValueError, self.network.create_networks, None,
@ -306,13 +310,14 @@ class VlanNetworkTestCase(test.TestCase):
cidr='192.168.0.1/24', network_size=100) cidr='192.168.0.1/24', network_size=100)
def test_validate_networks(self): def test_validate_networks(self):
self.mox.StubOutWithMock(db, 'project_get_networks_by_uuids') self.mox.StubOutWithMock(db, 'network_get_all_by_uuids')
self.mox.StubOutWithMock(db, "fixed_ip_get_by_address") self.mox.StubOutWithMock(db, "fixed_ip_get_by_address")
requested_networks = [("bbbbbbbb-bbbb-bbbb-bbbb-bbbbbbbbbbbb", requested_networks = [("bbbbbbbb-bbbb-bbbb-bbbb-bbbbbbbbbbbb",
"192.168.1.100")] "192.168.1.100")]
db.project_get_networks_by_uuids(mox.IgnoreArg(), db.network_get_all_by_uuids(mox.IgnoreArg(),
mox.IgnoreArg()).AndReturn(networks) mox.IgnoreArg(),
mox.IgnoreArg()).AndReturn(networks)
fixed_ips[1]['network'] = FakeModel(**networks[1]) fixed_ips[1]['network'] = FakeModel(**networks[1])
fixed_ips[1]['instance'] = None fixed_ips[1]['instance'] = None
@ -320,49 +325,51 @@ class VlanNetworkTestCase(test.TestCase):
mox.IgnoreArg()).AndReturn(fixed_ips[1]) mox.IgnoreArg()).AndReturn(fixed_ips[1])
self.mox.ReplayAll() self.mox.ReplayAll()
self.network.validate_networks(None, requested_networks) self.network.validate_networks(self.context, requested_networks)
def test_validate_networks_none_requested_networks(self): def test_validate_networks_none_requested_networks(self):
self.network.validate_networks(None, None) self.network.validate_networks(self.context, None)
def test_validate_networks_empty_requested_networks(self): def test_validate_networks_empty_requested_networks(self):
requested_networks = [] requested_networks = []
self.mox.ReplayAll() self.mox.ReplayAll()
self.network.validate_networks(None, requested_networks) self.network.validate_networks(self.context, requested_networks)
def test_validate_networks_invalid_fixed_ip(self): def test_validate_networks_invalid_fixed_ip(self):
self.mox.StubOutWithMock(db, 'project_get_networks_by_uuids') self.mox.StubOutWithMock(db, 'network_get_all_by_uuids')
requested_networks = [(1, "192.168.0.100.1")] requested_networks = [(1, "192.168.0.100.1")]
db.project_get_networks_by_uuids(mox.IgnoreArg(), db.network_get_all_by_uuids(mox.IgnoreArg(),
mox.IgnoreArg()).AndReturn(networks) mox.IgnoreArg(),
mox.IgnoreArg()).AndReturn(networks)
self.mox.ReplayAll() self.mox.ReplayAll()
self.assertRaises(exception.FixedIpInvalid, self.assertRaises(exception.FixedIpInvalid,
self.network.validate_networks, None, self.network.validate_networks, self.context,
requested_networks) requested_networks)
def test_validate_networks_empty_fixed_ip(self): def test_validate_networks_empty_fixed_ip(self):
self.mox.StubOutWithMock(db, 'project_get_networks_by_uuids') self.mox.StubOutWithMock(db, 'network_get_all_by_uuids')
requested_networks = [(1, "")] requested_networks = [(1, "")]
db.project_get_networks_by_uuids(mox.IgnoreArg(), db.network_get_all_by_uuids(mox.IgnoreArg(),
mox.IgnoreArg()).AndReturn(networks) mox.IgnoreArg(),
mox.IgnoreArg()).AndReturn(networks)
self.mox.ReplayAll() self.mox.ReplayAll()
self.assertRaises(exception.FixedIpInvalid, self.assertRaises(exception.FixedIpInvalid,
self.network.validate_networks, self.network.validate_networks,
None, requested_networks) self.context, requested_networks)
def test_validate_networks_none_fixed_ip(self): def test_validate_networks_none_fixed_ip(self):
self.mox.StubOutWithMock(db, 'project_get_networks_by_uuids') self.mox.StubOutWithMock(db, 'network_get_all_by_uuids')
requested_networks = [(1, None)] requested_networks = [(1, None)]
db.project_get_networks_by_uuids(mox.IgnoreArg(), db.network_get_all_by_uuids(mox.IgnoreArg(),
mox.IgnoreArg()).AndReturn(networks) mox.IgnoreArg(),
mox.IgnoreArg()).AndReturn(networks)
self.mox.ReplayAll() self.mox.ReplayAll()
self.network.validate_networks(self.context, requested_networks)
self.network.validate_networks(None, requested_networks)
class CommonNetworkTestCase(test.TestCase): class CommonNetworkTestCase(test.TestCase):