From e2581d50e76ac05488a284b4a1efb069248ba9f1 Mon Sep 17 00:00:00 2001 From: YAMAMOTO Takashi Date: Wed, 22 Jun 2016 17:38:18 +0900 Subject: [PATCH] Fix _get_id_for Raise BadRequest on the absense of the attribute. Also, ensure that the value is actually a UUID-like. Closes-Bug: #1595078 Change-Id: I6c236d5e4acdd6f13eca2877163facc2860a9445 --- neutron_dynamic_routing/db/bgp_db.py | 8 +++- .../tests/unit/db/test_bgp_db.py | 39 +++++++++++++++++-- 2 files changed, 41 insertions(+), 6 deletions(-) diff --git a/neutron_dynamic_routing/db/bgp_db.py b/neutron_dynamic_routing/db/bgp_db.py index 61d6fb7f..52ab5a6f 100644 --- a/neutron_dynamic_routing/db/bgp_db.py +++ b/neutron_dynamic_routing/db/bgp_db.py @@ -23,6 +23,7 @@ from sqlalchemy import orm from sqlalchemy.orm import aliased from sqlalchemy.orm import exc as sa_exc +from neutron_lib.api import validators from neutron_lib import constants as lib_consts from neutron_lib import exceptions as n_exc @@ -310,11 +311,14 @@ class BgpDbMixin(common_db.CommonDbMixin): def _get_id_for(self, resource, id_name): try: - return resource.get(id_name) - except AttributeError: + uuid = resource[id_name] + msg = validators.validate_uuid(uuid) + except KeyError: msg = _("%s must be specified") % id_name + if msg: raise n_exc.BadRequest(resource=bgp_ext.BGP_SPEAKER_RESOURCE_NAME, msg=msg) + return uuid def _get_bgp_peers_by_bgp_speaker_binding(self, context, bgp_speaker_id): with context.session.begin(subtransactions=True): diff --git a/neutron_dynamic_routing/tests/unit/db/test_bgp_db.py b/neutron_dynamic_routing/tests/unit/db/test_bgp_db.py index 14cd2585..08a7910a 100644 --- a/neutron_dynamic_routing/tests/unit/db/test_bgp_db.py +++ b/neutron_dynamic_routing/tests/unit/db/test_bgp_db.py @@ -31,6 +31,7 @@ from neutron_dynamic_routing.services.bgp import bgp_plugin _uuid = uuidutils.generate_uuid ADVERTISE_FIPS_KEY = 'advertise_floating_ip_host_routes' +IMAGINARY = '2b2334c8-adfe-42d9-82c6-ad866c7fc5d8' # non existent resource id class BgpEntityCreationMixin(object): @@ -306,7 +307,7 @@ class BgpTests(test_plugin.Ml2PluginV2TestCase, {'bgp_peer_id': peer['id']}) def test_remove_non_existent_bgp_peer(self): - bgp_peer_id = "imaginary" + bgp_peer_id = IMAGINARY with self.subnetpool_with_address_scope(4, prefixes=['8.0.0.0/8']) as sp: with self.bgp_speaker(sp['ip_version'], 1234) as speaker: @@ -317,7 +318,7 @@ class BgpTests(test_plugin.Ml2PluginV2TestCase, {'bgp_peer_id': bgp_peer_id}) def test_add_non_existent_bgp_peer(self): - bgp_peer_id = "imaginary" + bgp_peer_id = IMAGINARY with self.subnetpool_with_address_scope(4, prefixes=['8.0.0.0/8']) as sp: with self.bgp_speaker(sp['ip_version'], 1234) as speaker: @@ -327,6 +328,36 @@ class BgpTests(test_plugin.Ml2PluginV2TestCase, speaker['id'], {'bgp_peer_id': bgp_peer_id}) + def test_add_bgp_peer_without_id(self): + with self.subnetpool_with_address_scope(4, + prefixes=['8.0.0.0/8']) as sp: + with self.bgp_speaker(sp['ip_version'], 1234) as speaker: + self.assertRaises(n_exc.BadRequest, + self.bgp_plugin.add_bgp_peer, + self.context, + speaker['id'], + {}) + + def test_add_bgp_peer_with_bad_id(self): + with self.subnetpool_with_address_scope(4, + prefixes=['8.0.0.0/8']) as sp: + with self.bgp_speaker(sp['ip_version'], 1234) as speaker: + self.assertRaises(n_exc.BadRequest, + self.bgp_plugin.add_bgp_peer, + self.context, + speaker['id'], + {'bgp_peer_id': 'aaa'}) + + def test_add_bgp_peer_with_none_id(self): + with self.subnetpool_with_address_scope(4, + prefixes=['8.0.0.0/8']) as sp: + with self.bgp_speaker(sp['ip_version'], 1234) as speaker: + self.assertRaises(n_exc.BadRequest, + self.bgp_plugin.add_bgp_peer, + self.context, + speaker['id'], + {'bgp_peer_id': None}) + def test_add_gateway_network(self): with self.subnetpool_with_address_scope(4, prefixes=['8.0.0.0/8']) as sp: @@ -373,7 +404,7 @@ class BgpTests(test_plugin.Ml2PluginV2TestCase, self.assertEqual(1, len(new_speaker['networks'])) def test_add_non_existent_gateway_network(self): - network_id = "imaginary" + network_id = IMAGINARY with self.subnetpool_with_address_scope(4, prefixes=['8.0.0.0/8']) as sp: with self.bgp_speaker(sp['ip_version'], 1234) as speaker: @@ -383,7 +414,7 @@ class BgpTests(test_plugin.Ml2PluginV2TestCase, {'network_id': network_id}) def test_remove_non_existent_gateway_network(self): - network_id = "imaginary" + network_id = IMAGINARY with self.subnetpool_with_address_scope(4, prefixes=['8.0.0.0/8']) as sp: with self.bgp_speaker(sp['ip_version'], 1234) as speaker: