openstack/charm-neutron-gateway
Code Issues Proposed changes

Replace nova vendor metadata code

Browse Source 
Two generic contexts to handle nova vendor metadata have
been implemented in charm-helpers. So, replace the existing
one here in order to simplify and unify the implementation
across all charms that handle vendor metadata.

Change-Id: I2a802c763f2f4403a6dfb17575aa742ca8072e96
Related-Bug: #1777714
This commit is contained in:
Rodrigo Barbieri 2019-05-17 10:43:36 -03:00
parent b4a468f9aa
commit 206970c6c1
11 changed files with 283 additions and 130 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
hooks/charmhelpers/contrib/openstack/cert_utils.py
View File

@ -220,6 +220,8 @@ def process_certificates(service_name, relation_id, unit,
:type user: str
:param group: (Optional) Group of certificate files. Defaults to 'root'
:type group: str
:returns: True if certificates processed for local unit or False
:rtype: bool
"""
data = relation_get(rid=relation_id, unit=unit)
ssl_dir = os.path.join('/etc/apache2/ssl/', service_name)
@ -235,6 +237,8 @@ def process_certificates(service_name, relation_id, unit,
create_ip_cert_links(
ssl_dir,
custom_hostname_link=custom_hostname_link)
return True
return False
def get_requests_for_local_unit(relation_name=None):

186
hooks/charmhelpers/contrib/openstack/context.py
View File

@ -114,9 +114,15 @@ except ImportError:
apt_install('python3-psutil', fatal=True)
import psutil
if six.PY3:
json_error = json.decoder.JSONDecodeError
else:
json_error = ValueError
CA_CERT_PATH = '/usr/local/share/ca-certificates/keystone_juju_ca_cert.crt'
ADDRESS_TYPES = ['admin', 'internal', 'public']
HAPROXY_RUN_DIR = '/var/run/haproxy/'
DEFAULT_OSLO_MESSAGING_DRIVER = "messagingv2"
def ensure_packages(packages):
@ -351,10 +357,70 @@ class IdentityServiceContext(OSContextGenerator):
return cachedir
return None
def _get_pkg_name(self, python_name='keystonemiddleware'):
"""Get corresponding distro installed package for python
package name.
:param python_name: nameof the python package
:type: string
"""
pkg_names = map(lambda x: x + python_name, ('python3-', 'python-'))
for pkg in pkg_names:
if not filter_installed_packages((pkg,)):
return pkg
return None
def _get_keystone_authtoken_ctxt(self, ctxt, keystonemiddleware_os_rel):
"""Build Jinja2 context for full rendering of [keystone_authtoken]
section with variable names included. Re-constructed from former
template 'section-keystone-auth-mitaka'.
:param ctxt: Jinja2 context returned from self.__call__()
:type: dict
:param keystonemiddleware_os_rel: OpenStack release name of
keystonemiddleware package installed
"""
c = collections.OrderedDict((('auth_type', 'password'),))
# 'www_authenticate_uri' replaced 'auth_uri' since Stein,
# see keystonemiddleware upstream sources for more info
if CompareOpenStackReleases(keystonemiddleware_os_rel) >= 'stein':
c.update((
('www_authenticate_uri', "{}://{}:{}/v3".format(
ctxt.get('service_protocol', ''),
ctxt.get('service_host', ''),
ctxt.get('service_port', ''))),))
else:
c.update((
('auth_uri', "{}://{}:{}/v3".format(
ctxt.get('service_protocol', ''),
ctxt.get('service_host', ''),
ctxt.get('service_port', ''))),))
c.update((
('auth_url', "{}://{}:{}/v3".format(
ctxt.get('auth_protocol', ''),
ctxt.get('auth_host', ''),
ctxt.get('auth_port', ''))),
('project_domain_name', ctxt.get('admin_domain_name', '')),
('user_domain_name', ctxt.get('admin_domain_name', '')),
('project_name', ctxt.get('admin_tenant_name', '')),
('username', ctxt.get('admin_user', '')),
('password', ctxt.get('admin_password', '')),
('signing_dir', ctxt.get('signing_dir', '')),))
return c
def __call__(self):
log('Generating template context for ' + self.rel_name, level=DEBUG)
ctxt = {}
keystonemiddleware_os_release = None
if self._get_pkg_name():
keystonemiddleware_os_release = os_release(self._get_pkg_name())
cachedir = self._setup_pki_cache()
if cachedir:
ctxt['signing_dir'] = cachedir
@ -385,6 +451,14 @@ class IdentityServiceContext(OSContextGenerator):
ctxt.update({'admin_domain_name':
rdata.get('service_domain')})
# we keep all veriables in ctxt for compatibility and
# add nested dictionary for keystone_authtoken generic
# templating
if keystonemiddleware_os_release:
ctxt['keystone_authtoken'] = \
self._get_keystone_authtoken_ctxt(
ctxt, keystonemiddleware_os_release)
if self.context_complete(ctxt):
# NOTE(jamespage) this is required for >= icehouse
# so a missing value just indicates keystone needs
@ -452,6 +526,80 @@ class IdentityCredentialsContext(IdentityServiceContext):
return {}
class NovaVendorMetadataContext(OSContextGenerator):
"""Context used for configuring nova vendor metadata on nova.conf file."""
def __init__(self, os_release_pkg, interfaces=None):
"""Initialize the NovaVendorMetadataContext object.
:param os_release_pkg: the package name to extract the OpenStack
release codename from.
:type os_release_pkg: str
:param interfaces: list of string values to be used as the Context's
relation interfaces.
:type interfaces: List[str]
"""
self.os_release_pkg = os_release_pkg
if interfaces is not None:
self.interfaces = interfaces
def __call__(self):
cmp_os_release = CompareOpenStackReleases(
os_release(self.os_release_pkg))
ctxt = {}
vdata_providers = []
vdata = config('vendor-data')
vdata_url = config('vendor-data-url')
if vdata:
ctxt['vendor_data'] = True
# Mitaka does not support DynamicJSON
# so vendordata_providers is not needed
if cmp_os_release > 'mitaka':
vdata_providers.append('StaticJSON')
if vdata_url:
if cmp_os_release > 'mitaka':
ctxt['vendor_data_url'] = vdata_url
vdata_providers.append('DynamicJSON')
else:
log('Dynamic vendor data unsupported'
' for {}.'.format(cmp_os_release), level=ERROR)
if vdata_providers:
ctxt['vendordata_providers'] = ','.join(vdata_providers)
return ctxt
class NovaVendorMetadataJSONContext(OSContextGenerator):
"""Context used for writing nova vendor metadata json file."""
def __init__(self, os_release_pkg):
"""Initialize the NovaVendorMetadataJSONContext object.
:param os_release_pkg: the package name to extract the OpenStack
release codename from.
:type os_release_pkg: str
"""
self.os_release_pkg = os_release_pkg
def __call__(self):
ctxt = {'vendor_data_json': '{}'}
vdata = config('vendor-data')
if vdata:
try:
# validate the JSON. If invalid, we return empty.
json.loads(vdata)
except (TypeError, json_error) as e:
log('Error decoding vendor-data. {}'.format(e), level=ERROR)
else:
ctxt['vendor_data_json'] = vdata
return ctxt
class AMQPContext(OSContextGenerator):
def __init__(self, ssl_dir=None, rel_name='amqp', relation_prefix=None,
@ -569,6 +717,19 @@ class AMQPContext(OSContextGenerator):
ctxt['oslo_messaging_flags'] = config_flags_parser(
oslo_messaging_flags)
oslo_messaging_driver = conf.get(
'oslo-messaging-driver', DEFAULT_OSLO_MESSAGING_DRIVER)
if oslo_messaging_driver:
ctxt['oslo_messaging_driver'] = oslo_messaging_driver
notification_format = conf.get('notification-format', None)
if notification_format:
ctxt['notification_format'] = notification_format
send_notifications_to_logs = conf.get('send-notifications-to-logs', None)
if send_notifications_to_logs:
ctxt['send_notifications_to_logs'] = send_notifications_to_logs
if not self.complete:
return {}
@ -1110,7 +1271,9 @@ class NeutronPortContext(OSContextGenerator):
hwaddr_to_nic = {}
hwaddr_to_ip = {}
for nic in list_nics():
extant_nics = list_nics()
for nic in extant_nics:
# Ignore virtual interfaces (bond masters will be identified from
# their slaves)
if not is_phy_iface(nic):
@ -1141,10 +1304,11 @@ class NeutronPortContext(OSContextGenerator):
# Entry is a MAC address for a valid interface that doesn't
# have an IP address assigned yet.
resolved.append(hwaddr_to_nic[entry])
else:
# If the passed entry is not a MAC address, assume it's a valid
# interface, and that the user put it there on purpose (we can
# trust it to be the real external network).
elif entry in extant_nics:
# If the passed entry is not a MAC address and the interface
# exists, assume it's a valid interface, and that the user put
# it there on purpose (we can trust it to be the real external
# network).
resolved.append(entry)
# Ensure no duplicates
@ -1526,6 +1690,14 @@ class NeutronAPIContext(OSContextGenerator):
'rel_key': 'enable-nsg-logging',
'default': False,
},
'global_physnet_mtu': {
'rel_key': 'global-physnet-mtu',
'default': 1500,
},
'physical_network_mtus': {
'rel_key': 'physical-network-mtus',
'default': None,
},
}
ctxt = self.get_neutron_options({})
for rid in relation_ids('neutron-plugin-api'):
@ -1587,13 +1759,13 @@ class DataPortContext(NeutronPortContext):
def __call__(self):
ports = config('data-port')
if ports:
# Map of {port/mac:bridge}
# Map of {bridge:port/mac}
portmap = parse_data_port_mappings(ports)
ports = portmap.keys()
# Resolve provided ports or mac addresses and filter out those
# already attached to a bridge.
resolved = self.resolve_ports(ports)
# FIXME: is this necessary?
# Rebuild port index using resolved and filtered ports.
normalized = {get_nic_hwaddr(port): port for port in resolved
if port not in ports}
normalized.update({port: port for port in resolved

9
hooks/charmhelpers/contrib/openstack/templates/section-keystone-authtoken-v3only Normal file
View File

@ -0,0 +1,9 @@
{% if auth_host -%}
[keystone_authtoken]
{% for option_name, option_value in keystone_authtoken.items() -%}
{{ option_name }} = {{ option_value }}
{% endfor -%}
{% if use_memcache == true %}
memcached_servers = {{ memcache_url }}
{% endif -%}
{% endif -%}

6
hooks/charmhelpers/contrib/openstack/templates/section-oslo-notifications
View File

@ -1,11 +1,15 @@
{% if transport_url -%}
[oslo_messaging_notifications]
driver = messagingv2
driver = {{ oslo_messaging_driver }}
transport_url = {{ transport_url }}
{% if send_notifications_to_logs %}
driver = log
{% endif %}
{% if notification_topics -%}
topics = {{ notification_topics }}
{% endif -%}
{% if notification_format -%}
[notifications]
notification_format = {{ notification_format }}
{% endif -%}
{% endif -%}

1
hooks/charmhelpers/contrib/openstack/templates/vendor_data.json Normal file
View File

@ -0,0 +1 @@
{{ vendor_data_json }}

2
hooks/charmhelpers/contrib/openstack/utils.py
View File

@ -194,7 +194,7 @@ SWIFT_CODENAMES = OrderedDict([
('rocky',
['2.18.0', '2.19.0']),
('stein',
['2.20.0']),
['2.20.0', '2.21.0']),
])
# >= Liberty version->codename mapping

39
hooks/neutron_contexts.py
View File

@ -2,7 +2,6 @@
import os
import uuid
from charmhelpers.core.hookenv import (
log, ERROR,
config,
relation_get,
relation_ids,
@ -14,6 +13,8 @@ from charmhelpers.contrib.openstack.context import (
OSContextGenerator,
NeutronAPIContext,
config_flags_parser,
NovaVendorMetadataContext,
NovaVendorMetadataJSONContext,
)
from charmhelpers.contrib.openstack.utils import (
os_release,
@ -163,32 +164,18 @@ class NeutronGatewayContext(NeutronAPIContext):
return ctxt
class NovaMetadataContext(OSContextGenerator):
class NovaMetadataContext(NovaVendorMetadataContext):
def __init__(self, rel_name='quantum-network-service'):
super(NovaMetadataContext, self).__init__('neutron-common', [rel_name])
self.rel_name = rel_name
self.interfaces = [rel_name]
def __call__(self):
ctxt = {}
cmp_os_release = CompareOpenStackReleases(os_release('neutron-common'))
if cmp_os_release < 'rocky':
vdata_providers = []
vdata = config('vendor-data')
vdata_url = config('vendor-data-url')
if vdata:
ctxt['vendor_data'] = True
vdata_providers.append('StaticJSON')
if vdata_url:
if cmp_os_release > 'mitaka':
ctxt['vendor_data_url'] = vdata_url
vdata_providers.append('DynamicJSON')
else:
log('Dynamic vendor data unsupported'
' for {}.'.format(cmp_os_release), level=ERROR)
ctxt['vendordata_providers'] = ','.join(vdata_providers)
# if release is Rocky or later, we don't set vendor metadata here
ctxt.update(super(NovaMetadataContext, self).__call__())
for rid in relation_ids(self.rel_name):
for unit in related_units(rid):
rdata = relation_get(rid=rid, unit=unit)
@ -207,6 +194,20 @@ class NovaMetadataContext(OSContextGenerator):
return ctxt
class NovaMetadataJSONContext(NovaVendorMetadataJSONContext):
def __call__(self):
vdata_values = super(NovaMetadataJSONContext, self).__call__()
cmp_os_release = CompareOpenStackReleases(os_release('neutron-common'))
if cmp_os_release < 'rocky':
return vdata_values
else:
# if release is Rocky or later, we don't set vendor metadata here
return {'vendor_data_json': '{}'}
SHARED_SECRET = "/etc/{}/secret.txt"

4
hooks/neutron_hooks.py
View File

@ -67,7 +67,6 @@ from neutron_utils import (
assess_status,
install_systemd_override,
configure_apparmor,
write_vendordata,
pause_unit_helper,
resume_unit_helper,
remove_legacy_nova_metadata,
@ -131,9 +130,6 @@ def config_changed():
create_sysctl(sysctl_settings,
'/etc/sysctl.d/50-quantum-gateway.conf')
if config('vendor-data'):
write_vendordata(config('vendor-data'))
# Re-run joined hooks as config might have changed
for r_id in relation_ids('amqp'):
amqp_joined(relation_id=r_id)

21
hooks/neutron_utils.py
View File

@ -1,5 +1,4 @@
import os
import json
import shutil
import subprocess
from shutil import copy2
@ -76,6 +75,7 @@ from neutron_contexts import (
NeutronGatewayContext,
L3AgentContext,
NovaMetadataContext,
NovaMetadataJSONContext,
)
from charmhelpers.contrib.openstack.neutron import (
parse_bridge_mappings,
@ -338,6 +338,7 @@ NEUTRON_FWAAS_CONF = "/etc/neutron/fwaas_driver.ini"
NOVA_CONF_DIR = '/etc/nova'
NOVA_CONF = "/etc/nova/nova.conf"
VENDORDATA_FILE = '%s/vendor_data.json' % NOVA_CONF_DIR
NOVA_CONFIG_FILES = {
NOVA_CONF: {
@ -356,6 +357,10 @@ NOVA_CONFIG_FILES = {
context.AppArmorContext(NOVA_API_METADATA_AA_PROFILE)
],
},
VENDORDATA_FILE: {
'services': [],
'hook_contexts': [NovaMetadataJSONContext('neutron-common')],
},
}
NEUTRON_SHARED_CONFIG_FILES = {
@ -1068,20 +1073,6 @@ def configure_apparmor():
context.AppArmorContext(profile).setup_aa_profile()
VENDORDATA_FILE = '/etc/nova/vendor_data.json'
def write_vendordata(vdata):
try:
json_vdata = json.loads(vdata)
except (TypeError, json.decoder.JSONDecodeError) as e:
log('Error decoding vendor-data. {}'.format(e), level=ERROR)
return False
with open(VENDORDATA_FILE, 'w') as vdata_file:
vdata_file.write(json.dumps(json_vdata, sort_keys=True, indent=2))
return True
def get_availability_zone():
use_juju_az = config('customize-failure-domain')
juju_az = os.environ.get('JUJU_AVAILABILITY_ZONE')

122
unit_tests/test_neutron_contexts.py
View File

@ -453,88 +453,83 @@ class TestNovaMetadataContext(CharmTestCase):
TO_PATCH)
self.config.side_effect = self.test_config.get
@patch.object(neutron_contexts.NovaVendorMetadataContext, '__call__')
@patch.object(neutron_contexts, 'get_local_ip')
@patch.object(neutron_contexts, 'get_shared_secret')
@patch.object(neutron_contexts, 'relation_ids')
def test_vendordata_static(self, _relation_ids, _get_shared_secret,
_get_local_ip):
def test_vendordata_queens(self, _relation_ids, _get_shared_secret,
_get_local_ip, parent):
_get_shared_secret.return_value = 'asecret'
_get_local_ip.return_value = '127.0.0.1'
_relation_ids.return_value = []
_vdata = '{"good": "json"}'
self.os_release.return_value = 'pike'
_vdata_url = 'http://example.org/vdata'
_vdata_providers = 'StaticJSON,DynamicJSON'
self.os_release.return_value = 'queens'
parent.return_value = {
'vendor_data': True,
'vendor_data_url': _vdata_url,
'vendordata_providers': _vdata_providers,
}
self.test_config.set('vendor-data', _vdata)
ctxt = neutron_contexts.NovaMetadataContext()()
self.assertTrue(ctxt['vendor_data'])
self.assertEqual(ctxt['vendordata_providers'], 'StaticJSON')
@patch.object(neutron_contexts, 'get_local_ip')
@patch.object(neutron_contexts, 'get_shared_secret')
@patch.object(neutron_contexts, 'relation_ids')
def test_vendordata_dynamic(self, _relation_ids, _get_shared_secret,
_get_local_ip):
_get_shared_secret.return_value = 'asecret'
_get_local_ip.return_value = '127.0.0.1'
_relation_ids.return_value = []
_vdata_url = 'http://example.org/vdata'
self.os_release.return_value = 'pike'
self.test_config.set('vendor-data-url', _vdata_url)
ctxt = neutron_contexts.NovaMetadataContext()()
self.assertEqual(ctxt['vendordata_providers'], _vdata_providers)
self.assertEqual(ctxt['vendor_data_url'], _vdata_url)
self.assertEqual(ctxt['vendordata_providers'], 'DynamicJSON')
@patch.object(neutron_contexts.NovaVendorMetadataContext, '__call__')
@patch.object(neutron_contexts, 'get_local_ip')
@patch.object(neutron_contexts, 'get_shared_secret')
@patch.object(neutron_contexts, 'relation_ids')
def test_vendordata_static_and_dynamic(self, _relation_ids,
_get_shared_secret, _get_local_ip):
def test_vendordata_rocky(self, _relation_ids, _get_shared_secret,
_get_local_ip, parent):
_get_shared_secret.return_value = 'asecret'
_get_local_ip.return_value = '127.0.0.1'
_relation_ids.return_value = []
_vdata = '{"good": "json"}'
_vdata_url = 'http://example.org/vdata'
self.os_release.return_value = 'pike'
self.os_release.return_value = 'rocky'
parent.return_value = {
'vendor_data': True,
'vendor_data_url': 'http://example.org/vdata',
'vendordata_providers': 'StaticJSON,DynamicJSON',
}
self.test_config.set('vendor-data', _vdata)
self.test_config.set('vendor-data-url', _vdata_url)
ctxt = neutron_contexts.NovaMetadataContext()()
self.assertTrue(ctxt['vendor_data'])
self.assertEqual(ctxt['vendor_data_url'], _vdata_url)
self.assertEqual(ctxt['vendordata_providers'],
'StaticJSON,DynamicJSON')
self.assertNotIn('vendor_data', ctxt)
self.assertNotIn('vendor_data_url', ctxt)
self.assertNotIn('vendordata_providers', ctxt)
@patch.object(neutron_contexts, 'get_local_ip')
@patch.object(neutron_contexts, 'get_shared_secret')
@patch.object(neutron_contexts, 'relation_ids')
def test_vendordata_mitaka(self, _relation_ids, _get_shared_secret,
_get_local_ip):
_get_shared_secret.return_value = 'asecret'
_get_local_ip.return_value = '127.0.0.1'
_relation_ids.return_value = []
_vdata_url = 'http://example.org/vdata'
self.os_release.return_value = 'mitaka'
@patch.object(neutron_contexts.NovaVendorMetadataJSONContext, '__call__')
def test_vendordata_json_queens(self, parent):
self.os_release.return_value = 'queens'
result = {
'vendor_data_json': '{"good": "json"}',
}
parent.return_value = result
self.test_config.set('vendor-data-url', _vdata_url)
ctxt = neutron_contexts.NovaMetadataContext()()
self.assertEqual(
ctxt,
{
'nova_metadata_host': '127.0.0.1',
'nova_metadata_port': '8775',
'nova_metadata_protocol': 'http',
'shared_secret': 'asecret',
'vendordata_providers': ''})
ctxt = neutron_contexts.NovaMetadataJSONContext('neutron-common')()
self.assertEqual(result, ctxt)
@patch.object(neutron_contexts.NovaVendorMetadataJSONContext, '__call__')
def test_vendordata_json_rocky(self, parent):
self.os_release.return_value = 'rocky'
result = {
'vendor_data_json': '{}',
}
parent.return_value = {
'vendor_data_json': '{"good": "json"}',
}
ctxt = neutron_contexts.NovaMetadataJSONContext('neutron-common')()
self.assertEqual(result, ctxt)
@patch.object(neutron_contexts, 'relation_get')
@patch.object(neutron_contexts, 'related_units')
@patch.object(neutron_contexts, 'relation_ids')
def test_NovaMetadataContext(self, _relation_ids, _related_units,
_relation_get):
def test_NovaMetadataContext_with_relations(self, _relation_ids,
_related_units, _relation_get):
_relation_ids.return_value = ['rid:1']
_related_units.return_value = ['nova-cloud-contoller/0']
_relation_get.return_value = {
@ -542,7 +537,7 @@ class TestNovaMetadataContext(CharmTestCase):
'nova-metadata-port': '8775',
'nova-metadata-protocol': 'http',
'shared-metadata-secret': 'auuid'}
self.os_release.return_value = 'queens'
self.os_release.return_value = 'rocky'
self.assertEqual(
neutron_contexts.NovaMetadataContext()(),
@ -550,24 +545,23 @@ class TestNovaMetadataContext(CharmTestCase):
'nova_metadata_host': '10.0.0.10',
'nova_metadata_port': '8775',
'nova_metadata_protocol': 'http',
'shared_secret': 'auuid',
'vendordata_providers': ''})
'shared_secret': 'auuid'})
@patch.object(neutron_contexts, 'get_local_ip')
@patch.object(neutron_contexts, 'get_shared_secret')
@patch.object(neutron_contexts, 'relation_ids')
def test_NovaMetadataContext_legacy(self, _relation_ids,
_get_shared_secret,
_get_local_ip):
def test_NovaMetadataContext_no_relations(self, _relation_ids,
_get_shared_secret,
_get_local_ip):
_relation_ids.return_value = []
_get_shared_secret.return_value = 'buuid'
_get_local_ip.return_value = '127.0.0.1'
self.os_release.return_value = 'pike'
self.os_release.return_value = 'rocky'
self.assertEqual(
neutron_contexts.NovaMetadataContext()(),
{
'nova_metadata_host': '127.0.0.1',
'nova_metadata_port': '8775',
'nova_metadata_protocol': 'http',
'shared_secret': 'buuid',
'vendordata_providers': ''})
'shared_secret': 'buuid'})

19
unit_tests/test_neutron_utils.py
View File

@ -11,10 +11,6 @@ from test_utils import (
CharmTestCase
)
from test_neutron_contexts import (
patch_open
)
TO_PATCH = [
'config',
'get_os_codename_install_source',
@ -855,21 +851,6 @@ class TestNeutronUtils(CharmTestCase):
for config in EXC_CONFIG:
self.assertTrue(config not in actual_configs)
def test_write_valid_json_vendordata(self):
_jdata = '{"good": "json"}'
_tdata = '{\n "good": "json"\n}'
with patch_open() as (_open, _file):
self.assertEqual(neutron_utils.write_vendordata(_jdata), True)
_open.assert_called_with(neutron_utils.VENDORDATA_FILE, 'w')
_file.write.assert_called_with(_tdata)
@patch('json.loads')
def test_write_invalid_json_vendordata(self, _json_loads):
_jdata = '{ bad json }'
_json_loads.side_effect = TypeError
with patch_open() as (_open, _file):
self.assertEqual(neutron_utils.write_vendordata(_jdata), False)
@patch.object(neutron_utils.os.environ, 'get')
def test_get_az_customize_with_env(self, os_environ_get_mock):
self.config.side_effect = self.test_config.get