# under the License.
import os
-import re
import socket
import sys
import uuid
from quantum.agent.linux import interface
from quantum.agent.linux import ip_lib
from quantum.agent import rpc as agent_rpc
-from quantum.api.v2 import attributes
from quantum.common import exceptions
from quantum.common import topics
from quantum.openstack.common import cfg
from quantum.openstack.common import jsonutils
from quantum.openstack.common import log as logging
from quantum.openstack.common.rpc import proxy
+from quantum.openstack.common import uuidutils
LOG = logging.getLogger(__name__)
NS_PREFIX = 'qdhcp-'
else:
os.makedirs(dirname, 0755)
- def _validate_field(self, value, regex):
- """Validate value against a regular expression and return if valid."""
- match = re.match(regex, value)
-
- if match:
- return value
- raise ValueError(_("Value %s does not match regex: %s") %
- (value, regex))
-
def _handler(self, client_sock, client_addr):
"""Handle incoming lease relay stream connection.
data = jsonutils.loads(msg)
client_sock.close()
- network_id = self._validate_field(data['network_id'],
- attributes.UUID_PATTERN)
+ network_id = data['network_id']
+ if not uuidutils.is_uuid_like(network_id):
+ raise ValueError(_("Network ID %s is not a valid UUID") %
+ (network_id))
ip_address = str(netaddr.IPAddress(data['ip_address']))
lease_remaining = int(data['lease_remaining'])
self.callback(network_id, ip_address, lease_remaining)
exists.assert_called_once_with(
cfg.CONF.dhcp_lease_relay_socket)
- def test_validate_field_valid(self):
- relay = dhcp_agent.DhcpLeaseRelay(None)
- retval = relay._validate_field('1b', '\d[a-f]')
- self.assertEqual(retval, '1b')
-
- def test_validate_field_invalid(self):
- relay = dhcp_agent.DhcpLeaseRelay(None)
- with self.assertRaises(ValueError):
- retval = relay._validate_field('zz', '\d[a-f]')
-
def test_handler_valid_data(self):
network_id = 'cccccccc-cccc-cccc-cccc-cccccccccccc'
ip_address = '192.168.1.9'
relay = dhcp_agent.DhcpLeaseRelay(handler)
- with mock.patch.object(relay, '_validate_field') as validate:
- validate.side_effect = ValueError
+ with mock.patch('quantum.openstack.common.'
+ 'uuidutils.is_uuid_like') as validate:
+ validate.return_value = False
with mock.patch.object(dhcp_agent.LOG, 'warn') as log: