x/python-ganttclient
Code Issues Proposed changes

merge with trunk

Browse Source
This commit is contained in:
Isaku Yamahata
2011-06-27 20:20:42 +09:00
parent c48a599225 80b3eee8f7
commit b182690f21
6 changed files with 224 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files

13
bin/nova-manage
View File

@@ -56,11 +56,11 @@
import gettext
import glob
import json
import netaddr
import os
import sys
import time
import IPy
# If ../nova/__init__.py exists, add ../ to Python search path, so that
# it will override what happens to be installed in /usr/(local/)lib/python...
@@ -257,6 +257,11 @@ class RoleCommands(object):
"""adds role to user
if project is specified, adds project specific role
arguments: user, role [project]"""
if project:
projobj = self.manager.get_project(project)
if not projobj.has_member(user):
print "%s not a member of %s" % (user, project)
return
self.manager.add_role(user, role, project)
def has(self, user, role, project=None):
@@ -513,7 +518,7 @@ class FloatingIpCommands(object):
def create(self, host, range):
"""Creates floating ips for host by range
arguments: host ip_range"""
for address in IPy.IP(range):
for address in netaddr.IPNetwork(range):
db.floating_ip_create(context.get_admin_context(),
{'address': str(address),
'host': host})
@@ -521,7 +526,7 @@ class FloatingIpCommands(object):
def delete(self, ip_range):
"""Deletes floating ips by range
arguments: range"""
for address in IPy.IP(ip_range):
for address in netaddr.IPNetwork(ip_range):
db.floating_ip_destroy(context.get_admin_context(),
str(address))
@@ -873,7 +878,7 @@ class InstanceTypeCommands(object):
try:
instance_types.create(name, memory, vcpus, local_gb,
flavorid, swap, rxtx_quota, rxtx_cap)
except exception.InvalidInputException:
except exception.InvalidInput:
print "Must supply valid parameters to create instance_type"
print e
sys.exit(1)

111
nova/tests/test_adminapi.py Normal file
View File

@@ -0,0 +1,111 @@
# vim: tabstop=4 shiftwidth=4 softtabstop=4
# Copyright 2010 United States Government as represented by the
# Administrator of the National Aeronautics and Space Administration.
# 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 eventlet import greenthread
from nova import context
from nova import db
from nova import flags
from nova import log as logging
from nova import rpc
from nova import test
from nova import utils
from nova.auth import manager
from nova.api.ec2 import admin
from nova.image import fake
FLAGS = flags.FLAGS
LOG = logging.getLogger('nova.tests.adminapi')
class AdminApiTestCase(test.TestCase):
def setUp(self):
super(AdminApiTestCase, self).setUp()
self.flags(connection_type='fake')
self.conn = rpc.Connection.instance()
# set up our cloud
self.api = admin.AdminController()
# set up services
self.compute = self.start_service('compute')
self.scheduter = self.start_service('scheduler')
self.network = self.start_service('network')
self.volume = self.start_service('volume')
self.image_service = utils.import_object(FLAGS.image_service)
self.manager = manager.AuthManager()
self.user = self.manager.create_user('admin', 'admin', 'admin', True)
self.project = self.manager.create_project('proj', 'admin', 'proj')
self.context = context.RequestContext(user=self.user,
project=self.project)
host = self.network.get_network_host(self.context.elevated())
def fake_show(meh, context, id):
return {'id': 1, 'properties': {'kernel_id': 1, 'ramdisk_id': 1,
'type': 'machine', 'image_state': 'available'}}
self.stubs.Set(fake._FakeImageService, 'show', fake_show)
self.stubs.Set(fake._FakeImageService, 'show_by_name', fake_show)
# NOTE(vish): set up a manual wait so rpc.cast has a chance to finish
rpc_cast = rpc.cast
def finish_cast(*args, **kwargs):
rpc_cast(*args, **kwargs)
greenthread.sleep(0.2)
self.stubs.Set(rpc, 'cast', finish_cast)
def tearDown(self):
network_ref = db.project_get_network(self.context,
self.project.id)
db.network_disassociate(self.context, network_ref['id'])
self.manager.delete_project(self.project)
self.manager.delete_user(self.user)
super(AdminApiTestCase, self).tearDown()
def test_block_external_ips(self):
"""Make sure provider firewall rules are created."""
result = self.api.block_external_addresses(self.context, '1.1.1.1/32')
self.api.remove_external_address_block(self.context, '1.1.1.1/32')
self.assertEqual('OK', result['status'])
self.assertEqual('Added 3 rules', result['message'])
def test_list_blocked_ips(self):
"""Make sure we can see the external blocks that exist."""
self.api.block_external_addresses(self.context, '1.1.1.2/32')
result = self.api.describe_external_address_blocks(self.context)
num = len(db.provider_fw_rule_get_all(self.context))
self.api.remove_external_address_block(self.context, '1.1.1.2/32')
# we only list IP, not tcp/udp/icmp rules
self.assertEqual(num / 3, len(result['externalIpBlockInfo']))
def test_remove_ip_block(self):
"""Remove ip blocks."""
result = self.api.block_external_addresses(self.context, '1.1.1.3/32')
self.assertEqual('OK', result['status'])
num0 = len(db.provider_fw_rule_get_all(self.context))
result = self.api.remove_external_address_block(self.context,
'1.1.1.3/32')
self.assertEqual('OK', result['status'])
self.assertEqual('Deleted 3 rules', result['message'])
num1 = len(db.provider_fw_rule_get_all(self.context))
self.assert_(num1 < num0)

