# Copyright 2016 Canonical Ltd # # 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 json from mock import MagicMock, patch import neutron_api_context as context import charmhelpers from test_utils import CharmTestCase TO_PATCH = [ 'config', 'determine_api_port', 'determine_apache_port', 'log', 'os_release', 'relation_get', 'relation_ids', 'related_units', ] class GeneralTests(CharmTestCase): def setUp(self): super(GeneralTests, self).setUp(context, TO_PATCH) self.relation_get.side_effect = self.test_relation.get self.config.side_effect = self.test_config.get def test_l2population(self): self.test_config.set('l2-population', True) self.test_config.set('neutron-plugin', 'ovs') self.assertEqual(context.get_l2population(), True) def test_l2population_nonovs(self): self.test_config.set('l2-population', True) self.test_config.set('neutron-plugin', 'nsx') self.assertEqual(context.get_l2population(), False) def test_get_tenant_network_types(self): self.test_config.set('overlay-network-type', 'gre') self.assertEqual( context._get_tenant_network_types(), ['gre', 'vlan', 'flat', 'local']) def test_get_tenant_network_types_multi(self): self.test_config.set('overlay-network-type', 'gre vxlan') self.assertEqual( context._get_tenant_network_types(), ['gre', 'vxlan', 'vlan', 'flat', 'local']) def test_get_tenant_network_types_unsupported(self): self.test_config.set('overlay-network-type', 'tokenring') with self.assertRaises(ValueError): context._get_tenant_network_types() def test_get_tenant_network_types_default(self): self.test_config.set('overlay-network-type', 'gre vxlan') self.test_config.set('default-tenant-network-type', 'vxlan') self.assertEqual( context._get_tenant_network_types(), ['vxlan', 'gre', 'vlan', 'flat', 'local']) def test_get_tenant_network_types_default_dup(self): self.test_config.set('overlay-network-type', 'gre') self.test_config.set('default-tenant-network-type', 'vlan') self.assertEqual( context._get_tenant_network_types(), ['vlan', 'gre', 'flat', 'local']) def test_get_tenant_network_types_empty(self): self.test_config.set('overlay-network-type', '') self.test_config.set('default-tenant-network-type', 'vlan') self.assertEqual( context._get_tenant_network_types(), ['vlan', 'flat', 'local']) def test_get_tenant_network_types_unsupported_default(self): self.test_config.set('overlay-network-type', '') self.test_config.set('default-tenant-network-type', 'whizzy') with self.assertRaises(ValueError): context._get_tenant_network_types() def test_get_tenant_network_types_unconfigured_default(self): self.test_config.set('overlay-network-type', 'gre') self.test_config.set('default-tenant-network-type', 'vxlan') with self.assertRaises(ValueError): context._get_tenant_network_types() def test_get_l3ha(self): self.test_config.set('enable-l3ha', True) self.test_config.set('overlay-network-type', 'gre') self.test_config.set('neutron-plugin', 'ovs') self.test_config.set('l2-population', False) self.os_release.return_value = 'juno' self.assertEqual(context.get_l3ha(), True) def test_get_l3ha_prejuno(self): self.test_config.set('enable-l3ha', True) self.test_config.set('overlay-network-type', 'gre') self.test_config.set('neutron-plugin', 'ovs') self.test_config.set('l2-population', False) self.os_release.return_value = 'icehouse' self.assertEqual(context.get_l3ha(), False) def test_get_l3ha_l2pop(self): self.test_config.set('enable-l3ha', True) self.test_config.set('overlay-network-type', 'gre') self.test_config.set('neutron-plugin', 'ovs') self.test_config.set('l2-population', True) self.os_release.return_value = 'juno' self.assertEqual(context.get_l3ha(), False) def test_get_dvr(self): self.test_config.set('enable-dvr', True) self.test_config.set('enable-l3ha', False) self.test_config.set('overlay-network-type', 'vxlan') self.test_config.set('neutron-plugin', 'ovs') self.test_config.set('l2-population', True) self.os_release.return_value = 'juno' self.assertEqual(context.get_dvr(), True) def test_get_dvr_explicit_off(self): self.test_config.set('enable-dvr', False) self.test_config.set('enable-l3ha', False) self.test_config.set('overlay-network-type', 'vxlan') self.test_config.set('neutron-plugin', 'ovs') self.test_config.set('l2-population', True) self.os_release.return_value = 'juno' self.assertEqual(context.get_dvr(), False) def test_get_dvr_prejuno(self): self.test_config.set('enable-dvr', True) self.test_config.set('enable-l3ha', False) self.test_config.set('overlay-network-type', 'vxlan') self.test_config.set('neutron-plugin', 'ovs') self.test_config.set('l2-population', True) self.os_release.return_value = 'icehouse' self.assertEqual(context.get_dvr(), False) def test_get_dvr_gre(self): self.test_config.set('enable-dvr', True) self.test_config.set('enable-l3ha', False) self.test_config.set('overlay-network-type', 'gre') self.test_config.set('neutron-plugin', 'ovs') self.test_config.set('l2-population', True) self.os_release.return_value = 'juno' self.assertEqual(context.get_dvr(), False) def test_get_dvr_gre_kilo(self): self.test_config.set('enable-dvr', True) self.test_config.set('enable-l3ha', False) self.test_config.set('overlay-network-type', 'gre') self.test_config.set('neutron-plugin', 'ovs') self.test_config.set('l2-population', True) self.os_release.return_value = 'kilo' self.assertEqual(context.get_dvr(), True) def test_get_dvr_vxlan_kilo(self): self.test_config.set('enable-dvr', True) self.test_config.set('enable-l3ha', False) self.test_config.set('overlay-network-type', 'vxlan') self.test_config.set('neutron-plugin', 'ovs') self.test_config.set('l2-population', True) self.os_release.return_value = 'kilo' self.assertEqual(context.get_dvr(), True) def test_get_dvr_l3ha_on(self): self.test_config.set('enable-dvr', True) self.test_config.set('enable-l3ha', True) self.test_config.set('overlay-network-type', 'vxlan') self.test_config.set('neutron-plugin', 'ovs') self.test_config.set('l2-population', False) self.os_release.return_value = 'juno' self.assertEqual(context.get_dvr(), False) def test_get_dvr_l2pop(self): self.test_config.set('enable-dvr', True) self.test_config.set('enable-l3ha', False) self.test_config.set('overlay-network-type', 'vxlan') self.test_config.set('neutron-plugin', 'ovs') self.test_config.set('l2-population', False) self.os_release.return_value = 'juno' self.assertEqual(context.get_dvr(), False) def test_get_dns_domain(self): self.test_config.set('dns-domain', 'example.org.') self.test_config.set('enable-ml2-dns', True) self.os_release.return_value = 'mitaka' self.assertEqual(context.get_dns_domain(), 'example.org.') def test_get_dns_domain_bad_values(self): self.os_release.return_value = 'mitaka' self.test_config.set('enable-ml2-dns', True) bad_values = ['example@foo.org', 'exclamation!marks.notwelcom.ed', '%s.way.too.long' % ('x' * 64), '-hyphen.in.front', 'hypen-.in.back', 'no_.under_scor.es', ] for value in bad_values: self.test_config.set('dns-domain', value) self.assertRaises(ValueError, context.get_dns_domain) def test_get_ml2_mechanism_drivers(self): self.os_release.return_value = 'mitaka' self.assertEqual(context.get_ml2_mechanism_drivers(), 'openvswitch,hyperv,l2population') def test_get_ml2_mechanism_drivers_kilo(self): self.os_release.return_value = 'kilo' self.assertEqual(context.get_ml2_mechanism_drivers(), 'openvswitch,hyperv,l2population') def test_get_ml2_mechanism_drivers_liberty(self): self.os_release.return_value = 'liberty' self.assertEqual(context.get_ml2_mechanism_drivers(), 'openvswitch,l2population') def test_get_ml2_mechanism_drivers_no_l2pop(self): self.os_release.return_value = 'mitaka' self.test_config.set('l2-population', False) self.assertEqual(context.get_ml2_mechanism_drivers(), 'openvswitch,hyperv') def test_get_ml2_mechanism_drivers_sriov(self): self.os_release.return_value = 'mitaka' self.test_config.set('enable-sriov', True) self.assertEqual(context.get_ml2_mechanism_drivers(), 'openvswitch,hyperv,l2population,sriovnicswitch') def test_get_ml2_mechanism_drivers_no_l2pop_sriov(self): self.os_release.return_value = 'mitaka' self.test_config.set('enable-sriov', True) self.test_config.set('l2-population', False) self.assertEqual(context.get_ml2_mechanism_drivers(), 'openvswitch,hyperv,sriovnicswitch') def test_is_nfg_logging_enabled(self): self.os_release.return_value = 'stein' self.test_config.set('enable-firewall-group-logging', True) self.assertTrue(context.is_nfg_logging_enabled()) self.os_release.return_value = 'stein' self.test_config.set('enable-firewall-group-logging', False) self.assertFalse(context.is_nfg_logging_enabled()) self.os_release.return_value = 'queens' self.test_config.set('enable-firewall-group-logging', True) self.assertFalse(context.is_nfg_logging_enabled()) def test_is_port_forwarding_enabled(self): self.os_release.return_value = 'rocky' self.test_config.set('enable-port-forwarding', True) self.assertTrue(context.is_port_forwarding_enabled()) self.os_release.return_value = 'rocky' self.test_config.set('enable-port-forwarding', False) self.assertFalse(context.is_port_forwarding_enabled()) self.os_release.return_value = 'queens' self.test_config.set('enable-port-forwarding', True) self.assertFalse(context.is_port_forwarding_enabled()) class IdentityServiceContext(CharmTestCase): def setUp(self): super(IdentityServiceContext, self).setUp(context, TO_PATCH) self.relation_get.side_effect = self.test_relation.get self.config.side_effect = self.test_config.get self.test_config.set('region', 'region457') self.test_config.set('prefer-ipv6', False) @patch.object(charmhelpers.contrib.openstack.context, 'os_release') @patch.object(charmhelpers.contrib.openstack.context, 'format_ipv6_addr') @patch.object(charmhelpers.contrib.openstack.context, 'context_complete') @patch.object(charmhelpers.contrib.openstack.context, 'relation_get') @patch.object(charmhelpers.contrib.openstack.context, 'related_units') @patch.object(charmhelpers.contrib.openstack.context, 'relation_ids') @patch.object(charmhelpers.contrib.openstack.context, 'log') def test_ids_ctxt(self, _log, _rids, _runits, _rget, _ctxt_comp, format_ipv6_addr, _os_release): _os_release.return_value = 'rocky' _rids.return_value = 'rid1' _runits.return_value = 'runit' _ctxt_comp.return_value = True id_data = { 'service_port': 9876, 'service_host': '127.0.0.4', 'auth_host': '127.0.0.5', 'auth_port': 5432, 'service_tenant': 'ten', 'service_username': 'admin', 'service_password': 'adminpass', } _rget.return_value = id_data ids_ctxt = context.IdentityServiceContext() self.assertEqual(ids_ctxt()['region'], 'region457') @patch.object(charmhelpers.contrib.openstack.context, 'os_release') @patch.object(charmhelpers.contrib.openstack.context, 'relation_ids') @patch.object(charmhelpers.contrib.openstack.context, 'log') def test_ids_ctxt_no_rels(self, _log, _rids, _os_release): _os_release.return_value = 'rocky' _rids.return_value = [] ids_ctxt = context.IdentityServiceContext() self.assertEqual(ids_ctxt(), None) class HAProxyContextTest(CharmTestCase): def setUp(self): super(HAProxyContextTest, self).setUp(context, TO_PATCH) self.determine_api_port.return_value = 9686 self.determine_apache_port.return_value = 9686 self.api_port = 9696 def tearDown(self): super(HAProxyContextTest, self).tearDown() @patch.object(charmhelpers.contrib.openstack.context, 'mkdir') @patch.object(charmhelpers.contrib.openstack.context, 'relation_ids') @patch.object(charmhelpers.contrib.openstack.context, 'log') def test_context_No_peers(self, _log, _rids, _mkdir): _rids.return_value = [] hap_ctxt = context.HAProxyContext() with patch('builtins.__import__'): self.assertTrue('units' not in hap_ctxt()) @patch.object(charmhelpers.contrib.openstack.context, 'get_relation_ip') @patch.object(charmhelpers.contrib.openstack.context, 'mkdir') @patch.object( charmhelpers.contrib.openstack.context, 'get_netmask_for_address') @patch.object( charmhelpers.contrib.openstack.context, 'get_address_in_network') @patch.object(charmhelpers.contrib.openstack.context, 'config') @patch.object(charmhelpers.contrib.openstack.context, 'local_unit') @patch.object(charmhelpers.contrib.openstack.context, 'relation_get') @patch.object(charmhelpers.contrib.openstack.context, 'related_units') @patch.object(charmhelpers.contrib.openstack.context, 'relation_ids') @patch.object(charmhelpers.contrib.openstack.context, 'log') @patch.object(charmhelpers.contrib.openstack.context, 'kv') @patch('builtins.__import__') @patch('builtins.open') def test_context_peers(self, _open, _import, _kv, _log, _rids, _runits, _rget, _lunit, _config, _get_address_in_network, _get_netmask_for_address, _mkdir, _get_relation_ip): unit_addresses = { 'neutron-api-0': '10.10.10.10', 'neutron-api-1': '10.10.10.11', } _rids.return_value = ['rid1'] _runits.return_value = ['neutron-api/0'] _rget.return_value = unit_addresses['neutron-api-0'] _lunit.return_value = "neutron-api/1" _get_relation_ip.return_value = unit_addresses['neutron-api-1'] _config.return_value = None _get_address_in_network.return_value = None _get_netmask_for_address.return_value = '255.255.255.0' _kv().get.return_value = 'abcdefghijklmnopqrstuvwxyz123456' service_ports = {'neutron-server': [9696, 9686]} ctxt_data = { 'local_host': '127.0.0.1', 'haproxy_host': '0.0.0.0', 'local_host': '127.0.0.1', 'stat_port': '8888', 'stat_password': 'abcdefghijklmnopqrstuvwxyz123456', 'frontends': { '10.10.10.11': { 'network': '10.10.10.11/255.255.255.0', 'backends': unit_addresses, } }, 'default_backend': '10.10.10.11', 'service_ports': service_ports, 'neutron_bind_port': 9686, 'ipv6_enabled': True, } _import().api_port.return_value = 9696 hap_ctxt = context.HAProxyContext() self.maxDiff = None self.assertEqual(hap_ctxt(), ctxt_data) _open.assert_called_with('/etc/default/haproxy', 'w') class NeutronCCContextTest(CharmTestCase): def setUp(self): super(NeutronCCContextTest, self).setUp(context, TO_PATCH) self.relation_get.side_effect = self.test_relation.get self.config.side_effect = self.test_config.get self.api_port = 9696 self.determine_api_port.return_value = self.api_port self.test_config.set('neutron-plugin', 'ovs') self.test_config.set('neutron-security-groups', True) self.test_config.set('debug', True) self.test_config.set('verbose', True) self.test_config.set('neutron-external-network', 'bob') self.test_config.set('nsx-username', 'bob') self.test_config.set('nsx-password', 'hardpass') self.test_config.set('nsx-tz-uuid', 'tzuuid') self.test_config.set('nsx-l3-uuid', 'l3uuid') self.test_config.set('nsx-controllers', 'ctrl1 ctrl2') self.test_config.set('vsd-server', '192.168.2.202') self.test_config.set('vsd-auth', 'fooadmin:password') self.test_config.set('vsd-organization', 'foo') self.test_config.set('vsd-base-uri', '/nuage/api/v1_0') self.test_config.set('vsd-netpart-name', 'foo-enterprise') self.test_config.set('plumgrid-username', 'plumgrid') self.test_config.set('plumgrid-password', 'plumgrid') self.test_config.set('plumgrid-virtual-ip', '192.168.100.250') self.test_config.set('midonet-origin', 'mem-1.9') self.test_config.set('mem-username', 'yousir') self.test_config.set('mem-password', 'heslo') self.test_config.set('enable-ml2-port-security', True) self.test_config.set('dhcp-agents-per-network', 3) # Although set as True for all tests, only Ussuri templates # can apply this option. self.test_config.set('enable-igmp-snooping', True) def tearDown(self): super(NeutronCCContextTest, self).tearDown() def test_get_service_plugins(self): plugs = {"mitaka": ["A"], "queens": ["B"], "ussuri": ["C"]} p = context.NeutronCCContext().get_service_plugins('train', plugs) self.assertEquals(p, ["B"]) p = context.NeutronCCContext().get_service_plugins('ussuri', plugs) self.assertEquals(p, ["C"]) p = context.NeutronCCContext().get_service_plugins('wallaby', plugs) self.assertEquals(p, ["C"]) @patch.object(context, 'NeutronLoadBalancerContext') @patch.object(context.NeutronCCContext, 'network_manager') @patch.object(context.NeutronCCContext, 'plugin') @patch('builtins.__import__') def test_neutroncc_context_no_setting(self, _import, plugin, nm, nlb): plugin.return_value = None ctxt_data = { 'debug': True, 'enable_dvr': False, 'l3_ha': False, 'allow_automatic_dhcp_failover': True, 'allow_automatic_l3agent_failover': False, 'mechanism_drivers': 'openvswitch,l2population', 'dhcp_agents_per_network': 3, 'enable_sriov': False, 'external_network': 'bob', 'enable_igmp_snooping': True, 'neutron_bind_port': self.api_port, 'verbose': True, 'l2_population': True, 'overlay_network_type': 'gre', 'tenant_network_types': 'gre,vlan,flat,local', 'quota_floatingip': 50, 'quota_health_monitors': -1, 'quota_member': -1, 'quota_network': 10, 'quota_pool': 10, 'quota_port': 50, 'quota_router': 10, 'quota_security_group': 10, 'quota_security_group_rule': 100, 'quota_subnet': 10, 'quota_vip': 10, 'vlan_ranges': 'physnet1:1000:2000', 'vni_ranges': '1001:2000', 'extension_drivers': 'port_security', 'service_plugins': ( 'neutron.services.l3_router.l3_router_plugin.L3RouterPlugin,' 'neutron.services.firewall.fwaas_plugin.FirewallPlugin,' 'neutron.services.loadbalancer.plugin.LoadBalancerPlugin,' 'neutron.services.vpn.plugin.VPNDriverPlugin,' 'neutron.services.metering.metering_plugin.MeteringPlugin'), } napi_ctxt = context.NeutronCCContext() self.os_release.return_value = 'icehouse' with patch.object(napi_ctxt, '_ensure_packages'): self.assertEqual(ctxt_data, napi_ctxt()) @patch.object(context, 'NeutronLoadBalancerContext') @patch.object(context.NeutronCCContext, 'network_manager') @patch.object(context.NeutronCCContext, 'plugin') @patch('builtins.__import__') def test_neutroncc_context_no_setting_mitaka(self, _import, plugin, nm, nlb): plugin.return_value = None ctxt_data = { 'debug': True, 'enable_dvr': False, 'l3_ha': False, 'allow_automatic_dhcp_failover': True, 'allow_automatic_l3agent_failover': False, 'mechanism_drivers': 'openvswitch,hyperv,l2population', 'dhcp_agents_per_network': 3, 'enable_sriov': False, 'external_network': 'bob', 'global_physnet_mtu': 1500, 'enable_igmp_snooping': True, 'neutron_bind_port': self.api_port, 'verbose': True, 'l2_population': True, 'overlay_network_type': 'gre', 'path_mtu': 1500, 'tenant_network_types': 'gre,vlan,flat,local', 'quota_floatingip': 50, 'quota_health_monitors': -1, 'quota_member': -1, 'quota_network': 10, 'quota_pool': 10, 'quota_port': 50, 'quota_router': 10, 'quota_security_group': 10, 'quota_security_group_rule': 100, 'quota_subnet': 10, 'quota_vip': 10, 'vlan_ranges': 'physnet1:1000:2000', 'vni_ranges': '1001:2000', 'extension_drivers': 'port_security', 'service_plugins': 'router,firewall,lbaas,vpnaas,metering', 'network_scheduler_driver': ( 'neutron.scheduler.dhcp_agent_scheduler' '.AZAwareWeightScheduler'), 'dhcp_load_type': 'networks', 'router_scheduler_driver': ( 'neutron.scheduler.l3_agent_scheduler' '.AZLeastRoutersScheduler'), } napi_ctxt = context.NeutronCCContext() self.maxDiff = None self.os_release.return_value = 'mitaka' with patch.object(napi_ctxt, '_ensure_packages'): self.assertEqual(ctxt_data, napi_ctxt()) @patch.object(context, 'NeutronLoadBalancerContext') @patch.object(context.NeutronCCContext, 'network_manager') @patch.object(context.NeutronCCContext, 'plugin') def test_neutroncc_context_dns_setting(self, plugin, nm, nlb): plugin.return_value = None self.test_config.set('enable-ml2-dns', True) self.test_config.set('dns-domain', 'example.org.') self.os_release.return_value = 'mitaka' napi_ctxt = context.NeutronCCContext() with patch.object(napi_ctxt, '_ensure_packages'): ctxt = napi_ctxt() self.assertEqual('example.org.', ctxt['dns_domain']) self.assertEqual('port_security,dns', ctxt['extension_drivers']) self.os_release.return_value = 'queens' with patch.object(napi_ctxt, '_ensure_packages'): ctxt = napi_ctxt() self.assertEqual('example.org.', ctxt['dns_domain']) self.assertEqual('port_security,dns_domain_ports', ctxt['extension_drivers']) @patch.object(context, 'NeutronLoadBalancerContext') @patch.object(context.NeutronCCContext, 'network_manager') @patch.object(context.NeutronCCContext, 'plugin') def test_neutroncc_context_dns_no_port_security_setting(self, plugin, nm, nlb): """Verify extension drivers without port security.""" plugin.return_value = None self.test_config.set('enable-ml2-port-security', False) self.test_config.set('enable-ml2-dns', True) self.test_config.set('dns-domain', 'example.org.') self.os_release.return_value = 'mitaka' napi_ctxt = context.NeutronCCContext() with patch.object(napi_ctxt, '_ensure_packages'): ctxt = napi_ctxt() self.assertEqual('example.org.', ctxt['dns_domain']) self.assertEqual('dns', ctxt['extension_drivers']) @patch.object(context, 'NeutronLoadBalancerContext') @patch.object(context.NeutronCCContext, 'network_manager') @patch.object(context.NeutronCCContext, 'plugin') def test_neutroncc_context_dns_kilo(self, plugin, nm, nlb): """Verify dns extension and domain are not specified in kilo.""" plugin.return_value = None self.test_config.set('enable-ml2-port-security', False) self.test_config.set('enable-ml2-dns', True) self.test_config.set('dns-domain', 'example.org.') self.os_release.return_value = 'kilo' napi_ctxt = context.NeutronCCContext() with patch.object(napi_ctxt, '_ensure_packages'): ctxt = napi_ctxt() self.assertFalse('dns_domain' in ctxt) self.assertFalse('extension_drivers' in ctxt) @patch.object(context, 'NeutronLoadBalancerContext') @patch.object(context.NeutronCCContext, 'network_manager') @patch.object(context.NeutronCCContext, 'plugin') @patch('builtins.__import__') def test_neutroncc_context_vxlan(self, _import, plugin, nm, nlb): plugin.return_value = None self.test_config.set('flat-network-providers', 'physnet2 physnet3') self.test_config.set('overlay-network-type', 'vxlan') self.test_config.set('vni-ranges', '1001:2000 3001:4000') ctxt_data = { 'debug': True, 'enable_dvr': False, 'l3_ha': False, 'allow_automatic_dhcp_failover': True, 'allow_automatic_l3agent_failover': False, 'mechanism_drivers': 'openvswitch,l2population', 'dhcp_agents_per_network': 3, 'enable_sriov': False, 'external_network': 'bob', 'enable_igmp_snooping': True, 'neutron_bind_port': self.api_port, 'verbose': True, 'l2_population': True, 'overlay_network_type': 'vxlan', 'tenant_network_types': 'vxlan,vlan,flat,local', 'quota_floatingip': 50, 'quota_health_monitors': -1, 'quota_member': -1, 'quota_network': 10, 'quota_pool': 10, 'quota_port': 50, 'quota_router': 10, 'quota_security_group': 10, 'quota_security_group_rule': 100, 'quota_subnet': 10, 'quota_vip': 10, 'vlan_ranges': 'physnet1:1000:2000', 'vni_ranges': '1001:2000,3001:4000', 'network_providers': 'physnet2,physnet3', 'extension_drivers': 'port_security', 'service_plugins': ( 'neutron.services.l3_router.l3_router_plugin.L3RouterPlugin,' 'neutron.services.firewall.fwaas_plugin.FirewallPlugin,' 'neutron.services.loadbalancer.plugin.LoadBalancerPlugin,' 'neutron.services.vpn.plugin.VPNDriverPlugin,' 'neutron.services.metering.metering_plugin.MeteringPlugin'), } napi_ctxt = context.NeutronCCContext() self.os_release.return_value = 'icehouse' with patch.object(napi_ctxt, '_ensure_packages'): self.assertEqual(ctxt_data, napi_ctxt()) @patch.object(context, 'NeutronLoadBalancerContext') @patch.object(context.NeutronCCContext, 'network_manager') @patch.object(context.NeutronCCContext, 'plugin') @patch('builtins.__import__') def test_neutroncc_context_l3ha(self, _import, plugin, nm, nlb): plugin.return_value = None self.test_config.set('enable-l3ha', True) self.test_config.set('enable-qos', False) self.test_config.set('enable-vlan-trunking', False) self.test_config.set('overlay-network-type', 'gre') self.test_config.set('neutron-plugin', 'ovs') self.test_config.set('l2-population', False) self.os_release.return_value = 'juno' self.maxDiff = None ctxt_data = { 'debug': True, 'enable_dvr': False, 'l3_ha': True, 'mechanism_drivers': 'openvswitch', 'external_network': 'bob', 'enable_igmp_snooping': True, 'neutron_bind_port': self.api_port, 'verbose': True, 'l2_population': False, 'overlay_network_type': 'gre', 'tenant_network_types': 'gre,vlan,flat,local', 'max_l3_agents_per_router': 2, 'min_l3_agents_per_router': 2, 'allow_automatic_dhcp_failover': True, 'allow_automatic_l3agent_failover': False, 'dhcp_agents_per_network': 3, 'enable_sriov': False, 'quota_floatingip': 50, 'quota_health_monitors': -1, 'quota_member': -1, 'quota_network': 10, 'quota_pool': 10, 'quota_port': 50, 'quota_router': 10, 'quota_security_group': 10, 'quota_security_group_rule': 100, 'quota_subnet': 10, 'quota_vip': 10, 'vlan_ranges': 'physnet1:1000:2000', 'vni_ranges': '1001:2000', 'extension_drivers': 'port_security', 'service_plugins': ( 'neutron.services.l3_router.l3_router_plugin.L3RouterPlugin,' 'neutron.services.firewall.fwaas_plugin.FirewallPlugin,' 'neutron.services.loadbalancer.plugin.LoadBalancerPlugin,' 'neutron.services.vpn.plugin.VPNDriverPlugin,' 'neutron.services.metering.metering_plugin.MeteringPlugin'), } napi_ctxt = context.NeutronCCContext() with patch.object(napi_ctxt, '_ensure_packages'): self.assertEqual(ctxt_data, napi_ctxt()) @patch.object(context, 'NeutronLoadBalancerContext') @patch.object(context.NeutronCCContext, 'network_manager') @patch.object(context.NeutronCCContext, 'plugin') @patch('builtins.__import__') def test_neutroncc_context_no_fwaas(self, _import, plugin, nm, nlb): plugin.return_value = None self.test_config.set('enable-l3ha', True) self.test_config.set('enable-fwaas', False) self.test_config.set('enable-qos', False) self.test_config.set('enable-vlan-trunking', False) self.test_config.set('overlay-network-type', 'gre') self.test_config.set('neutron-plugin', 'ovs') self.test_config.set('l2-population', False) self.os_release.return_value = 'ussuri' self.maxDiff = None ctxt_data = { 'debug': True, 'enable_dvr': False, 'l3_ha': True, 'mechanism_drivers': 'openvswitch,hyperv', 'external_network': 'bob', 'global_physnet_mtu': 1500, 'enable_igmp_snooping': True, 'neutron_bind_port': self.api_port, 'verbose': True, 'l2_population': False, 'overlay_network_type': 'gre', 'path_mtu': 1500, 'tenant_network_types': 'gre,vlan,flat,local', 'max_l3_agents_per_router': 2, 'min_l3_agents_per_router': 2, 'network_scheduler_driver': ('neutron.scheduler.' 'dhcp_agent_scheduler.' 'AZAwareWeightScheduler'), 'allow_automatic_dhcp_failover': True, 'allow_automatic_l3agent_failover': False, 'dhcp_agents_per_network': 3, 'dhcp_load_type': 'networks', 'enable_sriov': False, 'quota_floatingip': 50, 'quota_health_monitors': -1, 'quota_member': -1, 'quota_network': 10, 'quota_pool': 10, 'quota_port': 50, 'quota_router': 10, 'quota_security_group': 10, 'quota_security_group_rule': 100, 'quota_subnet': 10, 'quota_vip': 10, 'vlan_ranges': 'physnet1:1000:2000', 'vni_ranges': '1001:2000', 'extension_drivers': 'port_security', 'router_scheduler_driver': ('neutron.scheduler.l3_agent_scheduler.' 'AZLeastRoutersScheduler'), 'service_plugins': ('router,metering,segments,' 'neutron_dynamic_routing.services.bgp.' 'bgp_plugin.BgpPlugin'), } napi_ctxt = context.NeutronCCContext() with patch.object(napi_ctxt, '_ensure_packages'): self.assertEqual(ctxt_data, napi_ctxt()) @patch.object(context.NeutronCCContext, 'network_manager') @patch.object(context.NeutronCCContext, 'plugin') @patch('builtins.__import__') def test_neutroncc_context_l3ha_l3_agents(self, _import, plugin, nm): plugin.return_value = None self.os_release.return_value = 'juno' self.test_config.set('enable-l3ha', True) self.test_config.set('l2-population', False) self.test_config.set('max-l3-agents-per-router', 2) self.test_config.set('min-l3-agents-per-router', 3) napi_ctxt = context.NeutronCCContext() with patch.object(napi_ctxt, '_ensure_packages'): self.assertRaises(ValueError, napi_ctxt) @patch.object(context, 'NeutronLoadBalancerContext') @patch.object(context.NeutronCCContext, 'network_manager') @patch.object(context.NeutronCCContext, 'plugin') @patch('builtins.__import__') def test_neutroncc_context_sriov(self, _import, plugin, nm, nlb): plugin.return_value = None self.test_config.set('enable-sriov', True) self.test_config.set('supported-pci-vendor-devs', '1111:3333 2222:4444') ctxt_data = { 'debug': True, 'enable_dvr': False, 'l3_ha': False, 'allow_automatic_dhcp_failover': True, 'allow_automatic_l3agent_failover': False, 'mechanism_drivers': 'openvswitch,hyperv,l2population' ',sriovnicswitch', 'dhcp_agents_per_network': 3, 'enable_sriov': True, 'supported_pci_vendor_devs': '1111:3333,2222:4444', 'external_network': 'bob', 'enable_igmp_snooping': True, 'neutron_bind_port': self.api_port, 'verbose': True, 'l2_population': True, 'overlay_network_type': 'gre', 'tenant_network_types': 'gre,vlan,flat,local', 'quota_floatingip': 50, 'quota_health_monitors': -1, 'quota_member': -1, 'quota_network': 10, 'quota_pool': 10, 'quota_port': 50, 'quota_router': 10, 'quota_security_group': 10, 'quota_security_group_rule': 100, 'quota_subnet': 10, 'quota_vip': 10, 'vlan_ranges': 'physnet1:1000:2000', 'vni_ranges': '1001:2000', 'extension_drivers': 'port_security', 'service_plugins': 'router,firewall,lbaas,vpnaas,metering', } napi_ctxt = context.NeutronCCContext() self.os_release.return_value = 'kilo' with patch.object(napi_ctxt, '_ensure_packages'): self.assertEqual(ctxt_data, napi_ctxt()) @patch.object(context.NeutronCCContext, 'network_manager') @patch.object(context.NeutronCCContext, 'plugin') @patch('builtins.__import__') def test_neutroncc_context_unsupported_overlay(self, _import, plugin, nm): plugin.return_value = None self.test_config.set('overlay-network-type', 'bobswitch') with self.assertRaises(Exception) as context: context.NeutronCCContext() @patch.object(context, 'NeutronLoadBalancerContext') @patch.object(context.NeutronCCContext, 'network_manager') @patch.object(context.NeutronCCContext, 'plugin') @patch('builtins.__import__') def test_neutroncc_context_api_rel(self, _import, plugin, nm, nlb): nova_url = 'http://127.0.0.10' plugin.return_value = None self.os_release.return_value = 'queens' self.related_units.return_value = ['unit1'] self.relation_ids.return_value = ['rid2'] self.test_relation.set({'nova_url': nova_url, 'restart_trigger': 'bob'}) napi_ctxt = context.NeutronCCContext() self.assertEqual(nova_url, napi_ctxt()['nova_url']) self.assertEqual('bob', napi_ctxt()['restart_trigger']) self.assertEqual(self.api_port, napi_ctxt()['neutron_bind_port']) def test_neutroncc_context_manager(self): napi_ctxt = context.NeutronCCContext() self.assertEqual(napi_ctxt.network_manager, 'neutron') self.assertEqual(napi_ctxt.plugin, 'ovs') self.assertEqual(napi_ctxt.neutron_security_groups, True) def test_neutroncc_context_manager_pkgs(self): napi_ctxt = context.NeutronCCContext() with patch.object(napi_ctxt, '_ensure_packages') as ep: napi_ctxt._ensure_packages() ep.assert_has_calls([]) @patch.object(context, 'NeutronLoadBalancerContext') @patch.object(context.NeutronCCContext, 'network_manager') @patch.object(context.NeutronCCContext, 'plugin') @patch('builtins.__import__') def test_neutroncc_context_nsx(self, _import, plugin, nm, nlb): plugin.return_value = 'nsx' self.os_release.return_value = 'havana' self.related_units.return_value = [] self.test_config.set('neutron-plugin', 'nsx') napi_ctxt = context.NeutronCCContext()() expect = { 'nsx_controllers': 'ctrl1,ctrl2', 'nsx_controllers_list': ['ctrl1', 'ctrl2'], 'nsx_l3_uuid': 'l3uuid', 'nsx_password': 'hardpass', 'nsx_tz_uuid': 'tzuuid', 'nsx_username': 'bob', } for key in expect.keys(): self.assertEqual(napi_ctxt[key], expect[key]) @patch.object(context, 'NeutronLoadBalancerContext') @patch.object(context.NeutronCCContext, 'network_manager') @patch.object(context.NeutronCCContext, 'plugin') @patch('builtins.__import__') def test_neutroncc_context_nuage(self, _import, plugin, nm, nlb): plugin.return_value = 'vsp' self.os_release.return_value = 'havana' self.related_units.return_value = ['vsdunit1'] self.relation_ids.return_value = ['vsdrid2'] self.test_config.set('neutron-plugin', 'vsp') napi_ctxt = context.NeutronCCContext()() expect = { 'vsd_server': '192.168.2.202', 'vsd_auth': 'fooadmin:password', 'vsd_organization': 'foo', 'vsd_base_uri': '/nuage/api/v1_0', 'vsd_netpart_name': 'foo-enterprise', } for key in expect.keys(): self.assertEqual(napi_ctxt[key], expect[key]) @patch.object(context, 'NeutronLoadBalancerContext') @patch.object(context.NeutronCCContext, 'network_manager') @patch.object(context.NeutronCCContext, 'plugin') @patch('builtins.__import__') def test_neutroncc_context_qos(self, _import, plugin, nm, nlb): plugin.return_value = None self.os_release.return_value = 'mitaka' self.test_config.set('enable-qos', True) self.test_config.set('enable-ml2-port-security', False) napi_ctxt = context.NeutronCCContext()() service_plugins = ('router,firewall,lbaas,vpnaas,metering,qos') expect = { 'extension_drivers': 'qos', 'service_plugins': service_plugins, } for key in expect.keys(): self.assertEqual(napi_ctxt[key], expect[key]) @patch.object(context, 'NeutronLoadBalancerContext') @patch.object(context.NeutronCCContext, 'network_manager') @patch.object(context.NeutronCCContext, 'plugin') @patch('builtins.__import__') def test_neutroncc_context_vlan_trunking(self, _import, plugin, nm, nlb): plugin.return_value = None self.os_release.return_value = 'newton' self.test_config.set('neutron-plugin', 'ovs') self.test_config.set('enable-vlan-trunking', True) napi_ctxt = context.NeutronCCContext()() expected_service_plugins = ('router,firewall,vpnaas,metering,' 'neutron_lbaas.services.loadbalancer.' 'plugin.LoadBalancerPluginv2,trunk') self.assertEqual(napi_ctxt['service_plugins'], expected_service_plugins) @patch.object(context, 'NeutronLoadBalancerContext') @patch.object(context.NeutronCCContext, 'network_manager') @patch.object(context.NeutronCCContext, 'plugin') @patch('builtins.__import__') def test_neutroncc_context_vlan_trunking_invalid_plugin(self, _import, plugin, nm, nlb): plugin.return_value = None self.os_release.return_value = 'newton' self.test_config.set('neutron-plugin', 'Calico') self.test_config.set('enable-vlan-trunking', True) napi_ctxt = context.NeutronCCContext()() expected_service_plugins = ('router,firewall,vpnaas,metering,' 'neutron_lbaas.services.loadbalancer.' 'plugin.LoadBalancerPluginv2') self.assertEqual(napi_ctxt['service_plugins'], expected_service_plugins) @patch.object(context, 'NeutronLoadBalancerContext') @patch.object(context.NeutronCCContext, 'network_manager') @patch.object(context.NeutronCCContext, 'plugin') @patch('builtins.__import__') def test_neutroncc_context_vlan_trunking_invalid_release(self, _import, plugin, nm, nlb): plugin.return_value = None self.os_release.return_value = 'mitaka' self.test_config.set('neutron-plugin', 'ovs') self.test_config.set('enable-vlan-trunking', True) napi_ctxt = context.NeutronCCContext()() expected_service_plugins = ('router,firewall,lbaas,vpnaas,metering') self.assertEqual(napi_ctxt['service_plugins'], expected_service_plugins) @patch.object(context, 'NeutronLoadBalancerContext') @patch.object(context.NeutronCCContext, 'network_manager') @patch.object(context.NeutronCCContext, 'plugin') @patch('builtins.__import__') def test_neutroncc_context_service_plugins(self, _import, plugin, nm, nlb): plugin.return_value = None self.test_config.set('enable-qos', False) self.test_config.set('enable-ml2-port-security', False) self.test_config.set('enable-vlan-trunking', False) # icehouse self.os_release.return_value = 'icehouse' service_plugins = ( 'neutron.services.l3_router.l3_router_plugin.L3RouterPlugin,' 'neutron.services.firewall.fwaas_plugin.FirewallPlugin,' 'neutron.services.loadbalancer.plugin.LoadBalancerPlugin,' 'neutron.services.vpn.plugin.VPNDriverPlugin,' 'neutron.services.metering.metering_plugin.MeteringPlugin') self.assertEqual(context.NeutronCCContext()()['service_plugins'], service_plugins) # juno self.os_release.return_value = 'juno' service_plugins = ( 'neutron.services.l3_router.l3_router_plugin.L3RouterPlugin,' 'neutron.services.firewall.fwaas_plugin.FirewallPlugin,' 'neutron.services.loadbalancer.plugin.LoadBalancerPlugin,' 'neutron.services.vpn.plugin.VPNDriverPlugin,' 'neutron.services.metering.metering_plugin.MeteringPlugin') self.assertEqual(context.NeutronCCContext()()['service_plugins'], service_plugins) # kilo self.os_release.return_value = 'kilo' service_plugins = 'router,firewall,lbaas,vpnaas,metering' self.assertEqual(context.NeutronCCContext()()['service_plugins'], service_plugins) # liberty self.os_release.return_value = 'liberty' service_plugins = 'router,firewall,lbaas,vpnaas,metering' self.assertEqual(context.NeutronCCContext()()['service_plugins'], service_plugins) # mitaka self.os_release.return_value = 'mitaka' service_plugins = 'router,firewall,lbaas,vpnaas,metering' self.assertEqual(context.NeutronCCContext()()['service_plugins'], service_plugins) # newton self.os_release.return_value = 'newton' service_plugins = ( 'router,firewall,vpnaas,metering,' 'neutron_lbaas.services.loadbalancer.plugin.LoadBalancerPluginv2') self.assertEqual(context.NeutronCCContext()()['service_plugins'], service_plugins) # ocata self.os_release.return_value = 'ocata' service_plugins = ( 'router,firewall,vpnaas,metering,' 'neutron_lbaas.services.loadbalancer.plugin.LoadBalancerPluginv2,' 'segments,' 'neutron_dynamic_routing.services.bgp.bgp_plugin.BgpPlugin' ) self.assertEqual(context.NeutronCCContext()()['service_plugins'], service_plugins) # pike self.os_release.return_value = 'pike' service_plugins = ( 'router,firewall,metering,segments,' 'neutron_lbaas.services.loadbalancer.plugin.LoadBalancerPluginv2,' 'neutron_dynamic_routing.services.bgp.bgp_plugin.BgpPlugin') self.assertEqual(context.NeutronCCContext()()['service_plugins'], service_plugins) # queens self.os_release.return_value = 'pike' service_plugins = ( 'router,firewall,metering,segments,' 'neutron_lbaas.services.loadbalancer.plugin.LoadBalancerPluginv2,' 'neutron_dynamic_routing.services.bgp.bgp_plugin.BgpPlugin') self.assertEqual(context.NeutronCCContext()()['service_plugins'], service_plugins) # rocky self.os_release.return_value = 'rocky' service_plugins = ( 'router,firewall,metering,segments,' 'neutron_dynamic_routing.services.bgp.bgp_plugin.BgpPlugin,' 'lbaasv2') ncc_context = context.NeutronCCContext()() self.assertEqual(ncc_context['service_plugins'], service_plugins) # stein self.os_release.return_value = 'stein' service_plugins = ( 'router,firewall_v2,metering,segments,' 'neutron_dynamic_routing.services.bgp.bgp_plugin.BgpPlugin,' 'lbaasv2') ncc_context = context.NeutronCCContext()() self.assertEqual(ncc_context['service_plugins'], service_plugins) self.assertTrue(ncc_context['firewall_v2']) # rocky and related to charm through neutron-load-balancer interface self.os_release.return_value = 'rocky' service_plugins = ( 'router,firewall,metering,segments,' 'neutron_dynamic_routing.services.bgp.bgp_plugin.BgpPlugin,' 'lbaasv2-proxy') lb_context = MagicMock() lb_context.return_value = {'load_balancer_name': 'octavia'} nlb.return_value = lb_context self.assertEqual(context.NeutronCCContext()()['service_plugins'], service_plugins) @patch.object(context, 'NeutronLoadBalancerContext') @patch.object(context.NeutronCCContext, 'network_manager') @patch.object(context.NeutronCCContext, 'plugin') def test_neutroncc_context_physical_network_mtus(self, plugin, nm, nlb): plugin.return_value = None self.test_config.set('physical-network-mtus', 'provider1:4000') self.os_release.return_value = 'mitaka' napi_ctxt = context.NeutronCCContext() with patch.object(napi_ctxt, '_ensure_packages'): ctxt = napi_ctxt() self.assertEqual(ctxt['physical_network_mtus'], 'provider1:4000') @patch.object(context, 'NeutronLoadBalancerContext') @patch.object(context.NeutronCCContext, 'network_manager') @patch.object(context.NeutronCCContext, 'plugin') def test_neutroncc_context_physical_network_mtus_multi(self, plugin, nm, nlb): plugin.return_value = None self.test_config.set('physical-network-mtus', 'provider1:4000 provider2:5000') self.os_release.return_value = 'mitaka' napi_ctxt = context.NeutronCCContext() with patch.object(napi_ctxt, '_ensure_packages'): ctxt = napi_ctxt() self.assertEqual(ctxt['physical_network_mtus'], 'provider1:4000,provider2:5000') class EtcdContextTest(CharmTestCase): def setUp(self): super(EtcdContextTest, self).setUp(context, TO_PATCH) self.relation_get.side_effect = self.test_relation.get self.config.side_effect = self.test_config.get self.test_config.set('neutron-plugin', 'Calico') def tearDown(self): super(EtcdContextTest, self).tearDown() def test_etcd_no_related_units(self): self.related_units.return_value = [] ctxt = context.EtcdContext()() expect = {'cluster': ''} self.assertEqual(expect, ctxt) def test_some_related_units(self): self.related_units.return_value = ['unit1'] self.relation_ids.return_value = ['rid2', 'rid3'] result = ( 'testname=http://172.18.18.18:8888,' 'testname=http://172.18.18.18:8888' ) self.test_relation.set({'cluster': result}) ctxt = context.EtcdContext()() expect = {'cluster': result} self.assertEqual(expect, ctxt) def test_early_exit(self): self.test_config.set('neutron-plugin', 'notCalico') self.related_units.return_value = ['unit1'] self.relation_ids.return_value = ['rid2', 'rid3'] self.test_relation.set({'ip': '172.18.18.18', 'port': 8888, 'name': 'testname'}) ctxt = context.EtcdContext()() expect = {'cluster': ''} self.assertEqual(expect, ctxt) class NeutronApiSDNContextTest(CharmTestCase): def setUp(self): super(NeutronApiSDNContextTest, self).setUp(context, TO_PATCH) self.relation_get.side_effect = self.test_relation.get def tearDown(self): super(NeutronApiSDNContextTest, self).tearDown() def test_init(self): napisdn_ctxt = context.NeutronApiSDNContext() self.assertEqual( napisdn_ctxt.interfaces, ['neutron-plugin-api-subordinate'] ) self.assertEqual(napisdn_ctxt.services, ['neutron-api']) self.assertEqual( napisdn_ctxt.config_file, '/etc/neutron/neutron.conf' ) @patch.object(charmhelpers.contrib.openstack.context, 'log') @patch.object(charmhelpers.contrib.openstack.context, 'relation_get') @patch.object(charmhelpers.contrib.openstack.context, 'related_units') @patch.object(charmhelpers.contrib.openstack.context, 'relation_ids') def ctxt_check(self, rel_settings, expect, _rids, _runits, _rget, _log, defaults=None, not_defaults=None): self.test_relation.set(rel_settings) _runits.return_value = ['unit1'] _rids.return_value = ['rid2'] _rget.side_effect = self.test_relation.get self.relation_ids.return_value = ['rid2'] self.related_units.return_value = ['unit1'] napisdn_instance = context.NeutronApiSDNContext() napisdn_ctxt = napisdn_instance() self.assertEqual(napisdn_ctxt, expect) defaults = defaults or [] for templ_key in defaults: self.assertTrue(napisdn_instance.is_default(templ_key)) not_defaults = not_defaults or [] for templ_key in not_defaults: self.assertFalse(napisdn_instance.is_default(templ_key)) def test_defaults(self): self.ctxt_check( {'neutron-plugin': 'ovs'}, { 'core_plugin': 'neutron.plugins.ml2.plugin.Ml2Plugin', 'neutron_plugin_config': ('/etc/neutron/plugins/ml2/' 'ml2_conf.ini'), 'neutron_plugin': 'ovs', }, defaults=['core_plugin', 'neutron_plugin_config'], ) def test_overrides(self): self.ctxt_check( { 'neutron-plugin': 'ovs', 'api-extensions-path': '/usr/local/share/neutron/extensions', 'core-plugin': 'neutron.plugins.ml2.plugin.MidoPlumODL', 'neutron-plugin-config': '/etc/neutron/plugins/fl/flump.ini', 'service-plugins': 'router,unicorn,rainbows', 'restart-trigger': 'restartnow', 'quota-driver': 'quotadriver', 'extension-drivers': 'dns,port_security', 'mechanism-drivers': 'ovn', 'tenant-network-types': 'geneve,gre,vlan,flat,local', 'neutron-security-groups': 'true', }, { 'api_extensions_path': '/usr/local/share/neutron/extensions', 'core_plugin': 'neutron.plugins.ml2.plugin.MidoPlumODL', 'neutron_plugin_config': '/etc/neutron/plugins/fl/flump.ini', 'service_plugins': 'router,unicorn,rainbows', 'restart_trigger': 'restartnow', 'quota_driver': 'quotadriver', 'neutron_plugin': 'ovs', 'extension_drivers': 'dns,port_security', 'mechanism_drivers': 'ovn', 'tenant_network_types': 'geneve,gre,vlan,flat,local', 'neutron_security_groups': 'true', }, not_defaults=[ 'api_extensions_path', 'core_plugin', 'neutron_plugin_config', 'service_plugins', 'restart_trigger', 'quota_driver', 'extension_drivers', 'mechanism_drivers', 'tenant_network_types', 'neutron_security_groups', ], ) def test_subordinateconfig(self): principle_config = { "neutron-api": { "/etc/neutron/neutron.conf": { "sections": { 'DEFAULT': [ ('neutronboost', True) ], } } } } self.ctxt_check( { 'neutron-plugin': 'ovs', 'subordinate_configuration': json.dumps(principle_config), }, { 'core_plugin': 'neutron.plugins.ml2.plugin.Ml2Plugin', 'neutron_plugin_config': ('/etc/neutron/plugins/ml2/' 'ml2_conf.ini'), 'neutron_plugin': 'ovs', 'sections': {u'DEFAULT': [[u'neutronboost', True]]}, } ) def test_empty(self): self.ctxt_check( {}, {}, ) def test_is_allowed(self): napisdn_instance = context.NeutronApiSDNContext() self.assertTrue(napisdn_instance.is_allowed('core_plugin')) self.assertFalse(napisdn_instance.is_allowed('non_existent_key')) class NeutronApiSDNConfigFileContextTest(CharmTestCase): def setUp(self): super(NeutronApiSDNConfigFileContextTest, self).setUp( context, TO_PATCH) self.relation_get.side_effect = self.test_relation.get def tearDown(self): super(NeutronApiSDNConfigFileContextTest, self).tearDown() def test_configset(self): self.test_relation.set({ 'neutron-plugin-config': '/etc/neutron/superplugin.ini' }) self.relation_ids.return_value = ['rid2'] self.related_units.return_value = ['unit1'] napisdn_ctxt = context.NeutronApiSDNConfigFileContext()() self.assertEqual(napisdn_ctxt, { 'config': '/etc/neutron/superplugin.ini' }) def test_default(self): self.relation_ids.return_value = ['rid3'] self.related_units.return_value = ['unit2'] napisdn_ctxt = context.NeutronApiSDNConfigFileContext()() self.assertEqual(napisdn_ctxt, { 'config': '/etc/neutron/plugins/ml2/ml2_conf.ini' }) def test_no_related_unites(self): self.relation_ids.return_value = ['rid4'] napisdn_ctxt = context.NeutronApiSDNConfigFileContext()() self.assertEqual(napisdn_ctxt, {}) class NeutronApiApiPasteContextTest(CharmTestCase): def setUp(self): super(NeutronApiApiPasteContextTest, self).setUp( context, TO_PATCH) self.relation_get.side_effect = self.test_relation.get def tearDown(self): super(NeutronApiApiPasteContextTest, self).tearDown() def test_default(self): middleware = [] self.test_relation.set({ 'extra_middleware': repr(middleware) }) self.relation_ids.return_value = ['rid'] self.related_units.return_value = ['testunit'] self.assertRaises(ValueError, context.NeutronApiApiPasteContext()) def test_string(self): self.test_relation.set({'extra_middleware': 'n42'}) self.relation_ids.return_value = ['rid'] self.related_units.return_value = ['testunit'] self.assertRaises(ValueError, context.NeutronApiApiPasteContext()) def test_dict(self): self.test_relation.set({'extra_middleware': {'dict_with': 'something'}}) self.relation_ids.return_value = ['rid'] self.related_units.return_value = ['testunit'] self.assertRaises(ValueError, context.NeutronApiApiPasteContext()) def test_configset(self): middleware = [{'name': 'middleware_1', 'type': 'filter', 'config': {'setting_1': 'value_1'}}, {'name': 'middleware_2', 'type': 'app', 'config': {'setting_2': 'value_2'}} ] # note repr is needed to simulate charm-helpers behavior # with regards to object serialization - the context # implementation should safely eval the string instead # of just using it self.test_relation.set({ 'extra_middleware': repr(middleware) }) self.relation_ids.return_value = ['rid2'] self.related_units.return_value = ['unit1'] napiapipaste_ctxt = context.NeutronApiApiPasteContext()() self.assertEqual(napiapipaste_ctxt, {'extra_middleware': middleware}) def __test_arg(self, key): middleware = [{'name': 'middleware_1', 'type': 'filter', 'config': {'setting_1': 'value_1'}}, {'name': 'middleware_2', 'type': 'composite', 'config': {'setting_2': 'value_2'}}] # invalidate a key middleware[0][key] = None self.test_relation.set({ 'extra_middleware': repr(middleware) }) self.relation_ids.return_value = ['rid'] self.related_units.return_value = ['testunit'] self.assertRaises(ValueError, context.NeutronApiApiPasteContext()) def test_no_type(self): self.__test_arg('type') def test_no_name(self): self.__test_arg('name') def test_no_config(self): self.__test_arg('config') class MidonetContextTest(CharmTestCase): def setUp(self): super(MidonetContextTest, self).setUp(context, TO_PATCH) self.relation_get.side_effect = self.test_relation.get self.config.side_effect = self.test_config.get self.test_config.set('neutron-plugin', 'midonet') self.test_config.set('midonet-origin', 'midonet-2015.06') def tearDown(self): super(MidonetContextTest, self).tearDown() def test_midonet_no_related_units(self): self.related_units.return_value = [] ctxt = context.MidonetContext()() expect = {} self.assertEqual(expect, ctxt) def test_some_related_units(self): self.related_units.return_value = ['unit1'] self.relation_ids.return_value = ['rid1'] self.test_relation.set({'host': '11.11.11.11', 'port': '8080'}) ctxt = context.MidonetContext()() expect = {'midonet_api_ip': '11.11.11.11', 'midonet_api_port': '8080'} self.assertEqual(expect, ctxt) class DesignateContextTest(CharmTestCase): def setUp(self): super(DesignateContextTest, self).setUp(context, TO_PATCH) self.relation_get.side_effect = self.test_relation.get def tearDown(self): super(DesignateContextTest, self).tearDown() def test_designate_no_related_units(self): self.related_units.return_value = [] ctxt = context.DesignateContext()() expect = {} self.assertEqual(expect, ctxt) def test_designate_related_units_no_reverse_dns_lookup(self): self.config.side_effect = self.test_config.get self.related_units.return_value = ['unit1'] self.relation_ids.return_value = ['rid1'] self.test_relation.set({'endpoint': 'http://1.1.1.1:9001'}) self.test_config.set('reverse-dns-lookup', False) ctxt = context.DesignateContext()() expect = {'enable_designate': True, 'designate_endpoint': 'http://1.1.1.1:9001', 'allow_reverse_dns_lookup': False} self.assertEqual(expect, ctxt) def test_designate_related_units_and_reverse_dns_lookup(self): self.config.side_effect = self.test_config.get self.related_units.return_value = ['unit1'] self.relation_ids.return_value = ['rid1'] self.test_relation.set({'endpoint': 'http://1.1.1.1:9001'}) self.test_config.set('reverse-dns-lookup', True) self.test_config.set('ipv4-ptr-zone-prefix-size', 24) self.test_config.set('ipv6-ptr-zone-prefix-size', 64) ctxt = context.DesignateContext()() expect = {'enable_designate': True, 'designate_endpoint': 'http://1.1.1.1:9001', 'allow_reverse_dns_lookup': True, 'ipv4_ptr_zone_prefix_size': 24, 'ipv6_ptr_zone_prefix_size': 64} self.assertEqual(expect, ctxt) class NeutronLoadBalancerContextTest(CharmTestCase): def setUp(self): super(NeutronLoadBalancerContextTest, self).setUp(context, TO_PATCH) self.relation_get.side_effect = self.test_relation.get def tearDown(self): super(NeutronLoadBalancerContextTest, self).tearDown() def test_neutron_load_balancer_context(self): expect = {} ctxt = context.NeutronLoadBalancerContext()() self.assertEqual(expect, ctxt) self.related_units.return_value = ['unit1'] self.relation_ids.return_value = ['rid1'] expect = {'load_balancer_name': 'FAKENAME', 'load_balancer_base_url': 'http://1.2.3.4:1234'} self.test_relation.set({'name': json.dumps('FAKENAME'), 'base_url': json.dumps('http://1.2.3.4:1234')}) ctxt = context.NeutronLoadBalancerContext()() self.assertEqual(expect, ctxt) expect = {} self.test_relation.set({'name': None, 'base_url': 'http://1.2.3.4:1234'}) ctxt = context.NeutronLoadBalancerContext()() self.assertEqual(expect, ctxt) expect = {} self.test_relation.set({'name': 'FAKENAME', 'base_url': 'http://1.2.3.4:1234'}) with self.assertRaises(ValueError): context.NeutronLoadBalancerContext()() class NeutronInfobloxContextTest(CharmTestCase): def setUp(self): super(NeutronInfobloxContextTest, self).setUp(context, TO_PATCH) self.relation_get.side_effect = self.test_relation.get self.config.side_effect = self.test_config.get def tearDown(self): super(NeutronInfobloxContextTest, self).tearDown() def test_infoblox_no_related_units(self): self.related_units.return_value = [] ctxt = context.NeutronInfobloxContext()() expect = {} self.assertEqual(expect, ctxt) def test_infoblox_related_units(self): self.related_units.return_value = ['unit1'] self.relation_ids.return_value = ['rid1'] self.test_relation.set( {'dc_id': '0', 'grid_master_host': 'foo', 'grid_master_name': 'bar', 'admin_user_name': 'faz', 'admin_password': 'baz'}) ctxt = context.NeutronInfobloxContext()() expect = {'enable_infoblox': True, 'cloud_data_center_id': '0', 'grid_master_host': 'foo', 'grid_master_name': 'bar', 'infoblox_admin_user_name': 'faz', 'infoblox_admin_password': 'baz', 'wapi_version': '2.3', 'wapi_max_results': '-50000', 'wapi_paging': True} self.assertEqual(expect, ctxt) def test_infoblox_related_units_missing_data(self): self.related_units.return_value = ['unit1'] self.relation_ids.return_value = ['rid1'] self.test_relation.set( {'dc_id': '0', 'grid_master_host': 'foo'}) ctxt = context.NeutronInfobloxContext()() expect = {} self.assertEqual(expect, ctxt)