631 lines
26 KiB
Python
631 lines
26 KiB
Python
import io
|
|
|
|
from contextlib import contextmanager
|
|
|
|
from unittest.mock import (
|
|
MagicMock,
|
|
patch
|
|
)
|
|
import neutron_contexts
|
|
|
|
from test_utils import (
|
|
CharmTestCase
|
|
)
|
|
|
|
TO_PATCH = [
|
|
'config',
|
|
'eligible_leader',
|
|
'unit_get',
|
|
'network_get_primary_address',
|
|
'os_release',
|
|
'charmhelpers.contrib.network.ip.log',
|
|
]
|
|
|
|
|
|
@contextmanager
|
|
def patch_open():
|
|
'''Patch open() to allow mocking both open() itself and the file that is
|
|
yielded.
|
|
|
|
Yields the mock for "open" and "file", respectively.'''
|
|
mock_open = MagicMock(spec=open)
|
|
mock_file = MagicMock(spec=io.FileIO)
|
|
|
|
@contextmanager
|
|
def stub_open(*args, **kwargs):
|
|
mock_open(*args, **kwargs)
|
|
yield mock_file
|
|
|
|
with patch('builtins.open', stub_open):
|
|
yield mock_open, mock_file
|
|
|
|
|
|
class DummyNeutronAPIContext():
|
|
|
|
def __init__(self, return_value):
|
|
self.return_value = return_value
|
|
|
|
def __call__(self):
|
|
return self.return_value
|
|
|
|
|
|
class TestL3AgentContext(CharmTestCase):
|
|
|
|
def setUp(self):
|
|
super(TestL3AgentContext, self).setUp(neutron_contexts,
|
|
TO_PATCH)
|
|
self.network_get_primary_address.side_effect = NotImplementedError
|
|
self.config.side_effect = self.test_config.get
|
|
|
|
@patch('neutron_contexts.NeutronAPIContext')
|
|
def test_new_ext_network(self, _NeutronAPIContext):
|
|
self.os_release.return_value = 'stein'
|
|
_NeutronAPIContext.return_value = \
|
|
DummyNeutronAPIContext(return_value={
|
|
'enable_dvr': False,
|
|
'report_interval': 30,
|
|
'rpc_response_timeout': 60,
|
|
'enable_l3ha': True,
|
|
'enable_fwaas': True,
|
|
'l3_extension_plugins': ['fwaas_v2'],
|
|
})
|
|
self.test_config.set('run-internal-router', 'none')
|
|
self.test_config.set('external-network-id', '')
|
|
self.eligible_leader.return_value = False
|
|
self.assertEqual(neutron_contexts.L3AgentContext()(),
|
|
{'agent_mode': 'legacy',
|
|
'report_interval': 30,
|
|
'rpc_response_timeout': 60,
|
|
'use_l3ha': True,
|
|
'external_configuration_new': True,
|
|
'handle_internal_only_router': False,
|
|
'plugin': 'ovs',
|
|
'l3_extension_plugins': 'fwaas_v2',
|
|
})
|
|
|
|
@patch('neutron_contexts.NeutronAPIContext')
|
|
def test_old_ext_network(self, _NeutronAPIContext):
|
|
self.os_release.return_value = 'rocky'
|
|
_NeutronAPIContext.return_value = \
|
|
DummyNeutronAPIContext(return_value={'enable_dvr': False,
|
|
'report_interval': 30,
|
|
'rpc_response_timeout': 60,
|
|
'enable_l3ha': True,
|
|
})
|
|
self.test_config.set('run-internal-router', 'none')
|
|
self.test_config.set('ext-port', 'eth1')
|
|
self.eligible_leader.return_value = False
|
|
self.assertEqual(neutron_contexts.L3AgentContext()(),
|
|
{'agent_mode': 'legacy',
|
|
'report_interval': 30,
|
|
'rpc_response_timeout': 60,
|
|
'use_l3ha': True,
|
|
'handle_internal_only_router': False,
|
|
'plugin': 'ovs',
|
|
'l3_extension_plugins': '',
|
|
})
|
|
|
|
@patch('neutron_contexts.NeutronAPIContext')
|
|
def test_hior_leader(self, _NeutronAPIContext):
|
|
self.os_release.return_value = 'rocky'
|
|
_NeutronAPIContext.return_value = \
|
|
DummyNeutronAPIContext(return_value={'enable_dvr': False,
|
|
'report_interval': 30,
|
|
'rpc_response_timeout': 60,
|
|
'enable_l3ha': True,
|
|
'l3_extension_plugins': '',
|
|
})
|
|
self.test_config.set('run-internal-router', 'leader')
|
|
self.test_config.set('external-network-id', 'netid')
|
|
self.eligible_leader.return_value = True
|
|
self.assertEqual(neutron_contexts.L3AgentContext()(),
|
|
{'agent_mode': 'legacy',
|
|
'report_interval': 30,
|
|
'rpc_response_timeout': 60,
|
|
'use_l3ha': True,
|
|
'handle_internal_only_router': True,
|
|
'ext_net_id': 'netid',
|
|
'plugin': 'ovs',
|
|
'l3_extension_plugins': '',
|
|
})
|
|
|
|
@patch('neutron_contexts.NeutronAPIContext')
|
|
def test_hior_all(self, _NeutronAPIContext):
|
|
self.os_release.return_value = 'rocky'
|
|
_NeutronAPIContext.return_value = \
|
|
DummyNeutronAPIContext(return_value={'enable_dvr': False,
|
|
'report_interval': 30,
|
|
'rpc_response_timeout': 60,
|
|
'enable_l3ha': True,
|
|
})
|
|
self.test_config.set('run-internal-router', 'all')
|
|
self.test_config.set('external-network-id', 'netid')
|
|
self.eligible_leader.return_value = True
|
|
self.assertEqual(neutron_contexts.L3AgentContext()(),
|
|
{'agent_mode': 'legacy',
|
|
'report_interval': 30,
|
|
'rpc_response_timeout': 60,
|
|
'use_l3ha': True,
|
|
'handle_internal_only_router': True,
|
|
'ext_net_id': 'netid',
|
|
'plugin': 'ovs',
|
|
'l3_extension_plugins': '',
|
|
})
|
|
|
|
@patch('neutron_contexts.NeutronAPIContext')
|
|
def test_dvr(self, _NeutronAPIContext):
|
|
self.os_release.return_value = 'rocky'
|
|
_NeutronAPIContext.return_value = \
|
|
DummyNeutronAPIContext(return_value={'enable_dvr': True,
|
|
'report_interval': 30,
|
|
'rpc_response_timeout': 60,
|
|
'enable_l3ha': True,
|
|
})
|
|
self.assertEqual(neutron_contexts.L3AgentContext()()['agent_mode'],
|
|
'dvr_snat')
|
|
|
|
|
|
class TestNeutronGatewayContext(CharmTestCase):
|
|
|
|
def setUp(self):
|
|
super(TestNeutronGatewayContext, self).setUp(neutron_contexts,
|
|
TO_PATCH)
|
|
self.config.side_effect = self.test_config.get
|
|
self.maxDiff = None
|
|
|
|
@patch.object(neutron_contexts, 'validate_nfg_log_path', lambda x: x)
|
|
@patch('charmhelpers.contrib.openstack.context.relation_get')
|
|
@patch('charmhelpers.contrib.openstack.context.related_units')
|
|
@patch('charmhelpers.contrib.openstack.context.relation_ids')
|
|
@patch.object(neutron_contexts, 'get_shared_secret')
|
|
def test_all(self, _secret, _rids, _runits, _rget):
|
|
rdata = {'l2-population': 'True',
|
|
'enable-dvr': 'True',
|
|
'overlay-network-type': 'gre',
|
|
'enable-l3ha': 'True',
|
|
'enable-qos': 'True',
|
|
'network-device-mtu': 9000,
|
|
'dns-domain': 'openstack.example.',
|
|
'enable-nfg-logging': 'True'}
|
|
self.test_config.set('plugin', 'ovs')
|
|
self.test_config.set('debug', False)
|
|
self.test_config.set('verbose', True)
|
|
self.test_config.set('instance-mtu', 1420)
|
|
self.test_config.set('dnsmasq-flags', 'dhcp-userclass=set:ipxe,iPXE,'
|
|
'dhcp-match=set:ipxe,175')
|
|
self.test_config.set('dns-servers', '8.8.8.8,4.4.4.4')
|
|
self.test_config.set('vlan-ranges',
|
|
'physnet1:1000:2000 physnet2:2001:3000')
|
|
self.test_config.set('flat-network-providers', 'physnet3 physnet4')
|
|
self.test_config.set('firewall-group-log-output-base',
|
|
'/var/log/firewall-logs')
|
|
self.test_config.set('firewall-group-log-rate-limit', 100)
|
|
self.test_config.set('firewall-group-log-burst-limit', 50)
|
|
|
|
self.test_config.set('customize-failure-domain', False)
|
|
self.test_config.set('default-availability-zone', 'nova')
|
|
|
|
self.test_config.set('ovsdb-timeout', 10)
|
|
|
|
self.test_config.set('keepalived-healthcheck-interval', 0)
|
|
|
|
self.network_get_primary_address.side_effect = NotImplementedError
|
|
self.unit_get.return_value = '10.5.0.1'
|
|
# Provided by neutron-api relation
|
|
_rids.return_value = ['neutron-plugin-api:0']
|
|
_runits.return_value = ['neutron-api/0']
|
|
_rget.side_effect = lambda *args, **kwargs: rdata
|
|
_secret.return_value = 'testsecret'
|
|
ctxt = neutron_contexts.NeutronGatewayContext()()
|
|
self.assertEqual(ctxt, {
|
|
'shared_secret': 'testsecret',
|
|
'enable_dvr': True,
|
|
'enable_l3ha': True,
|
|
'extension_drivers': 'qos',
|
|
'local_ip': '10.5.0.1',
|
|
'core_plugin': "ml2",
|
|
'plugin': 'ovs',
|
|
'debug': False,
|
|
'verbose': True,
|
|
'l2_population': True,
|
|
'overlay_network_type': 'gre',
|
|
'report_interval': 30,
|
|
'rpc_response_timeout': 60,
|
|
'bridge_mappings': 'physnet1:br-data',
|
|
'network_providers': 'physnet3,physnet4',
|
|
'vlan_ranges': 'physnet1:1000:2000,physnet2:2001:3000',
|
|
'network_device_mtu': 9000,
|
|
'veth_mtu': 9000,
|
|
'availability_zone': 'nova',
|
|
'enable_nfg_logging': True,
|
|
'nfg_log_burst_limit': 50,
|
|
'nfg_log_output_base': '/var/log/firewall-logs',
|
|
'nfg_log_rate_limit': 100,
|
|
'ovsdb_timeout': 10,
|
|
'keepalived_healthcheck_interval': 0,
|
|
'firewall_driver': "iptables_hybrid",
|
|
})
|
|
|
|
@patch.object(neutron_contexts, 'validate_nfg_log_path', lambda x: x)
|
|
@patch('charmhelpers.contrib.openstack.context.relation_get')
|
|
@patch('charmhelpers.contrib.openstack.context.related_units')
|
|
@patch('charmhelpers.contrib.openstack.context.relation_ids')
|
|
@patch.object(neutron_contexts, 'get_shared_secret')
|
|
def test_all_network_spaces(self, _secret, _rids, _runits, _rget):
|
|
rdata = {'l2-population': 'True',
|
|
'enable-dvr': 'True',
|
|
'overlay-network-type': 'gre',
|
|
'enable-l3ha': 'True',
|
|
'enable-qos': 'True',
|
|
'network-device-mtu': 9000,
|
|
'dns-domain': 'openstack.example.'}
|
|
self.test_config.set('plugin', 'ovs')
|
|
self.test_config.set('debug', False)
|
|
self.test_config.set('verbose', True)
|
|
self.test_config.set('instance-mtu', 1420)
|
|
self.test_config.set('dnsmasq-flags', 'dhcp-userclass=set:ipxe,iPXE,'
|
|
'dhcp-match=set:ipxe,175')
|
|
self.test_config.set('vlan-ranges',
|
|
'physnet1:1000:2000 physnet2:2001:3000')
|
|
self.test_config.set('flat-network-providers', 'physnet3 physnet4')
|
|
|
|
self.test_config.set('customize-failure-domain', False)
|
|
self.test_config.set('default-availability-zone', 'nova')
|
|
self.test_config.set('ovsdb-timeout', 60)
|
|
|
|
self.network_get_primary_address.return_value = '192.168.20.2'
|
|
self.unit_get.return_value = '10.5.0.1'
|
|
# Provided by neutron-api relation
|
|
_rids.return_value = ['neutron-plugin-api:0']
|
|
_runits.return_value = ['neutron-api/0']
|
|
_rget.side_effect = lambda *args, **kwargs: rdata
|
|
_secret.return_value = 'testsecret'
|
|
ctxt = neutron_contexts.NeutronGatewayContext()()
|
|
self.assertEqual(ctxt, {
|
|
'shared_secret': 'testsecret',
|
|
'enable_dvr': True,
|
|
'enable_l3ha': True,
|
|
'extension_drivers': 'qos',
|
|
'local_ip': '192.168.20.2',
|
|
'core_plugin': "ml2",
|
|
'plugin': 'ovs',
|
|
'debug': False,
|
|
'verbose': True,
|
|
'l2_population': True,
|
|
'overlay_network_type': 'gre',
|
|
'report_interval': 30,
|
|
'rpc_response_timeout': 60,
|
|
'bridge_mappings': 'physnet1:br-data',
|
|
'network_providers': 'physnet3,physnet4',
|
|
'vlan_ranges': 'physnet1:1000:2000,physnet2:2001:3000',
|
|
'network_device_mtu': 9000,
|
|
'veth_mtu': 9000,
|
|
'availability_zone': 'nova',
|
|
'enable_nfg_logging': False,
|
|
'nfg_log_burst_limit': 25,
|
|
'nfg_log_output_base': None,
|
|
'nfg_log_rate_limit': None,
|
|
'ovsdb_timeout': 60,
|
|
'keepalived_healthcheck_interval': 0,
|
|
'firewall_driver': "iptables_hybrid",
|
|
})
|
|
|
|
@patch('os.environ.get')
|
|
@patch('charmhelpers.contrib.openstack.context.relation_get')
|
|
@patch('charmhelpers.contrib.openstack.context.related_units')
|
|
@patch('charmhelpers.contrib.openstack.context.relation_ids')
|
|
@patch.object(neutron_contexts, 'get_shared_secret')
|
|
def test_availability_zone_no_juju_with_env(self, _secret, _rids,
|
|
_runits, _rget,
|
|
mock_get):
|
|
self.os_release.return_value = 'icehouse'
|
|
|
|
def environ_get_side_effect(key):
|
|
return {
|
|
'JUJU_AVAILABILITY_ZONE': 'az1',
|
|
'PATH': 'foobar',
|
|
}[key]
|
|
mock_get.side_effect = environ_get_side_effect
|
|
|
|
self.test_config.set('customize-failure-domain', False)
|
|
self.test_config.set('default-availability-zone', 'nova')
|
|
|
|
context = neutron_contexts.NeutronGatewayContext()
|
|
self.assertEqual(
|
|
'nova', context()['availability_zone'])
|
|
|
|
@patch('neutron_utils.config')
|
|
@patch('os.environ.get')
|
|
@patch('charmhelpers.contrib.openstack.context.relation_get')
|
|
@patch('charmhelpers.contrib.openstack.context.related_units')
|
|
@patch('charmhelpers.contrib.openstack.context.relation_ids')
|
|
@patch.object(neutron_contexts, 'get_shared_secret')
|
|
def test_availability_zone_no_juju_no_env(self, _secret, _rids,
|
|
_runits, _rget,
|
|
mock_get, mock_config):
|
|
self.os_release.return_value = 'icehouse'
|
|
|
|
def environ_get_side_effect(key):
|
|
return {
|
|
'JUJU_AVAILABILITY_ZONE': '',
|
|
'PATH': 'foobar',
|
|
}[key]
|
|
mock_get.side_effect = environ_get_side_effect
|
|
|
|
def config_side_effect(key):
|
|
return {
|
|
'customize-failure-domain': False,
|
|
'default-availability-zone': 'nova',
|
|
}[key]
|
|
|
|
mock_config.side_effect = config_side_effect
|
|
context = neutron_contexts.NeutronGatewayContext()
|
|
|
|
self.assertEqual(
|
|
'nova', context()['availability_zone'])
|
|
|
|
@patch('neutron_utils.config')
|
|
@patch('os.environ.get')
|
|
@patch('charmhelpers.contrib.openstack.context.relation_get')
|
|
@patch('charmhelpers.contrib.openstack.context.related_units')
|
|
@patch('charmhelpers.contrib.openstack.context.relation_ids')
|
|
@patch.object(neutron_contexts, 'get_shared_secret')
|
|
def test_availability_zone_juju(self, _secret, _rids,
|
|
_runits, _rget,
|
|
mock_get, mock_config):
|
|
self.os_release.return_value = 'icehouse'
|
|
|
|
def environ_get_side_effect(key):
|
|
return {
|
|
'JUJU_AVAILABILITY_ZONE': 'az1',
|
|
'PATH': 'foobar',
|
|
}[key]
|
|
mock_get.side_effect = environ_get_side_effect
|
|
|
|
mock_config.side_effect = self.test_config.get
|
|
self.test_config.set('customize-failure-domain', True)
|
|
context = neutron_contexts.NeutronGatewayContext()
|
|
self.assertEqual(
|
|
'az1', context()['availability_zone'])
|
|
|
|
@patch('charmhelpers.contrib.openstack.context.relation_get')
|
|
@patch('charmhelpers.contrib.openstack.context.related_units')
|
|
@patch('charmhelpers.contrib.openstack.context.relation_ids')
|
|
@patch.object(neutron_contexts, 'get_shared_secret')
|
|
def test_nfg_min_settings(self, _secret, _rids, _runits, _rget):
|
|
self.os_release.return_value = 'icehouse'
|
|
self.test_config.set('firewall-group-log-rate-limit', 90)
|
|
self.test_config.set('firewall-group-log-burst-limit', 20)
|
|
self.network_get_primary_address.return_value = '192.168.20.2'
|
|
self.unit_get.return_value = '10.5.0.1'
|
|
ctxt = neutron_contexts.NeutronGatewayContext()()
|
|
self.assertEqual(ctxt['nfg_log_burst_limit'], 25)
|
|
self.assertEqual(ctxt['nfg_log_rate_limit'], 100)
|
|
|
|
|
|
class TestSharedSecret(CharmTestCase):
|
|
|
|
def setUp(self):
|
|
super(TestSharedSecret, self).setUp(neutron_contexts,
|
|
TO_PATCH)
|
|
self.config.side_effect = self.test_config.get
|
|
self.network_get_primary_address.side_effect = NotImplementedError
|
|
|
|
@patch('os.path')
|
|
@patch('uuid.uuid4')
|
|
def test_secret_created_stored(self, _uuid4, _path):
|
|
_path.exists.return_value = False
|
|
_uuid4.return_value = 'secret_thing'
|
|
with patch_open() as (_open, _file):
|
|
self.assertEqual(neutron_contexts.get_shared_secret(),
|
|
'secret_thing')
|
|
_open.assert_called_with(
|
|
neutron_contexts.SHARED_SECRET.format('neutron'), 'w')
|
|
_file.write.assert_called_with('secret_thing')
|
|
|
|
@patch('os.path')
|
|
def test_secret_retrieved(self, _path):
|
|
_path.exists.return_value = True
|
|
with patch_open() as (_open, _file):
|
|
_file.read.return_value = 'secret_thing\n'
|
|
self.assertEqual(neutron_contexts.get_shared_secret(),
|
|
'secret_thing')
|
|
_open.assert_called_with(
|
|
neutron_contexts.SHARED_SECRET.format('neutron'), 'r')
|
|
|
|
|
|
class TestMisc(CharmTestCase):
|
|
|
|
def setUp(self):
|
|
super(TestMisc,
|
|
self).setUp(neutron_contexts,
|
|
TO_PATCH)
|
|
|
|
def test_core_plugin_ml2(self):
|
|
self.config.return_value = 'ovs'
|
|
self.assertEqual(neutron_contexts.core_plugin(),
|
|
neutron_contexts.NEUTRON_ML2_PLUGIN)
|
|
|
|
|
|
class TestNovaMetadataContext(CharmTestCase):
|
|
|
|
def setUp(self):
|
|
super(TestNovaMetadataContext, self).setUp(neutron_contexts,
|
|
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_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_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,
|
|
}
|
|
|
|
ctxt = neutron_contexts.NovaMetadataContext()()
|
|
|
|
self.assertTrue(ctxt['vendor_data'])
|
|
self.assertEqual(ctxt['vendordata_providers'], _vdata_providers)
|
|
self.assertEqual(ctxt['vendor_data_url'], _vdata_url)
|
|
|
|
@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_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 = []
|
|
self.os_release.return_value = 'rocky'
|
|
parent.return_value = {
|
|
'vendor_data': True,
|
|
'vendor_data_url': 'http://example.org/vdata',
|
|
'vendordata_providers': 'StaticJSON,DynamicJSON',
|
|
}
|
|
|
|
ctxt = neutron_contexts.NovaMetadataContext()()
|
|
|
|
self.assertNotIn('vendor_data', ctxt)
|
|
self.assertNotIn('vendor_data_url', ctxt)
|
|
self.assertNotIn('vendordata_providers', ctxt)
|
|
|
|
@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
|
|
|
|
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_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 = {
|
|
'nova-metadata-host': '10.0.0.10',
|
|
'nova-metadata-port': '8775',
|
|
'nova-metadata-protocol': 'http',
|
|
'shared-metadata-secret': 'auuid'}
|
|
self.os_release.return_value = 'rocky'
|
|
|
|
self.assertEqual(
|
|
neutron_contexts.NovaMetadataContext()(),
|
|
{
|
|
'nova_metadata_host': '10.0.0.10',
|
|
'nova_metadata_port': '8775',
|
|
'nova_metadata_protocol': 'http',
|
|
'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_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 = 'rocky'
|
|
|
|
self.assertEqual(
|
|
neutron_contexts.NovaMetadataContext()(),
|
|
{
|
|
'nova_metadata_host': '127.0.0.1',
|
|
'nova_metadata_port': '8775',
|
|
'nova_metadata_protocol': 'http',
|
|
'shared_secret': 'buuid'})
|
|
|
|
|
|
class TestGetAvailabilityZone(CharmTestCase):
|
|
|
|
def setUp(self):
|
|
super(TestGetAvailabilityZone, self).setUp(neutron_contexts, TO_PATCH)
|
|
|
|
@patch.object(neutron_contexts.os.environ, 'get')
|
|
def test_get_az_customize_with_env(self, os_environ_get_mock):
|
|
self.config.side_effect = self.test_config.get
|
|
self.test_config.set('customize-failure-domain', True)
|
|
self.test_config.set('default-availability-zone', 'nova')
|
|
|
|
def os_environ_get_side_effect(key):
|
|
return {
|
|
'JUJU_AVAILABILITY_ZONE': 'az1',
|
|
}[key]
|
|
os_environ_get_mock.side_effect = os_environ_get_side_effect
|
|
az = neutron_contexts.get_availability_zone()
|
|
self.assertEqual('az1', az)
|
|
|
|
@patch.object(neutron_contexts.os.environ, 'get')
|
|
def test_get_az_customize_without_env(self, os_environ_get_mock):
|
|
self.config.side_effect = self.test_config.get
|
|
self.test_config.set('customize-failure-domain', True)
|
|
self.test_config.set('default-availability-zone', 'mynova')
|
|
|
|
def os_environ_get_side_effect(key):
|
|
return {
|
|
'JUJU_AVAILABILITY_ZONE': '',
|
|
}[key]
|
|
os_environ_get_mock.side_effect = os_environ_get_side_effect
|
|
az = neutron_contexts.get_availability_zone()
|
|
self.assertEqual('mynova', az)
|
|
|
|
@patch.object(neutron_contexts.os.environ, 'get')
|
|
def test_get_az_no_customize_without_env(self, os_environ_get_mock):
|
|
self.config.side_effect = self.test_config.get
|
|
self.test_config.set('customize-failure-domain', False)
|
|
self.test_config.set('default-availability-zone', 'nova')
|
|
|
|
def os_environ_get_side_effect(key):
|
|
return {
|
|
'JUJU_AVAILABILITY_ZONE': '',
|
|
}[key]
|
|
os_environ_get_mock.side_effect = os_environ_get_side_effect
|
|
az = neutron_contexts.get_availability_zone()
|
|
self.assertEqual('nova', az)
|
|
|
|
@patch.object(neutron_contexts.os.environ, 'get')
|
|
def test_get_az_no_customize_with_env(self, os_environ_get_mock):
|
|
self.config.side_effect = self.test_config.get
|
|
self.test_config.set('customize-failure-domain', False)
|
|
self.test_config.set('default-availability-zone', 'nova')
|
|
|
|
def os_environ_get_side_effect(key):
|
|
return {
|
|
'JUJU_AVAILABILITY_ZONE': 'az1',
|
|
}[key]
|
|
os_environ_get_mock.side_effect = os_environ_get_side_effect
|
|
az = neutron_contexts.get_availability_zone()
|
|
self.assertEqual('nova', az)
|