Vinkesh/Deepak |Network ip allocation for given port and address

melange/ipam/models.py

@@ -424,13 +424,17 @@ class Network(ModelBase):
             raise ModelNotFoundError("Network {0} not found".format(id))
         return cls(id=id, ip_blocks=ip_blocks)
 
-    def allocate_ip(self):
+    def allocate_ip(self, address=None, port_id=None):
         for ip_block in self.ip_blocks:
             try:
-                return ip_block.allocate_ip()
-            except NoMoreAddressesError:
+                return ip_block.allocate_ip(address=address, port_id=port_id)
+            except (NoMoreAddressesError, AddressDoesNotBelongError):
                 pass
-        raise NoMoreAddressesError
+
+        if(address):
+            raise AddressDoesNotBelongError
+        else:
+            raise NoMoreAddressesError
 
 
 def models():

melange/ipam/service.py

@@ -225,8 +225,10 @@ class PoliciesController(BaseController):
 class NetworksController(BaseController):
 
     def allocate_ip(self, request, network_id, tenant_id=None):
-        ip_address = Network.find_by(id=network_id, tenant_id=tenant_id).\
-                     allocate_ip()
+        network = Network.find_by(id=network_id, tenant_id=tenant_id)
+        address, port_id = self._get_optionals(request.params,
+                                               *['address', 'port_id'])
+        ip_address = network.allocate_ip(address=address, port_id=port_id)
         return dict(ip_address=ip_address.data()), 201
 
 

tests/unit/test_ipam_models.py

@@ -20,7 +20,9 @@ from tests.unit import BaseTest
 
 from melange.ipam.models import (ModelBase, IpBlock, IpAddress, Policy,
                                  IpRange, IpOctet, Network)
-from melange.ipam.models import ModelNotFoundError, NoMoreAddressesError
+from melange.ipam.models import (ModelNotFoundError, NoMoreAddressesError,
+                                 AddressDoesNotBelongError,
+                                 DuplicateAddressError)
 from melange.ipam import models
 from melange.db import session
 from tests.unit.factories.models import (PublicIpBlockFactory,
@@ -773,3 +775,38 @@ class TestNetwork(BaseTest):
         network = Network.find_by(id=1)
 
         self.assertRaises(NoMoreAddressesError, network.allocate_ip)
+
+    def test_allocate_ip_assigns_given_port_and_address(self):
+        ip_block = PublicIpBlockFactory(network_id=1, cidr="10.0.0.0/31")
+        network = Network.find_by(id=1)
+
+        allocated_ip = network.allocate_ip(address="10.0.0.1", port_id=123)
+
+        self.assertEqual(allocated_ip.address, "10.0.0.1")
+        self.assertEqual(allocated_ip.port_id, 123)
+
+    def test_allocate_ip_assigns_given_address_from_its_block(self):
+        ip_block1 = PublicIpBlockFactory(network_id=1, cidr="10.0.0.0/31")
+        ip_block2 = PublicIpBlockFactory(network_id=1, cidr="20.0.0.0/31")
+        network = Network(ip_blocks=[ip_block1, ip_block2])
+
+        allocated_ip = network.allocate_ip(address="20.0.0.1")
+
+        self.assertEqual(allocated_ip.address, "20.0.0.1")
+        self.assertEqual(allocated_ip.ip_block_id, ip_block2.id)
+
+    def test_allocate_ip_fails_if_given_address_is_not_in_network(self):
+        ip_block = PublicIpBlockFactory(network_id=1, cidr="10.0.0.0/31")
+        network = Network.find_by(id=1)
+
+        self.assertRaises(AddressDoesNotBelongError,
+                          network.allocate_ip, address="20.0.0.1")
+
+    def test_allocate_ip_fails_if_given_address_is_already_allocated(self):
+        ip_block1 = PublicIpBlockFactory(network_id=1, cidr="10.0.0.0/31")
+        ip_block2 = PublicIpBlockFactory(network_id=1, cidr="20.0.0.0/31")
+        IpAddressFactory(ip_block_id=ip_block2.id, address="20.0.0.0")
+        network = Network(ip_blocks=[ip_block1, ip_block2])
+
+        self.assertRaises(DuplicateAddressError,
+                          network.allocate_ip, address="20.0.0.0")

tests/unit/test_ipam_service.py

@@ -1181,6 +1181,38 @@ class NetworksControllerBase():
         self.assertEqual(response.status_int, 201)
         self.assertEqual(ip_address.data(), response.json['ip_address'])
 
+    def test_allocate_ip_address_for_a_port(self):
+        ip_block = self._ip_block_factory(network_id=1)
+
+        response = self.app.post("{0}/networks/1/ip_addresses"\
+                                 .format(self.network_path),
+                                 {'port_id': '123'})
+
+        ip_address = IpAddress.find_by(ip_block_id=ip_block.id, port_id=123)
+        self.assertEqual(response.status_int, 201)
+        self.assertEqual(ip_address.data(), response.json['ip_address'])
+
+    def test_allocate_ip_with_given_address(self):
+        ip_block = self._ip_block_factory(network_id=1, cidr='10.0.0.0/31')
+
+        response = self.app.post("{0}/networks/1/ip_addresses"\
+                                 .format(self.network_path),
+                                 {'address': '10.0.0.1'})
+
+        ip_address = IpAddress.find_by(ip_block_id=ip_block.id,
+                                       address='10.0.0.1')
+        self.assertEqual(response.status_int, 201)
+        self.assertEqual(ip_address.data(), response.json['ip_address'])
+
+    def test_allocate_ip_fails_when_network_doesnt_have_given_address(self):
+        ip_block = self._ip_block_factory(network_id=1, cidr='10.0.0.0/31')
+
+        response = self.app.post("{0}/networks/1/ip_addresses"\
+                                 .format(self.network_path),
+                                 {'address': '20.0.0.0'}, status='*')
+
+        self.assertEqual(response.status_int, 422)
+
     def test_allocate_ip_fails_if_network_not_found(self):
         response = self.app.post("{0}/networks/1/ip_addresses"\
                                  .format(self.network_path), status="*")