openstack-attic
/
melange
Code Issues Proposed changes

New APIs as per blueprint interface-api-cleanup 

Also defines the 1.0 API that doesnt contain the old apis of 0.1 that we plan
to remove

Change-Id: I3aed4fcfefc07af7026f94ae62d4e842d6145007
This commit is contained in:
rajarammallya 2012-02-10 18:51:21 +05:30
parent ddca783309
commit fe4b075c15
6 changed files with 434 additions and 109 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

17
etc/melange/melange.conf.sample
View File

@ -63,13 +63,17 @@ notifier_queue_transport = memory
[composite:melange]
use = call:melange.common.wsgi:versioned_urlmap
/: versions
/v0.1: melangeapi
/v0.1: melangeapp_v0_1
/v1.0: melangeapp_v1_0
[app:versions]
paste.app_factory = melange.versions:app_factory
[pipeline:melangeapi]
pipeline = extensions melangeapp
[pipeline:melangeapi_v0_1]
pipeline = extensions melangeapp_v0_1
[pipeline:melangeapi_v1_0]
pipeline = extensions melangeapp_v1_0
[filter:extensions]
paste.filter_factory = melange.common.extensions:factory
@ -87,8 +91,11 @@ admin_token = 999888777666
[filter:authorization]
paste.filter_factory = melange.common.auth:AuthorizationMiddleware.factory
[app:melangeapp]
paste.app_factory = melange.ipam.service:app_factory
[app:melangeapp_v0_1]
paste.app_factory = melange.ipam.service:APIV01.app_factory
[app:melangeapp_v1_0]
paste.app_factory = melange.ipam.service:APIV10.app_factory
#Add this filter to log request and response for debugging
[filter:debug]

14
melange/ipam/models.py
View File

@ -810,6 +810,20 @@ class Interface(ModelBase):
tenant_id,
mac_address)
@classmethod
def create_and_allocate_ips(cls,
device_id=None,
network_params=None,
**kwargs):
interface = Interface.create_and_configure(device_id=device_id,
**kwargs)
if network_params:
network = Network.find_or_create_by(network_params.pop('id'),
network_params.pop('tenant_id'))
network.allocate_ips(interface=interface, **network_params)
return interface
@classmethod
def create_and_configure(cls, virtual_interface_id=None, device_id=None,
tenant_id=None, mac_address=None):

282
melange/ipam/service.py
View File

@ -439,7 +439,7 @@ class InterfacesController(BaseController, ShowAction, DeleteAction):
class InstanceInterfacesController(BaseController):
def update(self, request, device_id, body=None):
def update_all(self, request, device_id, body=None):
models.Interface.delete_by(device_id=device_id)
params = self._extract_required_params(body, 'instance')
@ -448,31 +448,77 @@ class InstanceInterfacesController(BaseController):
for iface in params['interfaces']:
network_params = utils.stringify_keys(iface.pop('network', None))
interface = models.Interface.create_and_configure(
device_id=device_id, tenant_id=tenant_id, **iface)
if network_params:
network = models.Network.find_or_create_by(
network_params.pop('id'),
network_params.pop('tenant_id'))
network.allocate_ips(interface=interface, **network_params)
interface = models.Interface.create_and_allocate_ips(
device_id=device_id,
network_params=network_params,
tenant_id=tenant_id,
**iface)
view_data = views.InterfaceConfigurationView(interface).data()
created_interfaces.append(view_data)
return {'instance': {'interfaces': created_interfaces}}
def show(self, request, device_id):
def index(self, request, device_id):
interfaces = models.Interface.find_all(device_id=device_id)
view_data = [views.InterfaceConfigurationView(iface).data()
for iface in interfaces]
return {'instance': {'interfaces': view_data}}
def delete(self, request, device_id):
def delete_all(self, request, device_id):
LOG.debug("Deleting instance interface (device_id=%s)" % device_id)
models.Interface.delete_by(device_id=device_id)
def create(self, request, device_id, body=None):
iface_params = self._extract_required_params(body, 'interface')
network_params = utils.stringify_keys(iface_params.pop('network',
None))
interface = models.Interface.create_and_allocate_ips(
device_id=device_id,
network_params=network_params,
**iface_params)
view_data = views.InterfaceConfigurationView(interface).data()
return dict(interface=view_data)
def show(self, request, id, device_id, tenant_id=None):
iface_params = dict(device_id=device_id, id=id)
if tenant_id:
iface_params.update(dict(tenant_id=tenant_id))
interface = models.Interface.find_by(**iface_params)
view_data = views.InterfaceConfigurationView(interface).data()
return dict(interface=view_data)
def delete(self, request, id, device_id):
interface = models.Interface.find_by(id=id, device_id=device_id)
interface.delete()
class InstanceInterfaceIpsController(BaseController):
def create(self, request, body, device_id, interface_id):
params = self._extract_required_params(body, 'network')
interface = models.Interface.find_by(id=interface_id,
device_id=device_id)
network = models.Network.find_by(params.pop('id'),
tenant_id=params.pop('tenant_id'))
ips = network.allocate_ips(interface=interface, **params)
ip_config_view = views.IpConfigurationView(*ips)
return wsgi.Result(dict(ip_addresses=ip_config_view.data()), 201)
def delete(self, request, device_id, interface_id, address):
interface = models.Interface.find_by(device_id=device_id,
id=interface_id)
network_id = interface.plugged_in_network_id()
if not network_id:
raise models.ModelNotFoundError(_("IpAddress Not Found"))
network = models.Network.find_by(network_id)
ip = network.find_allocated_ip(interface_id=interface.id,
address=address)
ip.deallocate()
class MacAddressRangesController(BaseController, ShowAction, DeleteAction):
@ -522,63 +568,54 @@ class InterfaceAllowedIpsController(BaseController):
interface.disallow_ip(ip)
class API(wsgi.Router):
class APIV01(wsgi.Router):
def __init__(self):
mapper = routes.Mapper()
super(API, self).__init__(mapper)
self._natting_mapper(mapper,
"inside_globals",
InsideGlobalsController().create_resource())
self._natting_mapper(mapper,
"inside_locals",
InsideLocalsController().create_resource())
self._block_and_nested_resource_mapper(mapper)
self._policy_and_rules_mapper(mapper)
super(APIV01, self).__init__(mapper)
self._networks_maper(mapper)
self._interface_ip_allocations_mapper(mapper)
self._allocated_ips_mapper(mapper)
self._ip_routes_mapper(mapper)
self._interface_mapper(mapper)
self._instance_interface_mapper(mapper)
self._mac_address_range_mapper(mapper)
self._allowed_ips_mapper(mapper)
APICommon(mapper)
def _allocated_ips_mapper(self, mapper):
allocated_ips_res = AllocatedIpAddressesController().create_resource()
self._connect(mapper,
"/ipam/allocated_ip_addresses",
controller=allocated_ips_res,
action="index",
conditions=dict(method=['GET']))
self._connect(mapper,
"/ipam/tenants/{tenant_id}/allocated_ip_addresses",
controller=allocated_ips_res,
action="index",
conditions=dict(method=['GET']))
def _networks_maper(self, mapper):
resource = NetworksController().create_resource()
path = "/ipam/tenants/{tenant_id}/networks/{network_id}"
mapper.resource("networks", path, controller=resource)
def _ip_routes_mapper(self, mapper):
ip_routes_res = IpRoutesController().create_resource()
path = ("/ipam/tenants/{tenant_id}/ip_blocks/{source_block_id}"
"/ip_routes")
mapper.resource("ip_routes", path, controller=ip_routes_res)
def _interface_ip_allocations_mapper(self, mapper):
path = ("/ipam/tenants/{tenant_id}/networks"
"/{network_id}/interfaces/{interface_id}")
resource = InterfaceIpAllocationsController().create_resource()
with mapper.submapper(controller=resource, path_prefix=path) as submap:
_connect(submap, "/ip_allocations", action='create',
conditions=dict(method=['POST']))
_connect(submap,
"/ip_allocations",
action='index',
conditions=dict(method=['GET']))
_connect(submap, "/ip_allocations", action='bulk_delete',
conditions=dict(method=['DELETE']))
def _interface_mapper(self, mapper):
interface_res = InterfacesController().create_resource()
interface_allowed_ips = InterfaceAllowedIpsController()
path = "/ipam/interfaces"
self._connect(mapper,
_connect(mapper,
"/ipam/tenants/{tenant_id}/"
"interfaces/{virtual_interface_id}",
controller=interface_res,
action="show",
conditions=dict(method=['GET']))
self._connect(mapper,
_connect(mapper,
"/ipam/interfaces/{virtual_interface_id}",
controller=interface_res,
action="delete",
conditions=dict(method=['DELETE']))
mapper.resource("interfaces", path, controller=interface_res)
def _allowed_ips_mapper(self, mapper):
interface_allowed_ips = InterfaceAllowedIpsController()
mapper.connect("/ipam/tenants/{tenant_id}/"
"interfaces/{interface_id}/allowed_ips/{address:.+?}",
action="delete",
@ -595,48 +632,106 @@ class API(wsgi.Router):
path_prefix=("/ipam/tenants/{tenant_id}/"
"interfaces/{interface_id}"))
@classmethod
def app_factory(cls, global_conf, **local_conf):
return APIV01()
class APIV10(wsgi.Router):
def __init__(self):
mapper = routes.Mapper()
super(APIV10, self).__init__(mapper)
APICommon(mapper)
self._instance_interface_ips_mapper(mapper)
def _instance_interface_ips_mapper(self, mapper):
res = InstanceInterfaceIpsController().create_resource()
path_prefix = ("/ipam/instances/{device_id}/"
"interfaces/{interface_id}")
with mapper.submapper(controller=res,
path_prefix=path_prefix) as submap:
_connect(submap,
"/ip_addresses/{address:.+?}",
action="delete",
conditions=dict(method=["DELETE"]))
mapper.resource("ip_addresses",
"/ip_addresses",
controller=res,
path_prefix=path_prefix)
@classmethod
def app_factory(cls, global_conf, **local_conf):
return APIV10()
class APICommon():
def __init__(self, mapper):
self._natting_mapper(mapper,
"inside_globals",
InsideGlobalsController().create_resource())
self._natting_mapper(mapper,
"inside_locals",
InsideLocalsController().create_resource())
self._block_and_nested_resource_mapper(mapper)
self._policy_and_rules_mapper(mapper)
self._allocated_ips_mapper(mapper)
self._ip_routes_mapper(mapper)
self._instance_interface_mapper(mapper)
self._mac_address_range_mapper(mapper)
def _allocated_ips_mapper(self, mapper):
allocated_ips_res = AllocatedIpAddressesController().create_resource()
_connect(mapper,
"/ipam/allocated_ip_addresses",
controller=allocated_ips_res,
action="index",
conditions=dict(method=['GET']))
_connect(mapper,
"/ipam/tenants/{tenant_id}/allocated_ip_addresses",
controller=allocated_ips_res,
action="index",
conditions=dict(method=['GET']))
def _ip_routes_mapper(self, mapper):
ip_routes_res = IpRoutesController().create_resource()
path = ("/ipam/tenants/{tenant_id}/ip_blocks/{source_block_id}"
"/ip_routes")
mapper.resource("ip_routes", path, controller=ip_routes_res)
def _instance_interface_mapper(self, mapper):
res = InstanceInterfacesController().create_resource()
self._connect(mapper,
"/ipam/instances/{device_id}/interfaces",
controller=res,
action="update",
conditions=dict(method=['PUT']))
self._connect(mapper,
"/ipam/instances/{device_id}/interfaces",
controller=res,
action="show",
conditions=dict(method=['GET']))
self._connect(mapper,
"/ipam/instances/{device_id}/interfaces",
controller=res,
action="delete",
conditions=dict(method=['DELETE']))
_connect(mapper,
"/ipam/instances/{device_id}/interfaces",
controller=res,
action="update_all",
conditions=dict(method=['PUT']))
_connect(mapper,
"/ipam/instances/{device_id}/interfaces",
controller=res,
action="index",
conditions=dict(method=['GET']))
_connect(mapper,
"/ipam/instances/{device_id}/interfaces",
controller=res,
action="delete_all",
conditions=dict(method=['DELETE']))
mapper.resource("interfaces",
"/ipam/instances/{device_id}/interfaces",
controller=res)
_connect(mapper,
"/ipam/tenants/{tenant_id}/instances/{device_id}/"
"interfaces/{id}",
controller=res,
action="show",
conditions=dict(method=['GET']))
def _mac_address_range_mapper(self, mapper):
range_res = MacAddressRangesController().create_resource()
path = ("/ipam/mac_address_ranges")
mapper.resource("mac_address_ranges", path, controller=range_res)
def _networks_maper(self, mapper):
resource = NetworksController().create_resource()
path = "/ipam/tenants/{tenant_id}/networks/{network_id}"
mapper.resource("networks", path, controller=resource)
def _interface_ip_allocations_mapper(self, mapper):
path = ("/ipam/tenants/{tenant_id}/networks"
"/{network_id}/interfaces/{interface_id}")
resource = InterfaceIpAllocationsController().create_resource()
with mapper.submapper(controller=resource, path_prefix=path) as submap:
self._connect(submap, "/ip_allocations", action='create',
conditions=dict(method=['POST']))
self._connect(submap,
"/ip_allocations",
action='index',
conditions=dict(method=['GET']))
self._connect(submap, "/ip_allocations", action='bulk_delete',
conditions=dict(method=['DELETE']))
def _policy_and_rules_mapper(self, mapper):
policy_path = "/ipam/tenants/{tenant_id}/policies"
ip_ranges_resource = UnusableIpRangesController().create_resource()
@ -672,10 +767,10 @@ class API(wsgi.Router):
parent_resource["member_name"])
with mapper.submapper(controller=subnet_controller,
path_prefix=path_prefix) as submap:
self._connect(submap, "/subnets",
_connect(submap, "/subnets",
action="index",
conditions=dict(method=["GET"]))
self._connect(submap, "/subnets",
_connect(submap, "/subnets",
action="create",
conditions=dict(method=["POST"]))
@ -685,23 +780,23 @@ class API(wsgi.Router):
parent_resource["member_name"])
with mapper.submapper(controller=ip_address_controller,
path_prefix=path_prefix) as submap:
self._connect(submap,
_connect(submap,
"/ip_addresses/{address:.+?}",
action="show",
conditions=dict(method=["GET"]))
self._connect(submap,
_connect(submap,
"/ip_addresses/{address:.+?}",
action="delete",
conditions=dict(method=["DELETE"]))
self._connect(submap,
_connect(submap,
"/ip_addresses/{address:.+?}""/restore",
action="restore",
conditions=dict(method=["PUT"]))
#mapper.resource here for ip addresses was slowing down the tests
self._connect(submap, "/ip_addresses", action="create",
_connect(submap, "/ip_addresses", action="create",
conditions=dict(method=["POST"]))
self._connect(submap, "/ip_addresses", action="index",
_connect(submap, "/ip_addresses", action="index",
conditions=dict(method=["GET"]))
def _natting_mapper(self, mapper, nat_type, nat_controller):
@ -709,20 +804,17 @@ class API(wsgi.Router):
"ip_addresses/{address:.+?}/")
with mapper.submapper(controller=nat_controller,
path_prefix=path_prefix) as submap:
self._connect(submap, nat_type, action="create",
_connect(submap, nat_type, action="create",
conditions=dict(method=["POST"]))
self._connect(submap, nat_type, action="index",
_connect(submap, nat_type, action="index",
conditions=dict(method=["GET"]))
self._connect(submap, nat_type, action="delete",
_connect(submap, nat_type, action="delete",
conditions=dict(method=["DELETE"]))
self._connect(submap,
_connect(submap,
"%(nat_type)s/{%(nat_type)s_address:.+?}" % locals(),
action="delete",
conditions=dict(method=["DELETE"]))
def _connect(self, mapper, path, *args, **kwargs):
return mapper.connect(path + "{.format:(json|xml)?}", *args, **kwargs)
def app_factory(global_conf, **local_conf):
return API()
def _connect(mapper, path, *args, **kwargs):
return mapper.connect(path + "{.format:(json|xml)?}", *args, **kwargs)

2
melange/tests/unit/__init__.py
View File

@ -73,6 +73,6 @@ class TestApp(webtest.TestApp):
def setup():
options = {"config_file": tests.test_config_file()}
conf = config.Config.load_paste_config("melangeapp", options, None)
conf = config.Config.load_paste_config("melange", options, None)
db_api.db_reset(conf, db_based_ip_generator, db_based_mac_generator)

4
melange/tests/unit/test_extensions.py
View File

@ -28,8 +28,8 @@ class TestExtensions(unittest.TestCase):
def test_extension_loads_with_melange_xmlns(self):
options = {'config_file': tests.test_config_file()}
conf, app = config.Config.load_paste_app('melangeapi',
options, None)
conf, app = config.Config.load_paste_app('melangeapi_v0_1',
options, None)
test_app = webtest.TestApp(app)
response = test_app.get("/extensions.xml")

224
melange/tests/unit/test_ipam_service.py
View File

@ -39,9 +39,12 @@ class ControllerTestBase(tests.BaseTest):
def setUp(self):
super(ControllerTestBase, self).setUp()
conf, melange_app = config.Config.load_paste_app('melangeapp',
conf, melange_v0_1 = config.Config.load_paste_app('melangeapp_v0_1',
{"config_file": tests.test_config_file()}, None)
self.app = unit.TestApp(melange_app)
self.app = unit.TestApp(melange_v0_1)
conf, melange_v1_0 = config.Config.load_paste_app('melangeapp_v1_0',
{"config_file": tests.test_config_file()}, None)
self.appv1_0 = unit.TestApp(melange_v1_0)
class DummyApp(wsgi.Router):
@ -49,7 +52,7 @@ class DummyApp(wsgi.Router):
def __init__(self, controller):
mapper = routes.Mapper()
mapper.resource("resource", "/resources",
controller=controller.create_resource())
controller=controller.create_resource())
super(DummyApp, self).__init__(mapper)
@ -2314,7 +2317,7 @@ class TestInterfacesController(ControllerTestBase):
class TestInstanceInterfacesController(ControllerTestBase):
def test_update_creates_interfaces(self):
def test_update_all_creates_interfaces(self):
net_ids = ["net_id_1", "net_id_2", "net_id_3"]
for net_id in net_ids:
factory_models.PrivateIpBlockFactory(tenant_id="RAX",
@ -2330,6 +2333,7 @@ class TestInstanceInterfacesController(ControllerTestBase):
response = self.app.put_json("/ipam/instances/instance_id/interfaces",
put_data)
self.assertEqual(response.status_int, 200)
ifaces = sorted(models.Interface.find_all(device_id='instance_id'),
key=lambda iface: iface.plugged_in_network_id())
@ -2357,7 +2361,7 @@ class TestInstanceInterfacesController(ControllerTestBase):
self.assertTrue(models.IpAddress.get(
previous_ip.id).marked_for_deallocation)
def test_get_interfaces(self):
def test_get_all_interfaces(self):
provider_block = factory_models.IpBlockFactory(tenant_id="RAX",
network_id="net_id")
self._setup_interface_and_ip("instance_id",
@ -2370,7 +2374,7 @@ class TestInstanceInterfacesController(ControllerTestBase):
self.assertEqual([self._get_iface_data(iface)],
response.json['instance']['interfaces'])
def test_delete_interfaces(self):
def test_delete_all_interfaces_of_instance(self):
provider_block = factory_models.IpBlockFactory(tenant_id="RAX",
network_id="net_id")
self._setup_interface_and_ip("instance_id",
@ -2393,6 +2397,112 @@ class TestInstanceInterfacesController(ControllerTestBase):
self.assertIsNone(deleted_instance_ifaces)
self.assertIsNotNone(existing_instance_ifaces)
def test_create_an_interface(self):
provider_block = factory_models.IpBlockFactory(tenant_id="RAX",
network_id="net_id")
existing_ip_on_instance = self._setup_interface_and_ip("instance_id",
"leasee_tenant",
provider_block)
response = self.app.post_json("/ipam/instances/instance_id/interfaces",
{'interface': {
'tenant_id': "leasee_tenant",
'network': {
'id': "net_id",
'tenant_id': "RAX"
}
}
})
self.assertIsNotNone(models.Interface.find_by(
device_id="instance_id", id=existing_ip_on_instance.interface_id))
created_interface = models.Interface.find_by(
device_id="instance_id", id=response.json['interface']['id'])
self.assertEqual(created_interface.plugged_in_network_id(), "net_id")
self.assertEqual(created_interface.tenant_id, "leasee_tenant")
self.assertEqual(response.json['interface'],
self._get_iface_data(created_interface))
def test_show_an_interface(self):
provider_block = factory_models.IpBlockFactory(tenant_id="RAX",
network_id="net_id")
allocated_ip = self._setup_interface_and_ip("instance_id",
"leasee_tenant",
provider_block)
response = self.app.get("/ipam/instances/instance_id/interfaces/%s" %
allocated_ip.interface_id)
expected_interface = models.Interface.find(allocated_ip.interface_id)
self.assertEqual(response.json['interface'],
self._get_iface_data(expected_interface))
def test_show_an_interface_raises_404_for_non_existant_interface(self):
provider_block = factory_models.IpBlockFactory(tenant_id="RAX",
network_id="net_id")
noise_ip = self._setup_interface_and_ip("instance_id",
"leasee_tenant",
provider_block)
response = self.app.get("/ipam/instances/instance_id/interfaces/"
"bad_iface_id", status="*")
self.assertErrorResponse(response,
webob.exc.HTTPNotFound,
"Interface Not Found")
def test_show_an_interface_with_tenant_id(self):
provider_block = factory_models.IpBlockFactory(tenant_id="RAX",
network_id="net_id")
allocated_ip = self._setup_interface_and_ip("instance_id",
"leasee_tenant",
provider_block)
response = self.app.get("/ipam/tenants/leasee_tenant/"
"instances/instance_id/interfaces/%s" %
allocated_ip.interface_id)
expected_interface = models.Interface.find(allocated_ip.interface_id)
self.assertEqual(response.json['interface'],
self._get_iface_data(expected_interface))
def test_show_an_inteface_fails_for_wrong_tenant_id(self):
provider_block = factory_models.IpBlockFactory(tenant_id="RAX",
network_id="net_id")
allocated_ip = self._setup_interface_and_ip("instance_id",
"leasee_tenant",
provider_block)
response = self.app.get("/ipam/tenants/wrong_tenant_id/"
"instances/instance_id/interfaces/%s" %
allocated_ip.interface_id, status="*")
self.assertErrorResponse(response, webob.exc.HTTPNotFound,
"Interface Not Found")
def test_delete_an_interface(self):
provider_block = factory_models.IpBlockFactory(tenant_id="RAX",
network_id="net_id")
allocated_ip = self._setup_interface_and_ip("instance_id",
"leasee_tenant",
provider_block)
self.app.delete("/ipam/instances/instance_id/interfaces/%s" %
allocated_ip.interface_id)
self.assertIsNone(models.Interface.get(allocated_ip.interface_id))
self.assertTrue(
models.IpAddress.get(allocated_ip.id).marked_for_deallocation)
def test_delete_an_interface_raises_404_for_non_existant_interface(self):
provider_block = factory_models.IpBlockFactory(tenant_id="RAX",
network_id="net_id")
noise_ip = self._setup_interface_and_ip("instance_id",
"leasee_tenant",
provider_block)
response = self.app.delete("/ipam/instances/instance_id/interfaces/"
"bad_iface_id", status="*")
self.assertErrorResponse(response,
webob.exc.HTTPNotFound,
"Interface Not Found")
def _get_iface_data(self, iface):
return unit.sanitize(views.InterfaceConfigurationView(iface).data())
@ -2402,6 +2512,108 @@ class TestInstanceInterfacesController(ControllerTestBase):
return _allocate_ip(block, interface=iface)
class TestInstanceInterfaceIpsController(ControllerTestBase):
def setUp(self):
super(TestInstanceInterfaceIpsController, self).setUp()
self.block = factory_models.IpBlockFactory(tenant_id="RAX",
network_id="net_id",
cidr="10.1.1.1/29")
self.iface = factory_models.InterfaceFactory(device_id="instance_id",
tenant_id="leasee_tenant")
def test_create(self):
path = ("/ipam/instances/instance_id/"
"interfaces/%s/ip_addresses" % self.iface.id)
body = {'network':
{'id': self.block.network_id,
'tenant_id': self.block.tenant_id,
'address': "10.1.1.3"
}
}
response = self.appv1_0.post_json(path, body)
created_address = models.IpAddress.find_by(address="10.1.1.3")
self.assertEqual(created_address.interface_id, self.iface.id)
self.assertEqual(created_address.ip_block_id, self.block.id)
expected_ip_data = unit.sanitize(
views.IpConfigurationView(created_address).data())
self.assertEqual(response.json['ip_addresses'], expected_ip_data)
def test_create_raises_404_for_non_existant_interface(self):
path = ("/ipam/instances/instance_id/"
"interfaces/bad_iface_id/ip_addresses")
body = {'network':
{'id': self.block.network_id,
'tenant_id': self.block.tenant_id,
'address': "10.1.1.3"
}
}
response = self.appv1_0.post_json(path, body, status="*")
self.assertErrorResponse(response,
webob.exc.HTTPNotFound,
"Interface Not Found")
def test_create_raises_404_for_non_existant_network(self):
path = ("/ipam/instances/instance_id/"
"interfaces/%s/ip_addresses" % self.iface.id)
body = {'network':
{'id': "bad_net_id",
'tenant_id': self.block.tenant_id,
'address': "10.1.1.3"
}
}
response = self.appv1_0.post_json(path, body, status="*")
self.assertErrorResponse(response,
webob.exc.HTTPNotFound,
"Network bad_net_id not found")
def test_delete(self):
ip = self.block.allocate_ip(interface=self.iface)
url = ("/ipam/instances/instance_id/"
"interfaces/%s/ip_addresses/%s" % (self.iface.id, ip.address))
self.appv1_0.delete(url)
self.assertTrue(models.IpAddress.find(ip.id).marked_for_deallocation)
def test_delete_raises_404_for_non_existant_interface(self):
url = ("/ipam/instances/instance_id/"
"interfaces/bad_iface_id/ip_addresses/10.1.1.1")
response = self.appv1_0.delete(url, status="*")
self.assertErrorResponse(response,
webob.exc.HTTPNotFound,
"Interface Not Found")
def test_delete_raises_404_for_unplugged_interface(self):
url = ("/ipam/instances/instance_id/"
"interfaces/%s/ip_addresses/22.22.22.22" % self.iface.id)
response = self.appv1_0.delete(url, status="*")
self.assertErrorResponse(response,
webob.exc.HTTPNotFound,
"IpAddress Not Found")
def test_delete_raises_404_for_non_existant_ip(self):
self.block.allocate_ip(interface=self.iface)
url = ("/ipam/instances/instance_id/"
"interfaces/%s/ip_addresses/22.22.22.22" % self.iface.id)
response = self.appv1_0.delete(url, status="*")
err_msg = ("IpAddress with {'address': u'22.22.22.22', "
"'interface_id': u'%s'} for network net_id not found"
% self.iface.id)
self.assertErrorResponse(response, webob.exc.HTTPNotFound, err_msg)
class TestMacAddressRangesController(ControllerTestBase):
def test_create(self):