x/cloudbase-init
Code Issues Proposed changes

Restructure the config options from metadata services

Browse Source 
Change-Id: I3724e7d70e4727214e712e32f72e773451c5c006
This commit is contained in:
Alexandru Coman 2016-08-10 16:25:06 +03:00
parent 9ddb45127b
commit b46605f71f
12 changed files with 167 additions and 119 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

10
cloudbaseinit/metadata/services/baseopenstackservice.py
View File

@ -25,13 +25,15 @@ from cloudbaseinit.utils import encoding
from cloudbaseinit.utils import x509constants
opts = [
cfg.StrOpt('metadata_base_url', default='http://169.254.169.254/',
help='The base URL where the service looks for metadata'),
OPENSTACK_OPTS = [
cfg.StrOpt("metadata_base_url", default="http://169.254.169.254/",
help="The base URL where the service looks for metadata",
deprecated_group="DEFAULT")
]
CONF = cfg.CONF
CONF.register_opts(opts)
CONF.register_group(cfg.OptGroup('openstack'))
CONF.register_opts(OPENSTACK_OPTS, 'openstack')
LOG = oslo_logging.getLogger(__name__)

54
cloudbaseinit/metadata/services/cloudstack.py
View File

@ -13,6 +13,7 @@
# under the License.
import contextlib
import posixpath
from oslo_config import cfg
from oslo_log import log as oslo_logging
@ -23,16 +24,19 @@ from cloudbaseinit.metadata.services import base
from cloudbaseinit.osutils import factory as osutils_factory
from cloudbaseinit.utils import encoding
CLOUDSTACK_OPTS = [
cfg.StrOpt("metadata_base_url", default="http://10.1.1.1/",
help="The base URL where the service looks for metadata",
deprecated_name="cloudstack_metadata_ip",
deprecated_group="DEFAULT"),
]
CONF = cfg.CONF
CONF.register_group(cfg.OptGroup("cloudstack"))
CONF.register_opts(CLOUDSTACK_OPTS, "cloudstack")
LOG = oslo_logging.getLogger(__name__)
OPTS = [
cfg.StrOpt('cloudstack_metadata_ip', default="10.1.1.1",
help='The IP adress where the service looks for metadata'),
]
CONF = cfg.CONF
CONF.register_opts(OPTS)
BAD_REQUEST = b"bad_request"
SAVED_PASSWORD = b"saved_password"
TIMEOUT = 10
@ -40,20 +44,22 @@ TIMEOUT = 10
class CloudStack(base.BaseMetadataService):
URI_TEMPLATE = 'http://%s/latest/meta-data/'
def __init__(self):
super(CloudStack, self).__init__()
self.osutils = osutils_factory.get_os_utils()
self._metadata_uri = None
self._metadata_url = None
self._router_ip = None
def _test_api(self, ip_address):
def _get_path(self, resource, version="latest"):
"""Get the relative path for the received resource."""
return posixpath.normpath(
posixpath.join(version, "meta-data", resource))
def _test_api(self, metadata_url):
"""Test if the CloudStack API is responding properly."""
self._metadata_uri = self.URI_TEMPLATE % ip_address
self._metadata_url = metadata_url
try:
response = self._http_request(self._metadata_uri)
self._get_data('service-offering')
response = self._get_data(self._get_path("service-offering"))
except urllib.error.HTTPError as exc:
LOG.debug('Error response code: %s', exc.code)
return False
@ -66,13 +72,13 @@ class CloudStack(base.BaseMetadataService):
return False
LOG.debug('Available services: %s', response)
self._router_ip = ip_address
self._router_ip = urllib.parse.urlparse(metadata_url).netloc
return True
def load(self):
"""Obtain all the required informations."""
super(CloudStack, self).load()
if self._test_api(CONF.cloudstack_metadata_ip):
if self._test_api(CONF.cloudstack.metadata_base_url):
return True
dhcp_servers = self.osutils.get_dhcp_hosts_in_use()
@ -81,7 +87,7 @@ class CloudStack(base.BaseMetadataService):
return False
for _, ip_address in dhcp_servers:
LOG.debug('Testing: %s', ip_address)
if self._test_api(ip_address):
if self._test_api('http://%s/' % ip_address):
return True
return False
@ -95,9 +101,9 @@ class CloudStack(base.BaseMetadataService):
def _get_data(self, path):
"""Getting required metadata using CloudStack metadata API."""
metadata_uri = urllib.parse.urljoin(self._metadata_uri, path)
metadata_url = urllib.parse.urljoin(self._metadata_url, path)
try:
content = self._http_request(metadata_uri)
content = self._http_request(metadata_url)
except urllib.error.HTTPError as exc:
if exc.code == 404:
raise base.NotExistingMetadataException()
@ -106,20 +112,22 @@ class CloudStack(base.BaseMetadataService):
def get_instance_id(self):
"""Instance name of the virtual machine."""
return self._get_cache_data('instance-id', decode=True)
return self._get_cache_data(self._get_path("instance-id"),
decode=True)
def get_host_name(self):
"""Hostname of the virtual machine."""
return self._get_cache_data('local-hostname', decode=True)
return self._get_cache_data(self._get_path("local-hostname"),
decode=True)
def get_user_data(self):
"""User data for this virtual machine."""
return self._get_cache_data('../user-data')
return self._get_cache_data(self._get_path('../user-data'))
def get_public_keys(self):
"""Available ssh public keys."""
ssh_keys = []
ssh_chunks = self._get_cache_data('public-keys',
ssh_chunks = self._get_cache_data(self._get_path("public-keys"),
decode=True).splitlines()
for ssh_key in ssh_chunks:
ssh_key = ssh_key.strip()

45
cloudbaseinit/metadata/services/configdrive.py
View File

@ -41,24 +41,35 @@ CD_LOCATIONS = {
"partition",
}
opts = [
cfg.BoolOpt('config_drive_raw_hhd', default=True,
help='Look for an ISO config drive in raw HDDs',
CONFIG_DRIVE_OPTS = [
cfg.BoolOpt("raw_hdd", default=True,
help="Look for an ISO config drive in raw HDDs",
deprecated_name="config_drive_raw_hhd",
deprecated_group="DEFAULT",
deprecated_for_removal=True),
cfg.BoolOpt('config_drive_cdrom', default=True,
help='Look for a config drive in the attached cdrom drives',
cfg.BoolOpt("cdrom", default=True,
help="Look for a config drive in the attached cdrom drives",
deprecated_name="config_drive_cdrom",
deprecated_group="DEFAULT",
deprecated_for_removal=True),
cfg.BoolOpt('config_drive_vfat', default=True,
help='Look for a config drive in VFAT filesystems',
cfg.BoolOpt("vfat", default=True,
help="Look for a config drive in VFAT filesystems",
deprecated_name="config_drive_vfat",
deprecated_group="DEFAULT",
deprecated_for_removal=True),
cfg.ListOpt('config_drive_types', default=list(CD_TYPES),
help='Supported formats of a configuration drive'),
cfg.ListOpt('config_drive_locations', default=list(CD_LOCATIONS),
help='Supported configuration drive locations'),
cfg.ListOpt("types", default=list(CD_TYPES),
help="Supported formats of a configuration drive",
deprecated_name="config_drive_types",
deprecated_group="DEFAULT",),
cfg.ListOpt("locations", default=list(CD_LOCATIONS),
deprecated_name="config_drive_locations",
deprecated_group="DEFAULT",
help="Supported configuration drive locations"),
]
CONF = cfg.CONF
CONF.register_opts(opts)
CONF.register_group(cfg.OptGroup("config_drive"))
CONF.register_opts(CONFIG_DRIVE_OPTS, "config_drive")
LOG = oslo_logging.getLogger(__name__)
@ -70,17 +81,17 @@ class ConfigDriveService(baseopenstackservice.BaseOpenStackService):
self._metadata_path = None
def _preprocess_options(self):
self._searched_types = set(CONF.config_drive_types)
self._searched_locations = set(CONF.config_drive_locations)
self._searched_types = set(CONF.config_drive.types)
self._searched_locations = set(CONF.config_drive.locations)
# Deprecation backward compatibility.
if CONF.config_drive_raw_hhd:
if CONF.config_drive.raw_hdd:
self._searched_types.add("iso")
self._searched_locations.add("hdd")
if CONF.config_drive_cdrom:
if CONF.config_drive.cdrom:
self._searched_types.add("iso")
self._searched_locations.add("cdrom")
if CONF.config_drive_vfat:
if CONF.config_drive.vfat:
self._searched_types.add("vfat")
self._searched_locations.add("hdd")

26
cloudbaseinit/metadata/services/ec2service.py
View File

@ -23,16 +23,20 @@ from six.moves.urllib import request
from cloudbaseinit.metadata.services import base
from cloudbaseinit.utils import network
opts = [
cfg.StrOpt('ec2_metadata_base_url',
default='http://169.254.169.254/',
help='The base URL where the service looks for metadata'),
cfg.BoolOpt('ec2_add_metadata_private_ip_route', default=True,
help='Add a route for the metadata ip address to the gateway'),
EC2_OPTS = [
cfg.StrOpt("metadata_base_url", default="http://169.254.169.254/",
help="The base URL where the service looks for metadata",
deprecated_name="ec2_metadata_base_url",
deprecated_group="DEFAULT"),
cfg.BoolOpt("add_metadata_private_ip_route", default=True,
help="Add a route for the metadata ip address to the gateway",
deprecated_name="ec2_add_metadata_private_ip_route",
deprecated_group="DEFAULT"),
]
CONF = cfg.CONF
CONF.register_opts(opts)
CONF.register_group(cfg.OptGroup("ec2"))
CONF.register_opts(EC2_OPTS, "ec2")
LOG = oslo_logging.getLogger(__name__)
@ -46,8 +50,8 @@ class EC2Service(base.BaseMetadataService):
def load(self):
super(EC2Service, self).load()
if CONF.ec2_add_metadata_private_ip_route:
network.check_metadata_ip_route(CONF.ec2_metadata_base_url)
if CONF.ec2.add_metadata_private_ip_route:
network.check_metadata_ip_route(CONF.ec2.metadata_base_url)
try:
self.get_host_name()
@ -55,7 +59,7 @@ class EC2Service(base.BaseMetadataService):
except Exception as ex:
LOG.exception(ex)
LOG.debug('Metadata not found at URL \'%s\'' %
CONF.ec2_metadata_base_url)
CONF.ec2.metadata_base_url)
return False
def _get_response(self, req):
@ -68,7 +72,7 @@ class EC2Service(base.BaseMetadataService):
raise
def _get_data(self, path):
norm_path = posixpath.join(CONF.ec2_metadata_base_url, path)
norm_path = posixpath.join(CONF.ec2.metadata_base_url, path)
LOG.debug('Getting metadata from: %(norm_path)s',
{'norm_path': norm_path})

24
cloudbaseinit/metadata/services/httpservice.py
View File

@ -23,15 +23,17 @@ from cloudbaseinit.metadata.services import base
from cloudbaseinit.metadata.services import baseopenstackservice
from cloudbaseinit.utils import network
opts = [
cfg.StrOpt('metadata_base_url', default='http://169.254.169.254/',
help='The base URL where the service looks for metadata'),
cfg.BoolOpt('add_metadata_private_ip_route', default=True,
help='Add a route for the metadata ip address to the gateway'),
OPENSTACK_OPTS = [
cfg.StrOpt("metadata_base_url", default="http://169.254.169.254/",
help="The base URL where the service looks for metadata",
deprecated_group="DEFAULT"),
cfg.BoolOpt("add_metadata_private_ip_route", default=True,
help="Add a route for the metadata ip address to the gateway",
deprecated_group="DEFAULT"),
]
CONF = cfg.CONF
CONF.register_opts(opts)
CONF.register_opts(OPENSTACK_OPTS, "openstack")
LOG = oslo_logging.getLogger(__name__)
@ -45,15 +47,15 @@ class HttpService(baseopenstackservice.BaseOpenStackService):
def load(self):
super(HttpService, self).load()
if CONF.add_metadata_private_ip_route:
network.check_metadata_ip_route(CONF.metadata_base_url)
if CONF.openstack.add_metadata_private_ip_route:
network.check_metadata_ip_route(CONF.openstack.metadata_base_url)
try:
self._get_meta_data()
return True
except Exception:
LOG.debug('Metadata not found at URL \'%s\'' %
CONF.metadata_base_url)
CONF.openstack.metadata_base_url)
return False
def _get_response(self, req):
@ -66,14 +68,14 @@ class HttpService(baseopenstackservice.BaseOpenStackService):
raise
def _get_data(self, path):
norm_path = posixpath.join(CONF.metadata_base_url, path)
norm_path = posixpath.join(CONF.openstack.metadata_base_url, path)
LOG.debug('Getting metadata from: %s', norm_path)
req = request.Request(norm_path)
response = self._get_response(req)
return response.read()
def _post_data(self, path, data):
norm_path = posixpath.join(CONF.metadata_base_url, path)
norm_path = posixpath.join(CONF.openstack.metadata_base_url, path)
LOG.debug('Posting metadata to: %s', norm_path)
req = request.Request(norm_path, data=data)
self._get_response(req)

49
cloudbaseinit/metadata/services/maasservice.py
View File

@ -24,21 +24,32 @@ from six.moves.urllib import request
from cloudbaseinit.metadata.services import base
from cloudbaseinit.utils import x509constants
opts = [
cfg.StrOpt('maas_metadata_url', default=None,
help='The base URL for MaaS metadata'),
cfg.StrOpt('maas_oauth_consumer_key', default="",
help='The MaaS OAuth consumer key'),
cfg.StrOpt('maas_oauth_consumer_secret', default="",
help='The MaaS OAuth consumer secret'),
cfg.StrOpt('maas_oauth_token_key', default="",
help='The MaaS OAuth token key'),
cfg.StrOpt('maas_oauth_token_secret', default="",
help='The MaaS OAuth token secret'),
MAAS_OPTS = [
cfg.StrOpt("metadata_base_url", default=None,
help="The base URL for MaaS metadata",
deprecated_name="maas_metadata_url",
deprecated_group="DEFAULT"),
cfg.StrOpt("oauth_consumer_key", default="",
help="The MaaS OAuth consumer key",
deprecated_name="maas_oauth_consumer_key",
deprecated_group="DEFAULT"),
cfg.StrOpt("oauth_consumer_secret", default="",
help="The MaaS OAuth consumer secret",
deprecated_name="maas_oauth_consumer_secret",
deprecated_group="DEFAULT"),
cfg.StrOpt("oauth_token_key", default="",
help="The MaaS OAuth token key",
deprecated_name="maas_oauth_token_key",
deprecated_group="DEFAULT"),
cfg.StrOpt("oauth_token_secret", default="",
help="The MaaS OAuth token secret",
deprecated_name="maas_oauth_token_secret",
deprecated_group="DEFAULT"),
]
CONF = cfg.CONF
CONF.register_opts(opts)
CONF.register_group(cfg.OptGroup("maas"))
CONF.register_opts(MAAS_OPTS, "maas")
LOG = oslo_logging.getLogger(__name__)
@ -65,7 +76,7 @@ class MaaSHttpService(base.BaseMetadataService):
def load(self):
super(MaaSHttpService, self).load()
if not CONF.maas_metadata_url:
if not CONF.maas.metadata_base_url:
LOG.debug('MaaS metadata url not set')
else:
try:
@ -74,7 +85,7 @@ class MaaSHttpService(base.BaseMetadataService):
except Exception as ex:
LOG.exception(ex)
LOG.debug('Metadata not found at URL \'%s\'' %
CONF.maas_metadata_url)
CONF.maas.metadata_base_url)
return False
def _get_response(self, req):
@ -88,17 +99,17 @@ class MaaSHttpService(base.BaseMetadataService):
def _get_oauth_headers(self, url):
client = oauth1.Client(
CONF.maas_oauth_consumer_key,
client_secret=CONF.maas_oauth_consumer_secret,
resource_owner_key=CONF.maas_oauth_token_key,
resource_owner_secret=CONF.maas_oauth_token_secret,
CONF.maas.oauth_consumer_key,
client_secret=CONF.maas.oauth_consumer_secret,
resource_owner_key=CONF.maas.oauth_token_key,
resource_owner_secret=CONF.maas.oauth_token_secret,
signature_method=oauth1.SIGNATURE_PLAINTEXT)
realm = _Realm("")
headers = client.sign(url, realm=realm)[1]
return headers
def _get_data(self, path):
norm_path = posixpath.join(CONF.maas_metadata_url, path)
norm_path = posixpath.join(CONF.maas.metadata_base_url, path)
oauth_headers = self._get_oauth_headers(norm_path)
LOG.debug('Getting metadata from: %(norm_path)s',

17
cloudbaseinit/tests/metadata/services/test_cloudstack.py
View File

@ -48,7 +48,6 @@ class CloudStackTest(unittest.TestCase):
url = '127.0.0.1'
mock_http_request.side_effect = [
'200 OK. Successfully!', # Request to Web Service
'service-offering', # Response for get_data
urllib.error.HTTPError(url=url, code=404, hdrs={}, fp=None,
msg='Testing 404 Not Found.'),
urllib.error.HTTPError(url=url, code=427, hdrs={}, fp=None,
@ -83,7 +82,8 @@ class CloudStackTest(unittest.TestCase):
self._service._test_api = mock_test_api
self.assertTrue(self._service.load())
mock_test_api.assert_called_once_with(CONF.cloudstack_metadata_ip)
mock_test_api.assert_called_once_with(
CONF.cloudstack.metadata_base_url)
@mock.patch('cloudbaseinit.osutils.factory.get_os_utils')
@mock.patch('cloudbaseinit.metadata.services.cloudstack.CloudStack'
@ -93,7 +93,8 @@ class CloudStackTest(unittest.TestCase):
mock_test_api.side_effect = [False]
self.assertFalse(self._service.load()) # No DHCP server was found.
mock_test_api.assert_called_once_with(CONF.cloudstack_metadata_ip)
mock_test_api.assert_called_once_with(
CONF.cloudstack.metadata_base_url)
@mock.patch('cloudbaseinit.osutils.factory.get_os_utils')
@mock.patch('cloudbaseinit.metadata.services.cloudstack.CloudStack'
@ -101,7 +102,7 @@ class CloudStackTest(unittest.TestCase):
def test_load_no_service(self, mock_test_api, mock_os_util):
self._service.osutils.get_dhcp_hosts_in_use = mock.Mock()
self._service.osutils.get_dhcp_hosts_in_use.side_effect = [
[(mock.sentinel.mac_address, CONF.cloudstack_metadata_ip)]
[(mock.sentinel.mac_address, CONF.cloudstack.metadata_base_url)]
]
mock_test_api.side_effect = [False, False]
@ -112,7 +113,7 @@ class CloudStackTest(unittest.TestCase):
@mock.patch('cloudbaseinit.metadata.services.cloudstack.CloudStack'
'._http_request')
def test_get_data(self, mock_http_request):
metadata = 'service-offering'
metadata = '/latest/meta-data/service-offering'
mock_http_request.side_effect = [
mock.sentinel.ok,
urllib.error.HTTPError(url=metadata, code=404, hdrs={}, fp=None,
@ -164,15 +165,15 @@ class CloudStackTest(unittest.TestCase):
def test_get_instance_id(self):
self._test_cache_response(method=self._service.get_instance_id,
metadata='instance-id')
metadata='latest/meta-data/instance-id')
def test_get_host_name(self):
self._test_cache_response(method=self._service.get_host_name,
metadata='local-hostname')
metadata='latest/meta-data/local-hostname')
def test_get_user_data(self):
self._test_cache_response(method=self._service.get_user_data,
metadata='../user-data', decode=False)
metadata='latest/user-data', decode=False)
@mock.patch('cloudbaseinit.metadata.services.cloudstack.CloudStack'
'._get_cache_data')

21
cloudbaseinit/tests/metadata/services/test_configdrive.py
View File

@ -52,24 +52,25 @@ class TestConfigDriveService(unittest.TestCase):
def _test_preprocess_options(self, fail=False):
if fail:
with testutils.ConfPatcher("config_drive_types",
["vfat", "ntfs"]):
with testutils.ConfPatcher("types", ["vfat", "ntfs"],
group="config_drive"):
with self.assertRaises(exception.CloudbaseInitException):
self._config_drive._preprocess_options()
with testutils.ConfPatcher("config_drive_locations",
["device"]):
with testutils.ConfPatcher("locations", ["device"],
group="config_drive"):
with self.assertRaises(exception.CloudbaseInitException):
self._config_drive._preprocess_options()
return
options = {
"config_drive_raw_hhd": False,
"config_drive_cdrom": False,
"config_drive_vfat": True,
"raw_hdd": False,
"cdrom": False,
"vfat": True,
# Deprecated options above.
"config_drive_types": ["vfat", "iso"],
"config_drive_locations": ["partition"]
"types": ["vfat", "iso"],
"locations": ["partition"]
}
contexts = [testutils.ConfPatcher(key, value)
contexts = [testutils.ConfPatcher(key, value, group="config_drive")
for key, value in options.items()]
with contexts[0], contexts[1], contexts[2], \
contexts[3], contexts[4]:

4
cloudbaseinit/tests/metadata/services/test_ec2service.py
View File

@ -45,7 +45,7 @@ class EC2ServiceTest(unittest.TestCase):
response = self._service.load()
mock_check_metadata_ip_route.assert_called_once_with(
CONF.ec2_metadata_base_url)
CONF.ec2.metadata_base_url)
mock_get_host_name.assert_called_once_with()
if side_effect is Exception:
self.assertFalse(response)
@ -92,7 +92,7 @@ class EC2ServiceTest(unittest.TestCase):
'._get_response')
def test_get_data(self, mock_get_response, mock_Request):
response = self._service._get_data('fake')
fake_path = posixpath.join(CONF.ec2_metadata_base_url, 'fake')
fake_path = posixpath.join(CONF.ec2.metadata_base_url, 'fake')
mock_Request.assert_called_once_with(fake_path)
mock_get_response.assert_called_once_with(mock_Request())
self.assertEqual(mock_get_response.return_value.read.return_value,

7
cloudbaseinit/tests/metadata/services/test_httpservice.py
View File

@ -41,7 +41,7 @@ class HttpServiceTest(unittest.TestCase):
mock_get_meta_data.side_effect = [side_effect]
response = self._httpservice.load()
mock_check_metadata_ip_route.assert_called_once_with(
CONF.metadata_base_url)
CONF.openstack.metadata_base_url)
mock_get_meta_data.assert_called_once_with()
if side_effect:
self.assertFalse(response)
@ -104,7 +104,8 @@ class HttpServiceTest(unittest.TestCase):
response = self._httpservice._get_data(fake_path)
mock_posix_join.assert_called_with(CONF.metadata_base_url, fake_path)
mock_posix_join.assert_called_with(CONF.openstack.metadata_base_url,
fake_path)
mock_Request.assert_called_once_with(mock_norm_path)
mock_get_response.assert_called_once_with(mock_req)
self.assertEqual(mock_data.read.return_value, response)
@ -126,7 +127,7 @@ class HttpServiceTest(unittest.TestCase):
response = self._httpservice._post_data(fake_path, fake_data)
mock_posix_join.assert_called_with(CONF.metadata_base_url,
mock_posix_join.assert_called_with(CONF.openstack.metadata_base_url,
fake_path)
mock_Request.assert_called_once_with(mock_norm_path, data=fake_data)
mock_get_response.assert_called_once_with(mock_req)

15
cloudbaseinit/tests/metadata/services/test_maasservice.py
View File

@ -43,7 +43,7 @@ class MaaSHttpServiceTest(unittest.TestCase):
if cache_data_fails:
mock_get_cache_data.side_effect = Exception
with testutils.ConfPatcher('maas_metadata_url', ip):
with testutils.ConfPatcher('metadata_base_url', ip, "maas"):
with testutils.LogSnatcher('cloudbaseinit.metadata.services.'
'maasservice') as snatcher:
response = self._maasservice.load()
@ -96,10 +96,10 @@ class MaaSHttpServiceTest(unittest.TestCase):
'test other error', {}, None)
self._test_get_response(ret_val=err)
@testutils.ConfPatcher('maas_oauth_consumer_key', 'consumer_key')
@testutils.ConfPatcher('maas_oauth_consumer_secret', 'consumer_secret')
@testutils.ConfPatcher('maas_oauth_token_key', 'token_key')
@testutils.ConfPatcher('maas_oauth_token_secret', 'token_secret')
@testutils.ConfPatcher('oauth_consumer_key', 'consumer_key', "maas")
@testutils.ConfPatcher('oauth_consumer_secret', 'consumer_secret', "maas")
@testutils.ConfPatcher('oauth_token_key', 'token_key', "maas")
@testutils.ConfPatcher('oauth_token_secret', 'token_secret', "maas")
def test_get_oauth_headers(self):
response = self._maasservice._get_oauth_headers(url='196.254.196.254')
self.assertIsInstance(response, dict)
@ -130,11 +130,12 @@ class MaaSHttpServiceTest(unittest.TestCase):
"._get_response")
def test_get_data(self, mock_get_response, mock_Request,
mock_get_oauth_headers):
with testutils.ConfPatcher('maas_metadata_url', '196.254.196.254'):
with testutils.ConfPatcher('metadata_base_url', '196.254.196.254',
'maas'):
fake_path = os.path.join('fake', 'path')
mock_get_oauth_headers.return_value = 'fake headers'
response = self._maasservice._get_data(path=fake_path)
norm_path = posixpath.join(CONF.maas_metadata_url, fake_path)
norm_path = posixpath.join(CONF.maas.metadata_base_url, fake_path)
mock_get_oauth_headers.assert_called_once_with(norm_path)
mock_Request.assert_called_once_with(norm_path,
headers='fake headers')

14
cloudbaseinit/tests/testutils.py
View File

@ -125,10 +125,14 @@ class ConfPatcher(object):
# but oslo.config.cfg doesn't support item
# assignment.
def __init__(self, key, value, conf=CONF):
self._original_value = conf.get(key)
def __init__(self, key, value, group=None, conf=CONF):
if group:
self._original_value = conf.get(group).get(key)
else:
self._original_value = conf.get(key)
self._key = key
self._value = value
self._group = group
self._conf = conf
def __call__(self, func, *args, **kwargs):
@ -140,11 +144,13 @@ class ConfPatcher(object):
return _wrapped_f
def __enter__(self):
self._conf.set_override(self._key, self._value)
self._conf.set_override(self._key, self._value,
group=self._group)
return self
def __exit__(self, exc_type, exc_val, exc_tb):
self._conf.set_override(self._key, self._original_value)
self._conf.set_override(self._key, self._original_value,
group=self._group)
class CloudbaseInitTestBase(unittest.TestCase):