6
nova/tests/test_flat_network.py
View File

@@ -18,7 +18,7 @@
"""
Unit Tests for flat network code
"""
import IPy
import netaddr
import os
import unittest
@@ -45,8 +45,8 @@ class FlatNetworkTestCase(base.NetworkTestCase):
self.context._project = self.projects[0]
self.context.project_id = self.projects[0].id
pubnet = IPy.IP(flags.FLAGS.floating_range)
address = str(pubnet[0])
pubnet = netaddr.IPRange(flags.FLAGS.floating_range)
address = str(list(pubnet)[0])
try:
db.floating_ip_get_by_address(context.get_admin_context(), address)
except exception.NotFound:

67
nova/tests/test_libvirt.py
View File

@@ -799,6 +799,8 @@ class IptablesFirewallTestCase(test.TestCase):
self.network = utils.import_object(FLAGS.network_manager)
class FakeLibvirtConnection(object):
def nwfilterDefineXML(*args, **kwargs):
"""setup_basic_rules in nwfilter calls this."""
pass
self.fake_libvirt_connection = FakeLibvirtConnection()
self.fw = firewall.IptablesFirewallDriver(
@@ -1035,7 +1037,6 @@ class IptablesFirewallTestCase(test.TestCase):
fakefilter.filterDefineXMLMock
self.fw.nwfilter._conn.nwfilterLookupByName =\
fakefilter.nwfilterLookupByName
instance_ref = self._create_instance_ref()
inst_id = instance_ref['id']
instance = db.instance_get(self.context, inst_id)
@@ -1057,6 +1058,70 @@ class IptablesFirewallTestCase(test.TestCase):
db.instance_destroy(admin_ctxt, instance_ref['id'])
def test_provider_firewall_rules(self):
# setup basic instance data
instance_ref = self._create_instance_ref()
nw_info = _create_network_info(1)
ip = '10.11.12.13'
network_ref = db.project_get_network(self.context, 'fake')
admin_ctxt = context.get_admin_context()
fixed_ip = {'address': ip, 'network_id': network_ref['id']}
db.fixed_ip_create(admin_ctxt, fixed_ip)
db.fixed_ip_update(admin_ctxt, ip, {'allocated': True,
'instance_id': instance_ref['id']})
# FRAGILE: peeks at how the firewall names chains
chain_name = 'inst-%s' % instance_ref['id']
# create a firewall via setup_basic_filtering like libvirt_conn.spawn
# should have a chain with 0 rules
self.fw.setup_basic_filtering(instance_ref, network_info=nw_info)
self.assertTrue('provider' in self.fw.iptables.ipv4['filter'].chains)
rules = [rule for rule in self.fw.iptables.ipv4['filter'].rules
if rule.chain == 'provider']
self.assertEqual(0, len(rules))
# add a rule and send the update message, check for 1 rule
provider_fw0 = db.provider_fw_rule_create(admin_ctxt,
{'protocol': 'tcp',
'cidr': '10.99.99.99/32',
'from_port': 1,
'to_port': 65535})
self.fw.refresh_provider_fw_rules()
rules = [rule for rule in self.fw.iptables.ipv4['filter'].rules
if rule.chain == 'provider']
self.assertEqual(1, len(rules))
# Add another, refresh, and make sure number of rules goes to two
provider_fw1 = db.provider_fw_rule_create(admin_ctxt,
{'protocol': 'udp',
'cidr': '10.99.99.99/32',
'from_port': 1,
'to_port': 65535})
self.fw.refresh_provider_fw_rules()
rules = [rule for rule in self.fw.iptables.ipv4['filter'].rules
if rule.chain == 'provider']
self.assertEqual(2, len(rules))
# create the instance filter and make sure it has a jump rule
self.fw.prepare_instance_filter(instance_ref, network_info=nw_info)
self.fw.apply_instance_filter(instance_ref)
inst_rules = [rule for rule in self.fw.iptables.ipv4['filter'].rules
if rule.chain == chain_name]
jump_rules = [rule for rule in inst_rules if '-j' in rule.rule]
provjump_rules = []
# IptablesTable doesn't make rules unique internally
for rule in jump_rules:
if 'provider' in rule.rule and rule not in provjump_rules:
provjump_rules.append(rule)
self.assertEqual(1, len(provjump_rules))
# remove a rule from the db, cast to compute to refresh rule
db.provider_fw_rule_destroy(admin_ctxt, provider_fw1['id'])
self.fw.refresh_provider_fw_rules()
rules = [rule for rule in self.fw.iptables.ipv4['filter'].rules
if rule.chain == 'provider']
self.assertEqual(1, len(rules))
class NWFilterTestCase(test.TestCase):
def setUp(self):

32
nova/tests/test_network.py
View File

@@ -18,7 +18,7 @@
"""
Unit Tests for network code
"""
import IPy
import netaddr
import os
from nova import test
@@ -164,3 +164,33 @@ class IptablesManagerTestCase(test.TestCase):
self.assertTrue('-A %s -j run_tests.py-%s' \
% (chain, chain) in new_lines,
"Built-in chain %s not wrapped" % (chain,))
def test_will_empty_chain(self):
self.manager.ipv4['filter'].add_chain('test-chain')
self.manager.ipv4['filter'].add_rule('test-chain', '-j DROP')
old_count = len(self.manager.ipv4['filter'].rules)
self.manager.ipv4['filter'].empty_chain('test-chain')
self.assertEqual(old_count - 1, len(self.manager.ipv4['filter'].rules))
def test_will_empty_unwrapped_chain(self):
self.manager.ipv4['filter'].add_chain('test-chain', wrap=False)
self.manager.ipv4['filter'].add_rule('test-chain', '-j DROP',
wrap=False)
old_count = len(self.manager.ipv4['filter'].rules)
self.manager.ipv4['filter'].empty_chain('test-chain', wrap=False)
self.assertEqual(old_count - 1, len(self.manager.ipv4['filter'].rules))
def test_will_not_empty_wrapped_when_unwrapped(self):
self.manager.ipv4['filter'].add_chain('test-chain')
self.manager.ipv4['filter'].add_rule('test-chain', '-j DROP')
old_count = len(self.manager.ipv4['filter'].rules)
self.manager.ipv4['filter'].empty_chain('test-chain', wrap=False)
self.assertEqual(old_count, len(self.manager.ipv4['filter'].rules))
def test_will_not_empty_unwrapped_when_wrapped(self):
self.manager.ipv4['filter'].add_chain('test-chain', wrap=False)
self.manager.ipv4['filter'].add_rule('test-chain', '-j DROP',
wrap=False)
old_count = len(self.manager.ipv4['filter'].rules)
self.manager.ipv4['filter'].empty_chain('test-chain')
self.assertEqual(old_count, len(self.manager.ipv4['filter'].rules))

6
nova/tests/test_vlan_network.py
View File

@@ -18,7 +18,7 @@
"""
Unit Tests for vlan network code
"""
import IPy
import netaddr
import os
from nova import context
@@ -44,8 +44,8 @@ class VlanNetworkTestCase(base.NetworkTestCase):
# TODO(vish): better way of adding floating ips
self.context._project = self.projects[0]
self.context.project_id = self.projects[0].id
pubnet = IPy.IP(flags.FLAGS.floating_range)
address = str(pubnet[0])
pubnet = netaddr.IPNetwork(flags.FLAGS.floating_range)
address = str(list(pubnet)[0])
try:
db.floating_ip_get_by_address(context.get_admin_context(), address)
except exception.NotFound: