diff options
| -rw-r--r-- | nova/api/openstack/compute/contrib/floating_ip_dns.py | 53 | ||||
| -rw-r--r-- | nova/tests/api/openstack/compute/contrib/test_floating_ip_dns.py | 104 |
2 files changed, 109 insertions, 48 deletions
diff --git a/nova/api/openstack/compute/contrib/floating_ip_dns.py b/nova/api/openstack/compute/contrib/floating_ip_dns.py index fbea0acf9..bddf3580c 100644 --- a/nova/api/openstack/compute/contrib/floating_ip_dns.py +++ b/nova/api/openstack/compute/contrib/floating_ip_dns.py @@ -14,6 +14,7 @@ # License for the specific language governing permissions and limitations # under the License +import socket import urllib import webob @@ -206,32 +207,40 @@ class FloatingIPDNSEntryController(object): context = req.environ['nova.context'] authorize(context) domain = _unquote_domain(domain_id) - name = id - entries = self.network_api.get_dns_entries_by_name(context, - name, domain) - entry = _create_dns_entry(entries[0], name, domain) - return _translate_dns_entry_view(entry) - - @wsgi.serializers(xml=FloatingIPDNSsTemplate) - def index(self, req, domain_id): - """Return a list of dns entries for the specified domain and ip.""" - context = req.environ['nova.context'] - authorize(context) - params = req.GET - floating_ip = params.get('ip') - domain = _unquote_domain(domain_id) + floating_ip = None + # Check whether id is a valid ipv4/ipv6 address. + try: + socket.inet_pton(socket.AF_INET, id) + floating_ip = id + except socket.error: + try: + socket.inet_pton(socket.AF_INET6, id) + floating_ip = id + except socket.error: + pass + + if floating_ip: + entries = self.network_api.get_dns_entries_by_address(context, + floating_ip, + domain) + else: + entries = self.network_api.get_dns_entries_by_name(context, id, + domain) - if not floating_ip: - raise webob.exc.HTTPUnprocessableEntity() + if not entries: + explanation = _("DNS entries not found.") + raise webob.exc.HTTPNotFound(explanation=explanation) - entries = self.network_api.get_dns_entries_by_address(context, - floating_ip, - domain) - entrylist = [_create_dns_entry(floating_ip, entry, domain) - for entry in entries] + if floating_ip: + entrylist = [_create_dns_entry(floating_ip, entry, domain) + for entry in entries] + dns_entries = _translate_dns_entries_view(entrylist) + return wsgi.ResponseObject(dns_entries, + xml=FloatingIPDNSsTemplate) - return _translate_dns_entries_view(entrylist) + entry = _create_dns_entry(entries[0], id, domain) + return _translate_dns_entry_view(entry) @wsgi.serializers(xml=FloatingIPDNSTemplate) def update(self, req, domain_id, id, body): diff --git a/nova/tests/api/openstack/compute/contrib/test_floating_ip_dns.py b/nova/tests/api/openstack/compute/contrib/test_floating_ip_dns.py index 4753ede32..ac636a4b5 100644 --- a/nova/tests/api/openstack/compute/contrib/test_floating_ip_dns.py +++ b/nova/tests/api/openstack/compute/contrib/test_floating_ip_dns.py @@ -29,8 +29,10 @@ from nova.tests.api.openstack import fakes name = "arbitraryname" name2 = "anotherarbitraryname" -testaddress = '10.0.0.66' -testaddress2 = '10.0.0.67' +test_ipv4_address = '10.0.0.66' +test_ipv4_address2 = '10.0.0.67' + +test_ipv6_address = 'fe80:0:0:0:0:0:a00:42' domain = "example.org" domain2 = "example.net" @@ -48,7 +50,7 @@ def _quote_domain(domain): def network_api_get_floating_ip(self, context, id): - return {'id': floating_ip_id, 'address': testaddress, + return {'id': floating_ip_id, 'address': test_ipv4_address, 'fixed_ip': None} @@ -65,11 +67,11 @@ def network_get_dns_entries_by_address(self, context, address, domain): def network_get_dns_entries_by_name(self, context, address, domain): - return [testaddress] + return [test_ipv4_address] def network_add_dns_entry(self, context, address, name, dns_type, domain): - return {'dns_entry': {'ip': testaddress, + return {'dns_entry': {'ip': test_ipv4_address, 'name': name, 'type': dns_type, 'domain': domain}} @@ -85,12 +87,16 @@ class FloatingIpDNSTest(test.TestCase): def _create_floating_ip(self): """Create a floating ip object.""" host = "fake_host" - return db.floating_ip_create(self.context, - {'address': testaddress, - 'host': host}) + db.floating_ip_create(self.context, + {'address': test_ipv4_address, + 'host': host}) + db.floating_ip_create(self.context, + {'address': test_ipv6_address, + 'host': host}) def _delete_floating_ip(self): - db.floating_ip_destroy(self.context, testaddress) + db.floating_ip_destroy(self.context, test_ipv4_address) + db.floating_ip_destroy(self.context, test_ipv6_address) def setUp(self): super(FloatingIpDNSTest, self).setUp() @@ -133,14 +139,17 @@ class FloatingIpDNSTest(test.TestCase): self.assertFalse(entries[2]['project']) self.assertEqual(entries[2]['availability_zone'], "avzone") - def test_get_dns_entries_by_address(self): - qparams = {'ip': testaddress} - params = "?%s" % urllib.urlencode(qparams) if qparams else "" + def _test_get_dns_entries_by_address(self, address): - req = fakes.HTTPRequest.blank('/v2/123/os-floating-ip-dns/%s/entries%s' - % (_quote_domain(domain), params)) - entries = self.entry_controller.index(req, _quote_domain(domain)) + qparams = {'ip': address} + params = "?%s" % urllib.urlencode(qparams) if qparams else "" + req = fakes.HTTPRequest.blank( + '/v2/123/os-floating-ip-dns/%s/entries/%s' + % (_quote_domain(domain), params)) + entries = self.entry_controller.show(req, _quote_domain(domain), + address) + entries = entries.obj self.assertEqual(len(entries['dns_entries']), 2) self.assertEqual(entries['dns_entries'][0]['name'], name) @@ -149,6 +158,35 @@ class FloatingIpDNSTest(test.TestCase): self.assertEqual(entries['dns_entries'][0]['domain'], domain) + def test_get_dns_entries_by_ipv4_address(self): + self._test_get_dns_entries_by_address(test_ipv4_address) + + def test_get_dns_entries_by_ipv6_address(self): + self._test_get_dns_entries_by_address(test_ipv6_address) + + def test_get_dns_entries_by_invalid_ipv4_or_ipv6(self): + # If it's not a valid ipv4 neither ipv6, the method 'show' + # will try to get dns entries by name instead. We use this + # to test if address is being correctly validated. + def fake_get_dns_entries_by_name(self, context, address, domain): + raise webob.exc.HTTPUnprocessableEntity() + + self.stubs.Set(network.api.API, "get_dns_entries_by_name", + fake_get_dns_entries_by_name) + + invalid_addr = '333.333.333.333' + + qparams = {'ip': invalid_addr} + params = "?%s" % urllib.urlencode(qparams) if qparams else "" + + req = fakes.HTTPRequest.blank( + '/v2/123/os-floating-ip-dns/%s/entries/%s' + % (_quote_domain(domain), params)) + + self.assertRaises(webob.exc.HTTPUnprocessableEntity, + self.entry_controller.show, + req, _quote_domain(domain), invalid_addr) + def test_get_dns_entries_by_name(self): req = fakes.HTTPRequest.blank( '/v2/123/os-floating-ip-dns/%s/entries/%s' % @@ -156,20 +194,34 @@ class FloatingIpDNSTest(test.TestCase): entry = self.entry_controller.show(req, _quote_domain(domain), name) self.assertEqual(entry['dns_entry']['ip'], - testaddress) + test_ipv4_address) self.assertEqual(entry['dns_entry']['domain'], domain) + def test_dns_entries_not_found(self): + def fake_get_dns_entries_by_name(self, context, address, domain): + raise webob.exc.HTTPNotFound() + + self.stubs.Set(network.api.API, "get_dns_entries_by_name", + fake_get_dns_entries_by_name) + + req = fakes.HTTPRequest.blank( + '/v2/123/os-floating-ip-dns/%s/entries/%s' % + (_quote_domain(domain), 'nonexistent')) + self.assertRaises(webob.exc.HTTPNotFound, + self.entry_controller.show, + req, _quote_domain(domain), 'nonexistent') + def test_create_entry(self): body = {'dns_entry': - {'ip': testaddress, + {'ip': test_ipv4_address, 'dns_type': 'A'}} req = fakes.HTTPRequest.blank( '/v2/123/os-floating-ip-dns/%s/entries/%s' % (_quote_domain(domain), name)) entry = self.entry_controller.update(req, _quote_domain(domain), name, body) - self.assertEqual(entry['dns_entry']['ip'], testaddress) + self.assertEqual(entry['dns_entry']['ip'], test_ipv4_address) def test_create_domain(self): req = fakes.HTTPRequest.blank('/v2/123/os-floating-ip-dns/%s' % @@ -264,13 +316,13 @@ class FloatingIpDNSTest(test.TestCase): def test_modify(self): body = {'dns_entry': - {'ip': testaddress2, + {'ip': test_ipv4_address2, 'dns_type': 'A'}} req = fakes.HTTPRequest.blank( '/v2/123/os-floating-ip-dns/%s/entries/%s' % (domain, name)) entry = self.entry_controller.update(req, domain, name, body) - self.assertEqual(entry['dns_entry']['ip'], testaddress2) + self.assertEqual(entry['dns_entry']['ip'], test_ipv4_address2) class FloatingIpDNSSerializerTest(test.TestCase): @@ -305,11 +357,11 @@ class FloatingIpDNSSerializerTest(test.TestCase): serializer = floating_ip_dns.FloatingIPDNSsTemplate() text = serializer.serialize(dict( dns_entries=[ - dict(ip=testaddress, + dict(ip=test_ipv4_address, type='A', domain=domain, name=name), - dict(ip=testaddress2, + dict(ip=test_ipv4_address2, type='C', domain=domain, name=name2)])) @@ -319,11 +371,11 @@ class FloatingIpDNSSerializerTest(test.TestCase): self.assertEqual(2, len(tree)) self.assertEqual('dns_entry', tree[0].tag) self.assertEqual('dns_entry', tree[1].tag) - self.assertEqual(testaddress, tree[0].get('ip')) + self.assertEqual(test_ipv4_address, tree[0].get('ip')) self.assertEqual('A', tree[0].get('type')) self.assertEqual(domain, tree[0].get('domain')) self.assertEqual(name, tree[0].get('name')) - self.assertEqual(testaddress2, tree[1].get('ip')) + self.assertEqual(test_ipv4_address2, tree[1].get('ip')) self.assertEqual('C', tree[1].get('type')) self.assertEqual(domain, tree[1].get('domain')) self.assertEqual(name2, tree[1].get('name')) @@ -332,7 +384,7 @@ class FloatingIpDNSSerializerTest(test.TestCase): serializer = floating_ip_dns.FloatingIPDNSTemplate() text = serializer.serialize(dict( dns_entry=dict( - ip=testaddress, + ip=test_ipv4_address, type='A', domain=domain, name=name))) @@ -340,6 +392,6 @@ class FloatingIpDNSSerializerTest(test.TestCase): tree = etree.fromstring(text) self.assertEqual('dns_entry', tree.tag) - self.assertEqual(testaddress, tree.get('ip')) + self.assertEqual(test_ipv4_address, tree.get('ip')) self.assertEqual(domain, tree.get('domain')) self.assertEqual(name, tree.get('name')) |