Heat templates for deploying OpenStack
You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
 
 
 
 

303 lines
13 KiB

#!/usr/bin/env python
# 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 os
import tempfile
import yaml
from oslotest import base
from tools import convert_heat_nic_config_to_ansible_j2 as converter
FAKE_PARAM_TO_VAR_MAP = {
'ControlPlaneDefaultRoute': '{{ ctlplane_gateway_ip }}',
'ControlPlaneIp': '{{ ctlplane_ip }}',
'ControlPlaneMtu': '{{ ctlplane_mtu }}',
'ControlPlaneStaticRoutes': '{{ ctlplane_host_routes }}',
'ControlPlaneSubnetCidr': '{{ ctlplane_subnet_cidr }}',
'DnsSearchDomains': '{{ dns_search_domains }}',
'DnsServers': '{{ ctlplane_dns_nameservers }}',
'InternalApiInterfaceDefaultRoute': '{{ internal_api_gateway_ip }}',
'InternalApiInterfaceRoutes': '{{ internal_api_host_routes }}',
'InternalApiIpSubnet': '{{ internal_api_ip }}/{{ internal_api_cidr }}',
'InternalApiMtu': '{{ internal_api_mtu }}',
'InternalApiNetworkVlanID': '{{ internal_api_vlan_id }}',
'NumDpdkInterfaceRxQueues': '{{ num_dpdk_interface_rx_queues }}',
'BondInterfaceOvsOptions': '{{ bond_interface_ovs_options }}',
'TenantInterfaceDefaultRoute': '{{ tenant_gateway_ip }}',
'TenantInterfaceRoutes': '{{ tenant_host_routes }}',
'TenantIpSubnet': '{{ tenant_ip }}/{{ tenant_cidr }}',
'TenantMtu': '{{ tenant_mtu }}',
'TenantNetworkVlanID': '{{ tenant_vlan_id }}',
}
class ConvertToAnsibleJ2TestCase(base.BaseTestCase):
def setUp(self):
super(ConvertToAnsibleJ2TestCase, self).setUp()
self.fake_param_to_var_map = FAKE_PARAM_TO_VAR_MAP
stack_env_file = os.path.join(
os.path.dirname(__file__),
'nic_config_convert_samples/stack_env_simple.yaml')
networks_file = os.path.join(
os.path.dirname(__file__),
'nic_config_convert_samples/networks_file_simple.yaml')
with open(stack_env_file, 'r') as f:
self.fake_stack_env = yaml.safe_load(f.read())
self.convert = converter.ConvertToAnsibleJ2(self.fake_stack_env,
networks_file)
def test_to_j2_var(self):
self.assertEqual('{{ some_var }}',
converter.ConvertToAnsibleJ2.to_j2_var('some_var'))
def test_to_j2_var_raises_on_unsupported_type(self):
for _type in [list(), dict(), int(), bool()]:
self.assertRaises(RuntimeError,
converter.ConvertToAnsibleJ2.to_j2_var,
_type)
def test_create_param_to_var_map(self):
networks_file = os.path.join(
os.path.dirname(__file__),
'nic_config_convert_samples/networks_file_simple.yaml')
param_to_var_map = self.convert.create_param_to_var_map(networks_file)
self.assertEqual(self.fake_param_to_var_map.keys(),
param_to_var_map.keys())
for key in param_to_var_map:
self.assertEqual(self.fake_param_to_var_map[key],
param_to_var_map[key])
def test_convert_get_param(self):
for k, v in self.fake_param_to_var_map.items():
old = {'get_param': k}
new = self.convert.convert_get_param(old)
self.assertEqual(v, new)
for k, v in self.fake_stack_env['parameter_defaults'].items():
old = {'get_param': k}
new = self.convert.convert_get_param(old)
if isinstance(v, str):
self.assertTrue(v in new)
self.assertTrue(new.startswith(converter.QUOTE_FIX + v))
elif isinstance(v, (bool, int, list, dict)):
self.assertEqual(v, new)
def test_convert_get_param_unable_param_not_defined(self):
old = {'get_param': 'UNKNOWN_PARAM'}
new = self.convert.convert_get_param(old)
self.assertEqual(converter.QUOTE_FIX
+ ('NEED MANUAL CONVERSION: {\'get_param\': '
'\'UNKNOWN_PARAM\'}') + converter.QUOTE_FIX, new)
def test_convert_get_param_comlex_data_struct_not_supported(self):
old = {'get_param': ['COMPLEX', 'DATA']}
new = self.convert.convert_get_param(old)
self.assertEqual(converter.QUOTE_FIX
+ ('NEED MANUAL CONVERSION: {\'get_param\': '
'[\'COMPLEX\', \'DATA\']}') + converter.QUOTE_FIX,
new)
def test_convert_get_attr(self):
old = {'get_attr': ['MinViableMtu', 'value']}
new = self.convert.convert_get_attr(old)
self.assertEqual('{{ min_viable_mtu }}', new)
def test_convert_get_attr_not_suported(self):
old = {'get_attr': ['UNSUPPORTED', 'value']}
new = self.convert.convert_get_attr(old)
self.assertEqual(('NEED MANUAL CONVERSION: '
'{\'get_attr\': [\'UNSUPPORTED\', \'value\']}'), new)
def test_convert_list_join(self):
list_join_attrs = ['/', ['foo', 'bar']]
new = self.convert.convert_list_join(list_join_attrs)
self.assertEqual('foo/bar', new)
def test_convert_list_concat(self):
list_concat_attrs = [['a_list', 'list_a'], ['list_b', 'b_list']]
new = self.convert.convert_list_concat(list_concat_attrs)
self.assertEqual(
("{{ [['%_fix_quote_%a_list%_fix_quote_%', "
"'%_fix_quote_%list_a%_fix_quote_%'], "
"['%_fix_quote_%list_b%_fix_quote_%', "
"'%_fix_quote_%b_list%_fix_quote_%']] "
"| flatten }}"), new)
def test_convert_list_concat_raise_if_not_list(self):
list_concat_attrs = 'NOT_A_LIST'
self.assertRaises(RuntimeError,
self.convert.convert_list_concat,
list_concat_attrs)
def test_convert_list_concat_unique(self):
convert_list_concat_unique = [['a_list', 'b_list'],
['list_b', 'b_list']]
new = self.convert.convert_list_concat_unique(
convert_list_concat_unique)
self.assertEqual(
("{{ [['%_fix_quote_%a_list%_fix_quote_%', "
"'%_fix_quote_%b_list%_fix_quote_%'], "
"['%_fix_quote_%list_b%_fix_quote_%', "
"'%_fix_quote_%b_list%_fix_quote_%']] "
"| flatten | unique }}"), new)
def test_convert_list_concat_unique_raise_if_not_list(self):
convert_list_concat_unique = 'NOT_A_LIST'
self.assertRaises(RuntimeError,
self.convert.convert_list_concat_unique,
convert_list_concat_unique)
def test_recursive_convert_str_bool_int_not_converted(self):
# old is a string
self.assertEqual('string', self.convert.recursive_convert('string'))
# old is a boolean
self.assertEqual(True, self.convert.recursive_convert(True))
# old is a number
self.assertEqual(1, self.convert.recursive_convert(1))
def test_recursive_convert_nothing_to_convert(self):
old = {'foo': 'bar', 'baz': ['a', 'b', {'c': 'd'}]}
self.assertEqual(old, self.convert.recursive_convert(old))
def test_recursive_convert_complex(self):
addresses = [
{'ip_netmask': {
'list_join': ['/',
[{'get_param': 'ControlPlaneIp'},
{'get_param': 'ControlPlaneSubnetCidr'}]
]}}]
old = {'type': 'interface', 'name': 'nic1', 'use_dhcp': 'false',
'addresses': addresses}
expected = {'name': 'nic1', 'type': 'interface', 'use_dhcp': 'false',
'addresses': [
{'ip_netmask':
'{{ ctlplane_ip }}/{{ ctlplane_subnet_cidr }}'}]}
self.assertEqual(expected, self.convert.recursive_convert(old))
def test_recursive_convert_unsupported_intrinsic_fn(self):
for fn in converter.UNSUPPORTED_HEAT_INTRINSIC_FUNCTIONS:
old = {'foo': 'bar', 'baz': ['a', 'b', {fn: 'd'}]}
unsupported_str = (
"UNSUPPORTED HEAT INTRINSIC FUNCTION {x} REQUIRES MANUAL "
"CONVERSION {{'{x}': 'd'}}".format(x=fn))
new = {'foo': 'bar', 'baz': ['a', 'b', unsupported_str]}
self.assertEqual(new, self.convert.recursive_convert(old))
def test_convert_template(self):
template_file = os.path.join(
os.path.dirname(__file__),
'nic_config_convert_samples/heat_templates/simple.yaml')
j2_config, mtu_header, j2_header = self.convert.convert_template(
template_file)
self.assertIsNone(mtu_header)
self.assertIsNone(j2_header)
expected = [
{'name': 'nic1',
'type': 'interface',
'use_dhcp': False,
'addresses': [{
'ip_netmask': '{{ ctlplane_ip }}/{{ ctlplane_subnet_cidr }}'}
]},
{'name': 'nic2',
'type': 'interface',
'use_dhcp': False},
{'type': 'vlan',
'device': 'nic2',
'vlan_id': '{{ internal_api_vlan_id }}',
'addresses': [{
'ip_netmask': '{{ internal_api_ip }}/{{ internal_api_cidr }}'}
]},
{'type': 'ovs_bridge',
'name': 'bridge_name',
'dns_servers': '{{ ctlplane_dns_nameservers }}',
'members': [
{'type': 'interface',
'name': 'nic3',
'primary': True},
{'type': 'vlan',
'vlan_id': '{{ tenant_vlan_id }}',
'addresses': [{
'ip_netmask': '{{ tenant_ip }}/{{ tenant_cidr }}'}
]},
]},
]
self.assertEqual(expected, j2_config['network_config'])
def convert_heat_to_ansible_j2(self, heat_template, j2_reference,
networks_file='network_file_complex.yaml',
stack_env='stack_env_simple.yaml'):
networks_file = os.path.join(
os.path.dirname(__file__),
'nic_config_convert_samples/' + networks_file)
template_file = os.path.join(
os.path.dirname(__file__),
'nic_config_convert_samples/heat_templates/' + heat_template)
reference_file = os.path.join(
os.path.dirname(__file__),
'nic_config_convert_samples/j2_references/' + j2_reference)
stack_env_file = os.path.join(
os.path.dirname(__file__),
'nic_config_convert_samples/' + stack_env)
with open(stack_env_file, 'r') as f:
stack_env = yaml.safe_load(f.read())
convert = converter.ConvertToAnsibleJ2(stack_env, networks_file)
j2_config, mtu_header, j2_header = convert.convert_template(
template_file)
with tempfile.TemporaryDirectory() as temp_dir:
j2_template = os.path.abspath(temp_dir) + '/j2_template.j2'
converter.write_j2_template(j2_template, j2_config, mtu_header,
j2_header)
with open(reference_file, 'r') as a:
with open(j2_template, 'r') as b:
reference = a.read()
result = b.read()
self.assertEqual(reference, result)
def test_convert_and_write_file_simple(self):
self.convert_heat_to_ansible_j2(
'simple.yaml', 'simple.j2',
networks_file='networks_file_simple.yaml')
def test_convert_and_write_file_complex01_incomplete(self):
self.convert_heat_to_ansible_j2('complex.yaml',
'complex_incomplete.j2')
def test_convert_and_write_file_complex01_complete(self):
self.convert_heat_to_ansible_j2('complex.yaml', 'complex_complete.j2',
stack_env='stack_env_complex.yaml')
def test_convert_2_linux_bonds_vlan_controller(self):
self.convert_heat_to_ansible_j2('2-linux-bonds-vlans-controller.yaml',
'2-linux-bonds-vlans-controller.j2')
def test_convert_bond_vlans_controller(self):
self.convert_heat_to_ansible_j2('bond-vlans-controller.yaml',
'bond-vlans-controller.j2')
def test_convert_multiple_nics_vlans_controller(self):
self.convert_heat_to_ansible_j2('multiple-nics-vlans-controller.yaml',
'multiple-nics-vlans-controller.j2')
def test_convert_single_nic_linux_bridge_vlans_controller(self):
self.convert_heat_to_ansible_j2(
'single-nic-linux-bridge-vlans-controller.yaml',
'single-nic-linux-bridge-vlans-controller.j2')
def test_convert_single_nic_vlans_controller(self):
self.convert_heat_to_ansible_j2('single-nic-vlans-controller.yaml',
'single-nic-vlans-controller.j2')