Merge "Create INI file ConfigurationSourceDriver."

This commit is contained in:
Zuul 2018-07-13 16:08:31 +00:00 committed by Gerrit Code Review
commit e95a4ce8c3
3 changed files with 247 additions and 0 deletions

View File

@ -13,6 +13,11 @@ import abc
import six
# We cannot use None as a sentinel indicating a missing value because it
# may be a valid value or default, so we use a custom singleton instead.
_NoValue = object()
@six.add_metaclass(abc.ABCMeta)
class ConfigurationSourceDriver(object):
"""A config driver option for Oslo.config.

112
oslo_config/sources/ini.py Normal file
View File

@ -0,0 +1,112 @@
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
import requests
import tempfile
from oslo_config import cfg
from oslo_config import sources
class INIConfigurationSourceDriver(sources.ConfigurationSourceDriver):
"""A configuration source driver for INI files served through http[s].
Required options:
- uri: URI containing the file location.
Non-required options:
- ca_path: The path to a CA_BUNDLE file or directory with
certificates of trusted CAs.
- client_cert: Client side certificate, as a single file path
containing either the certificate only or the
private key and the certificate.
- client_key: Client side private key, in case client_cert is
specified but does not includes the private key.
"""
def open_source_from_opt_group(self, conf, group_name):
group = cfg.OptGroup(group_name)
conf.register_opt(cfg.URIOpt("uri",
schemes=["http", "https"],
required=True),
group)
conf.register_opt(cfg.StrOpt("ca_path"), group)
conf.register_opt(cfg.StrOpt("client_cert"), group)
conf.register_opt(cfg.StrOpt("client_key"), group)
return INIConfigurationSource(
conf[group_name].uri,
conf[group_name].ca_path,
conf[group_name].client_cert,
conf[group_name].client_key)
class INIConfigurationSource(sources.ConfigurationSource):
"""A configuration source for INI files server through http[s].
:uri: The Uniform Resource Identifier of the configuration to be
retrieved.
:ca_path: The path to a CA_BUNDLE file or directory with
certificates of trusted CAs.
:client_cert: Client side certificate, as a single file path
containing either the certificate only or the
private key and the certificate.
:client_key: Client side private key, in case client_cert is
specified but does not includes the private key.
"""
def __init__(self, uri, ca_path, client_cert, client_key):
self._uri = uri
self._namespace = cfg._Namespace(cfg.ConfigOpts())
data = self._fetch_uri(uri, ca_path, client_cert, client_key)
with tempfile.NamedTemporaryFile() as tmpfile:
tmpfile.write(data.encode("utf-8"))
tmpfile.flush()
cfg.ConfigParser._parse_file(tmpfile.name, self._namespace)
def _fetch_uri(self, uri, ca_path, client_cert, client_key):
verify = ca_path if ca_path else True
cert = (client_cert, client_key) if client_cert and client_key else \
client_cert
with requests.get(uri, verify=verify, cert=cert) as response:
response.raise_for_status() # raises only in case of HTTPError
return response.text
def get(self, group_name, option_name, opt):
"""Return the value of the option from the group.
:param group_name: Name of the group.
:type group_name: str
:param option_name: Name of the option.
:type option_name: str
:param opt: The option definition.
:type opt: Opt
:return: Option value or NoValue.
"""
try:
return self._namespace._get_value(
[(group_name, option_name)],
multi=opt.multi)
except KeyError:
return (sources._NoValue, None)

View File

@ -0,0 +1,130 @@
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
import tempfile
from oslo_config import cfg
from oslo_config import sources
from oslotest import base
_GROUP = "group"
_OPTIONS = "options"
_DEFAULT = "DEFAULT"
_extra_ini_opt_group = "extra_conf_from_ini"
_extra_conf_url = "https://oslo.config/extra.conf"
_conf_data = {
_DEFAULT: {
"config_sources": (cfg.StrOpt, _extra_ini_opt_group)
},
_extra_ini_opt_group: {
"driver": (cfg.StrOpt, "ini"),
"uri": (cfg.URIOpt, _extra_conf_url)
}
}
_extra_conf_data = {
_DEFAULT: {
"foo": (cfg.StrOpt, "bar")
},
"test": {
"opt_str": (cfg.StrOpt, "a nice string"),
"opt_bool": (cfg.BoolOpt, True),
"opt_int": (cfg.IntOpt, 42),
"opt_float": (cfg.FloatOpt, 3.14),
"opt_ip": (cfg.IPOpt, "127.0.0.1"),
"opt_port": (cfg.PortOpt, 443),
"opt_host": (cfg.HostnameOpt, "www.openstack.org"),
"opt_uri": (cfg.URIOpt, "https://www.openstack.org"),
"opt_multi": (cfg.MultiStrOpt, ["abc", "def", "ghi"])
}
}
def register_opts(conf, opts):
# 'g': group, 'o': option, and 't': type
for g in opts.keys():
for o, (t, _) in opts[g].items():
try:
conf.register_opt(t(o), g if g != "DEFAULT" else None)
except cfg.DuplicateOptError:
pass
def opts_to_ini(opts):
result = ""
# 'g': group, 'o': option, 't': type, and 'v': value
for g in opts.keys():
result += "[{}]\n".format(g)
for o, (t, v) in opts[g].items():
if t == cfg.MultiStrOpt:
for i in v:
result += "{} = {}\n".format(o, i)
else:
result += "{} = {}\n".format(o, v)
return result
def mocked_get(*args, **kwargs):
class MockResponse(object):
def __init__(self, text_data, status_code):
self.text = text_data
self.status_code = status_code
def __enter__(self, *args, **kwargs):
return self
def __exit__(self, *args, **kwargs):
pass
def raise_for_status(self):
if self.status_code != 200:
raise
if args[0] in _extra_conf_url:
return MockResponse(opts_to_ini(_extra_conf_data), 200)
return MockResponse(None, 404)
class INISourceTestCase(base.BaseTestCase):
def setUp(self):
super(INISourceTestCase, self).setUp()
self.conf = cfg.ConfigOpts()
with tempfile.NamedTemporaryFile() as tmp_file:
tmp_file.write(opts_to_ini(_conf_data).encode("utf-8"))
tmp_file.flush()
self.conf(["--config-file", tmp_file.name])
@base.mock.patch(
"oslo_config.sources.ini.requests.get", side_effect=mocked_get)
def test_configuration_source(self, mock_requests_get):
driver = sources.ini.INIConfigurationSourceDriver()
source = driver.open_source_from_opt_group(
self.conf, _extra_ini_opt_group)
# non-existing option
self.assertIs(sources._NoValue,
source.get("DEFAULT", "bar", cfg.StrOpt("bar"))[0])
# 'g': group, 'o': option, 't': type, and 'v': value
for g in _extra_conf_data:
for o, (t, v) in _extra_conf_data[g].items():
self.assertEqual(str(v), str(source.get(g, o, t(o))[0]))