Merge "Check subnetpool extension support"

changes/95/243395/5
Jenkins 7 years ago committed by Gerrit Code Review
commit 964a8ea19c
  1. 12
      kuryr/common/exceptions.py
  2. 15
      kuryr/controllers.py
  3. 18
      kuryr/tests/test_config.py
  4. 2
      scripts/run_server.py

@ -26,8 +26,16 @@ class BindingFailure(KuryrException):
class DuplicatedResourceException(KuryrException):
"""Exception represents there're multiple resources for the ID.
This exception is thrown when you query the Neutron resouce associated with
the ID and you get multiple resources.
For example, this exception is thrown when you query the Neutron resource
associated with the ID and you get multiple resources.
"""
class MandatoryApiMissing(KuryrException):
"""Exception represents that mandatory api is not found.
For example, this exception is thrown when expected neutron
extension(subnetpools) APIs are not found.
"""

@ -28,6 +28,8 @@ from kuryr.common import exceptions
from kuryr import schemata
from kuryr import utils
MANDATORY_NEUTRON_EXTENSION = "subnet_allocation"
cfg.CONF.import_group('neutron_client', 'kuryr.common.config')
cfg.CONF.import_group('keystone_client', 'kuryr.common.config')
@ -49,6 +51,19 @@ else:
app.neutron = utils.get_neutron_client_simple(
url=neutron_uri, token=auth_token)
def check_for_neutron_ext_support():
"""Validates for mandatory extension support availability in neutron.
"""
try:
app.neutron.show_extension(MANDATORY_NEUTRON_EXTENSION)
except n_exceptions.NeutronClientException as e:
if e.status_code == n_exceptions.NotFound.status_code:
raise exceptions.MandatoryApiMissing(
"Neutron extension with alias '{0}' not found"
.format(MANDATORY_NEUTRON_EXTENSION))
# TODO(tfukushima): Retrieve the following subnet names from the config file.
SUBNET_POOLS_V4 = [
p.strip() for p in os.environ.get('SUBNET_POOLS_V4', 'kuryr').split(',')]

@ -12,11 +12,15 @@
import os
from neutronclient.common import exceptions as n_exceptions
from kuryr.common import config
from kuryr.common import exceptions
from kuryr import controllers
from kuryr.tests import base
class ConfigurationTest(base.TestCase):
class ConfigurationTest(base.TestKuryrBase):
def test_defaults(self):
basepath = os.path.abspath(os.path.join(os.path.dirname(__file__),
@ -33,3 +37,15 @@ class ConfigurationTest(base.TestCase):
self.assertEqual('http://127.0.0.1:35357',
config.CONF.keystone_client.auth_uri)
def test_check_for_neutron_ext_support_with_ex(self):
ext_alias = "subnet_allocation"
self.mox.StubOutWithMock(controllers.app.neutron, "show_extension")
err = n_exceptions.NotFound.status_code
ext_not_found_ex = n_exceptions.NeutronClientException(status_code=err,
message="")
neutron = controllers.app.neutron
neutron.show_extension(ext_alias).AndRaise(ext_not_found_ex)
self.mox.ReplayAll()
ex = exceptions.MandatoryApiMissing
self.assertRaises(ex, controllers.check_for_neutron_ext_support)

@ -12,6 +12,8 @@
# License for the specific language governing permissions and limitations
# under the License.
from kuryr import controllers
from kuryr import server
controllers.check_for_neutron_ext_support()
server.start()

Loading…
Cancel
Save