# License for the specific language governing permissions and limitations
# under the License.
-import datetime
import random
import netaddr
from neutron import neutron_plugin_base_v2
from neutron.openstack.common import excutils
from neutron.openstack.common import log as logging
-from neutron.openstack.common import timeutils
from neutron.openstack.common import uuidutils
return True
return False
- def update_fixed_ip_lease_expiration(self, context, network_id,
- ip_address, lease_remaining):
-
- expiration = (timeutils.utcnow() +
- datetime.timedelta(seconds=lease_remaining))
-
- query = context.session.query(models_v2.IPAllocation)
- query = query.filter_by(network_id=network_id, ip_address=ip_address)
-
- try:
- with context.session.begin(subtransactions=True):
- fixed_ip = query.one()
- fixed_ip.expiration = expiration
- except exc.NoResultFound:
- LOG.debug(_("No fixed IP found that matches the network "
- "%(network_id)s and ip address %(ip_address)s."),
- {'network_id': network_id,
- 'ip_address': ip_address})
-
@staticmethod
def _delete_ip_allocation(context, network_id, subnet_id, ip_address):
res = port_req.get_response(self.api)
self.assertEqual(res.status_int, webob.exc.HTTPClientError.code)
- def test_update_fixed_ip_lease_expiration_invalid_address(self):
- cfg.CONF.set_override('dhcp_lease_duration', 10)
- plugin = NeutronManager.get_plugin()
- with self.subnet() as subnet:
- with self.port(subnet=subnet) as port:
- update_context = context.Context('', port['port']['tenant_id'])
- with mock.patch.object(db_base_plugin_v2, 'LOG') as log:
- plugin.update_fixed_ip_lease_expiration(
- update_context,
- subnet['subnet']['network_id'],
- '255.255.255.0',
- 120)
- self.assertTrue(log.mock_calls)
-
def test_max_fixed_ips_exceeded(self):
with self.subnet(gateway_ip='10.0.0.3',
cidr='10.0.0.0/24') as subnet:
allocation_pools=allocation_pools)
def test_subnet_with_allocation_range(self):
- cfg.CONF.set_override('dhcp_lease_duration', 0)
with self.network() as network:
net_id = network['network']['id']
data = {'subnet': {'network_id': net_id,
port = self.deserialize(self.fmt, res)
# delete the port
self._delete('ports', port['port']['id'])
- cfg.CONF.set_override('dhcp_lease_duration', 120)
def test_create_subnet_with_none_gateway_allocation_pool(self):
cidr = '10.0.0.0/24'