openstack-attic/melange
Code Issues Proposed changes

Rajaram/Vinkesh | Removed some unused code. Added few tests. Miscellaneous small refactoring

Browse Source
This commit is contained in:
Rajaram Mallya
2011-09-14 15:55:46 +05:30
parent 2197867025
commit b4cc6224a7
10 changed files with 102 additions and 114 deletions
Download Patch File Download Diff File Expand all files Collapse all files

22
melange/common/client.py
View File

@@ -19,6 +19,8 @@ import httplib
import socket
import urllib
from melange.common import exception
class HTTPClient(object):
@@ -28,19 +30,6 @@ class HTTPClient(object):
self.use_ssl = use_ssl
self.timeout = timeout
def get(self, path, params=None, headers=None):
params = params or {}
headers = headers or {}
return self.do_request("GET", path, params=params, headers=headers)
def post(self, path, body=None, headers=None):
headers = headers or {}
return self.do_request("POST", path, body=body, headers=headers)
def delete(self, path, headers=None):
headers = headers or {}
return self.do_request("DELETE", path, headers=headers)
def _get_connection(self):
if self.use_ssl:
return httplib.HTTPSConnection(self.host, self.port,
@@ -59,7 +48,10 @@ class HTTPClient(object):
connection = self._get_connection()
connection.request(method, url, body, headers)
response = connection.getresponse()
if response.status >= 400:
raise exception.MelangeServiceResponseError(response.read())
return response
except (socket.error, IOError) as error:
raise Exception(_("Error while communicating with server. "
"Got error: %s") % error)
raise exception.ClientConnectionError(
_("Error while communicating with server. "
"Got error: %s") % error)

6
melange/common/exception.py
View File

@@ -18,6 +18,7 @@
from openstack.common import exception as openstack_exception
ClientConnectionError = openstack_exception.ClientConnectionError
ProcessExecutionError = openstack_exception.ProcessExecutionError
DatabaseMigrationError = openstack_exception.DatabaseMigrationError
@@ -29,3 +30,8 @@ class MelangeError(openstack_exception.OpenstackException):
if message is not None:
self.message = message
super(MelangeError, self).__init__(**kwargs)
class MelangeServiceResponseError(MelangeError):
message = "Error while responding to service call"

11
melange/common/pagination.py
View File

@@ -22,23 +22,14 @@ from xml.dom import minidom
class AtomLink(object):
def __init__(self, rel, href, link_type=None, hreflang=None, title=None):
def __init__(self, rel, href):
self.rel = rel
self.href = href
self.link_type = link_type
self.hreflang = hreflang
self.title = title
def to_xml(self):
ATOM_NAMESPACE = "http://www.w3.org/2005/Atom"
doc = minidom.Document()
atom_elem = doc.createElementNS(ATOM_NAMESPACE, "link")
if self.link_type:
atom_elem.setAttribute("link_type", self.link_type)
if self.hreflang:
atom_elem.setAttribute("hreflang", self.hreflang)
if self.title:
atom_elem.setAttribute("title", self.title)
atom_elem.setAttribute("rel", self.rel)
atom_elem.setAttribute("href", self.href)
return atom_elem

39
melange/ipam/models.py
View File

@@ -187,10 +187,6 @@ class ModelBase(object):
def __getitem__(self, key):
return getattr(self, key)
def __iter__(self):
self._i = iter(db_api.columns_of(self))
return self
def __eq__(self, other):
if not hasattr(other, 'id'):
return False
@@ -200,23 +196,7 @@ class ModelBase(object):
return not self == other
def __hash__(self):
return id.__hash__()
def next(self):
n = self._i.next().name
return n, getattr(self, n)
def keys(self):
return self.__dict__.keys()
def values(self):
return self.__dict__.values()
def items(self):
return self.__dict__.items()
def to_dict(self):
return self.__dict__()
return self.id.__hash__()
def data(self, **options):
data_fields = self._data_fields + self._auto_generated_attrs
@@ -233,9 +213,6 @@ class ModelBase(object):
self.errors[attribute_name] = self.errors.get(attribute_name, [])
self.errors[attribute_name].append(error_message)
def _has_error_on(self, attribute):
return self.errors.get(attribute, None) is not None
def ipv6_address_generator_factory(cidr, **kwargs):
default_generator = "melange.ipv6.tenant_based_generator."\
@@ -282,10 +259,6 @@ class IpBlock(ModelBase):
return (allocated_ip or block.allocate_ip(address=address))
@classmethod
def find_all_by_policy(cls, policy_id):
return cls.find_all(policy_id=policy_id)
@classmethod
def allowed_by_policy(cls, ip_block, policy, address):
return policy == None or policy.allows(ip_block.cidr, address)
@@ -682,10 +655,6 @@ class IpOctet(ModelBase):
_columns = {'octet': 'integer'}
_data_fields = ['octet', 'policy_id']
@classmethod
def find_all_by_policy(cls, policy_id):
return cls.find_all(policy_id=policy_id)
def applies_to(self, address):
return self.octet == netaddr.IPAddress(address).words[-1]
@@ -710,6 +679,12 @@ class Network(ModelBase):
type="private")
return cls(id=id, ip_blocks=[ip_block])
def allocated_ips(self, interface_id):
ips_by_block = [IpAddress.find_all(interface_id=interface_id,
ip_block_id=ip_block.id).all()
for ip_block in self.ip_blocks]
return [ip for sublist in ips_by_block for ip in sublist]
def allocate_ips(self, addresses=None, **kwargs):
if addresses:
return filter(None, [self._allocate_specific_ip(address, **kwargs)

12
melange/ipam/service.py
View File

@@ -319,14 +319,10 @@ class NetworksController(BaseController):
network.deallocate_ips(interface_id)
def get_allocated_ips(self, request, network_id, interface_id, tenant_id):
ip_blocks = models.IpBlock.find_all(network_id=network_id,
tenant_id=tenant_id)
addresses = [models.IpAddress.find_all(interface_id=interface_id,
ip_block_id=ip_block.id)
for ip_block in ip_blocks]
return dict(ip_addresses=[item.data(with_ip_block=True)
for sublist in addresses
for item in sublist])
network = models.Network.find_by(id=network_id, tenant_id=tenant_id)
ips_on_interface = network.allocated_ips(interface_id=interface_id)
return dict(ip_addresses=[ip.data(with_ip_block=True)
for ip in ips_on_interface])
class API(wsgi.Router):

4
melange/tests/functional/__init__.py
View File

@@ -33,10 +33,10 @@ def setup():
print "Restarting melange server..."
shutil.copyfile(melange.melange_etc_path("melange.conf.sample"),
os.path.expanduser("~/melange.conf"))
svr = server.Server("melange",
srv = server.Server("melange",
melange.melange_bin_path('melange'))
_db_sync()
svr.restart(port=setup_unused_port())
srv.restart(port=setup_unused_port())
_configure_db()

25
melange/tests/functional/test_service.py
View File

@@ -17,6 +17,7 @@
from melange import tests
from melange.common import client
from melange.common import exception
from melange.tests import functional
@@ -26,22 +27,28 @@ class FunctionalTest(tests.BaseTest):
super(FunctionalTest, self).setUp()
self.client = client.HTTPClient(port=functional.get_api_port())
def client_get(self, path, params=None, headers=None):
params = params or {}
headers = headers or {}
return self.client.do_request("GET", path, params=params,
headers=headers)
class TestServiceConf(FunctionalTest):
def test_root_url_returns_versions(self):
response = self.client.get("/")
response = self.client_get("/")
self.assertEqual(response.status, 200)
self.assertTrue("versions" in response.read())
def test_extensions_are_loaded(self):
response = self.client.get("/v0.1/extensions")
response = self.client_get("/v0.1/extensions")
self.assertEqual(response.status, 200)
self.assertTrue("extensions" in response.read())
def test_ipam_service_can_be_accessed(self):
response = self.client.get("/v0.1/ipam/tenants/123/ip_blocks")
response = self.client_get("/v0.1/ipam/tenants/123/ip_blocks")
self.assertEqual(response.status, 200)
self.assertTrue("ip_blocks" in response.read())
@@ -55,7 +62,7 @@ class TestMimeTypeVersioning(FunctionalTest):
"version=0.1",
}
response = self.client.get("/ipam/tenants/123/ip_blocks",
response = self.client_get("/ipam/tenants/123/ip_blocks",
headers=headers)
self.assertEqual(response.status, 200)
@@ -69,8 +76,8 @@ class TestMimeTypeVersioning(FunctionalTest):
"version=99.1",
}
response = self.client.get("/ipam/tenants/123/ip_blocks",
headers=headers)
self.assertEqual(response.status, 406)
self.assertTrue("version not supported" in response.read())
self.assertRaisesExcMessage(exception.MelangeServiceResponseError,
"version not supported",
self.client_get,
"/ipam/tenants/123/ip_blocks",
headers=headers)

48
melange/tests/unit/test_auth.py
View File

@@ -52,31 +52,45 @@ class TestAuthMiddleware(tests.BaseTest):
self.request.headers = {'X_TENANT': "tenant_id", 'X_ROLE': "Member"}
def test_forbids_based_on_auth_providers(self):
self.auth_provider1.authorize(self.request, "tenant_id", ['Member']).\
AndReturn(True)
self.auth_provider1.authorize(self.request,
"tenant_id",
['Member']).AndReturn(True)
self.auth_provider2.authorize(self.request, "tenant_id", ['Member']).\
AndRaise(webob.exc.HTTPForbidden("Auth Failed"))
self.mock.ReplayAll()
self.assertRaisesExcMessage(webob.exc.HTTPForbidden, "Auth Failed",
self.auth_middleware, self.request)
self.assertRaisesExcMessage(webob.exc.HTTPForbidden,
"Auth Failed",
self.auth_middleware,
self.request)
def test_authorizes_based_on_auth_providers(self):
self.auth_provider1.authorize(self.request, "tenant_id", ['Member']).\
AndReturn(True)
self.auth_provider2.authorize(self.request, "tenant_id", ['Member']).\
AndReturn(True)
self.auth_provider1.authorize(self.request,
"tenant_id",
['Member']).AndReturn(True)
self.auth_provider2.authorize(self.request,
"tenant_id",
['Member']).AndReturn(True)
self.mock.ReplayAll()
response = self.auth_middleware(self.request)
self.assertEqual(response.status_int, 200)
def test_factory_adds_tenant_based_auth_as_one_of_auth_providers(self):
factory = auth.AuthorizationMiddleware.factory({})
class DecoratorTestApp(wsgi.Router):
self.mock.StubOutWithMock(auth, 'TenantBasedAuth')
tenant_auth = self.mock.CreateMockAnything()
tenant_auth.authorize(self.request,
"tenant_id",
['Member']).AndReturn(True)
def __init__(self):
super(DecoratorTestApp, self).__init__(mapper())
auth.TenantBasedAuth().AndReturn(tenant_auth)
self.mock.ReplayAll()
auth_middleware = factory(self.dummy_app)
self.assertTrue(auth_middleware(self.request))
def mapper():
@@ -89,9 +103,6 @@ def mapper():
class StubController(wsgi.Controller):
def admin_action(self, request):
pass
def unrestricted(self, request):
pass
@@ -172,7 +183,8 @@ class TestKeyStoneClient(tests.BaseTest):
response_body = json.dumps({'auth': {'token': {'id': "auth_token"}}})
res = httplib2.Response(dict(status='200'))
client.request(urlparse.urljoin(url, "/v2.0/tokens"), "POST",
client.request(urlparse.urljoin(url, "/v2.0/tokens"),
"POST",
headers=IgnoreArg(),
body=request_body).AndReturn((res, response_body))
@@ -185,12 +197,14 @@ class TestKeyStoneClient(tests.BaseTest):
self.mock.StubOutWithMock(client, "request")
res = httplib2.Response(dict(status='401'))
response_body = "Failed to get token"
client.request(urlparse.urljoin(url, "/v2.0/tokens"), "POST",
client.request(urlparse.urljoin(url, "/v2.0/tokens"),
"POST",
headers=IgnoreArg(),
body=IgnoreArg()).AndReturn((res, response_body))
self.mock.ReplayAll()
expected_error_msg = ("Error occured while retrieving token :"
" Failed to get token")
self.assertRaisesExcMessage(Exception, expected_error_msg,
self.assertRaisesExcMessage(Exception,
expected_error_msg,
client.get_token)

46
melange/tests/unit/test_ipam_models.py
View File

@@ -81,16 +81,22 @@ class TestModelBase(tests.BaseTest):
self.assertEqual(model.foo, True)
def test_equals(self):
def test_equals_is_true_when_ids_and_class_are_equal(self):
self.assertEqual(models.ModelBase(id=1), models.ModelBase(id=1))
self.assertEqual(models.ModelBase(id=1, name="foo"),
models.ModelBase(id=1, name="bar"))
def test_not_equals(self):
def test_equals_is_false_when_id_or_class_differ(self):
self.assertNotEqual(models.ModelBase(), models.ModelBase())
self.assertNotEqual(models.ModelBase(id=1), models.ModelBase(id=2))
self.assertNotEqual(models.IpBlock(id=1), models.IpAddress(id=1))
def test_hash_is_correct(self):
a = models.ModelBase(id="123", name="foo")
b = models.ModelBase(id="123", name="bar")
self.assertEqual(hash(a), hash(b))
class TestQuery(tests.BaseTest):
@@ -492,9 +498,10 @@ class TestIpBlock(tests.BaseTest):
def test_find_allocated_ip_for_nonexistent_address(self):
block = factory_models.PrivateIpBlockFactory(cidr="10.0.0.1/8")
self.assertRaises(models.ModelNotFoundError,
block.find_allocated_ip,
'10.0.0.1')
self.assertRaisesExcMessage(models.ModelNotFoundError,
"IpAddress Not Found",
block.find_allocated_ip,
"10.0.0.1")
def test_policy(self):
policy = factory_models.PolicyFactory(name="Some Policy")
@@ -1388,19 +1395,6 @@ class TestIpOctet(tests.BaseTest):
self.assertEqual(data['created_at'], ip_octet.created_at)
self.assertEqual(data['updated_at'], ip_octet.updated_at)
def test_find_all_by_policy(self):
policy1 = factory_models.PolicyFactory(name='blah')
policy2 = factory_models.PolicyFactory(name='blah')
ip_octet1 = factory_models.IpOctetFactory(octet=123,
policy_id=policy1.id)
ip_octet2 = factory_models.IpOctetFactory(octet=123,
policy_id=policy1.id)
noise_ip_octet = factory_models.IpOctetFactory(octet=123,
policy_id=policy2.id)
actual_octets = models.IpOctet.find_all_by_policy(policy1.id).all()
self.assertModelsEqual(actual_octets, [ip_octet1, ip_octet2])
def test_applies_to_is_true_if_address_last_octet_matches(self):
ip_octet = factory_models.IpOctetFactory(octet=123)
self.assertTrue(ip_octet.applies_to("10.0.0.123"))
@@ -1532,3 +1526,19 @@ class TestNetwork(tests.BaseTest):
for ip in ips:
ip_address = models.IpAddress.get(ip.id)
self.assertTrue(ip_address.marked_for_deallocation)
def test_retrives_allocated_ips(self):
ip_block1 = factory_models.IpBlockFactory(network_id=1,
cidr="10.0.0.0/24")
ip_block2 = factory_models.IpBlockFactory(network_id=1,
cidr="20.0.0.0/24")
ip1 = ip_block1.allocate_ip(interface_id="123")
ip2 = ip_block1.allocate_ip(interface_id="123")
ip3 = ip_block2.allocate_ip(interface_id="321")
ip4 = ip_block2.allocate_ip(interface_id="123")
network = models.Network.find_by(id=1)
self.assertModelsEqual(network.allocated_ips(interface_id="123"),
[ip1, ip2, ip4])

3
melange/tests/unit/test_wsgi.py
View File

@@ -237,9 +237,6 @@ class StubController(wsgi.Controller):
def index(self, request, format=None):
return {'fort': 'knox'}
def show(self, request, id, format=None):
return {'fort': 'knox'}
class TestController(tests.BaseTest):