Browse Source

Move get_ctlplane_attrs to utils

Change-Id: Icf1189d78ccccac303d42ae9b35a532a7ecf226f
changes/04/794804/5
Harald Jensås 2 months ago
parent
commit
78190843e3
  1. 81
      tripleoclient/tests/fakes.py
  2. 60
      tripleoclient/tests/test_utils.py
  3. 81
      tripleoclient/tests/v1/overcloud_deploy/fakes.py
  4. 78
      tripleoclient/tests/v1/overcloud_deploy/test_overcloud_deploy.py
  5. 56
      tripleoclient/tests/v1/overcloud_update/test_overcloud_update.py
  6. 39
      tripleoclient/utils.py
  7. 40
      tripleoclient/v1/overcloud_deploy.py

81
tripleoclient/tests/fakes.py

@ -330,3 +330,84 @@ class FakePlaybookExecution(utils.TestCommand):
def fake_ansible_runner_run_return(rc=0):
return 'Test Status', rc
class FakeNeutronNetwork(dict):
def __init__(self, **attrs):
NETWORK_ATTRS = ['id',
'name',
'status',
'tenant_id',
'is_admin_state_up',
'mtu',
'segments',
'is_shared',
'subnet_ids',
'provider:network_type',
'provider:physical_network',
'provider:segmentation_id',
'router:external',
'availability_zones',
'availability_zone_hints',
'is_default',
'tags']
raw = dict.fromkeys(NETWORK_ATTRS)
raw.update(attrs)
raw.update({
'provider_physical_network': attrs.get(
'provider:physical_network', None),
'provider_network_type': attrs.get(
'provider:network_type', None),
'provider_segmentation_id': attrs.get(
'provider:segmentation_id', None)
})
super(FakeNeutronNetwork, self).__init__(raw)
def __getattr__(self, key):
try:
return self[key]
except KeyError:
raise AttributeError(key)
def __setattr__(self, key, value):
if key in self:
self[key] = value
else:
raise AttributeError(key)
class FakeNeutronSubnet(dict):
def __init__(self, **attrs):
SUBNET_ATTRS = ['id',
'name',
'network_id',
'cidr',
'tenant_id',
'is_dhcp_enabled',
'dns_nameservers',
'allocation_pools',
'host_routes',
'ip_version',
'gateway_ip',
'ipv6_address_mode',
'ipv6_ra_mode',
'subnetpool_id',
'segment_id',
'tags']
raw = dict.fromkeys(SUBNET_ATTRS)
raw.update(attrs)
super(FakeNeutronSubnet, self).__init__(raw)
def __getattr__(self, key):
try:
return self[key]
except KeyError:
raise AttributeError(key)
def __setattr__(self, key, value):
if key in self:
self[key] = value
else:
raise AttributeError(key)

60
tripleoclient/tests/test_utils.py

@ -19,6 +19,7 @@ import argparse
import datetime
import logging
import mock
import openstack
import os
import os.path
import shutil
@ -1911,3 +1912,62 @@ class TestTempDirs(base.TestCase):
mock_log.assert_called_once_with(
"Not cleaning temporary directory [ foo ]")
class TestGetCtlplaneAttrs(base.TestCase):
@mock.patch('openstack.connect', autospec=True)
@mock.patch.object(openstack.connection, 'Connection', autospec=True)
def test_get_ctlplane_attrs_no_network(self, mock_conn, mock_connect):
mock_connect.return_value = mock_conn
mock_conn.network.find_network.return_value = None
expected = dict()
self.assertEqual(expected, utils.get_ctlplane_attrs())
@mock.patch('openstack.connect', autospec=True)
def test_get_ctlplane_attrs_no_config(self, mock_connect):
mock_connect.side_effect = openstack.exceptions.ConfigException
expected = dict()
self.assertEqual(expected, utils.get_ctlplane_attrs())
@mock.patch('openstack.connect', autospec=True)
@mock.patch.object(openstack.connection, 'Connection', autospec=True)
def test_get_ctlplane_attrs(self, mock_conn, mock_connect):
mock_connect.return_value = mock_conn
fake_network = fakes.FakeNeutronNetwork(
name='net_name',
mtu=1440,
dns_domain='ctlplane.localdomain.',
tags=[],
subnet_ids=['subnet_id'])
fake_subnet = fakes.FakeNeutronSubnet(
id='subnet_id',
name='subnet_name',
cidr='192.168.24.0/24',
gateway_ip='192.168.24.1',
host_routes=[
{'destination': '192.168.25.0/24', 'nexthop': '192.168.24.1'}],
dns_nameservers=['192.168.24.254'],
ip_version=4
)
mock_conn.network.find_network.return_value = fake_network
mock_conn.network.get_subnet.return_value = fake_subnet
expected = {
'network': {
'dns_domain': 'ctlplane.localdomain.',
'mtu': 1440,
'name': 'net_name',
'tags': []},
'subnets': {
'subnet_name': {
'cidr': '192.168.24.0/24',
'dns_nameservers': ['192.168.24.254'],
'gateway_ip': '192.168.24.1',
'host_routes': [{'destination': '192.168.25.0/24',
'nexthop': '192.168.24.1'}],
'ip_version': 4,
'name': 'subnet_name'}
}
}
self.assertEqual(expected, utils.get_ctlplane_attrs())

81
tripleoclient/tests/v1/overcloud_deploy/fakes.py

@ -108,84 +108,3 @@ class TestDeployOvercloud(fakes.FakePlaybookExecution):
def setUp(self):
super(TestDeployOvercloud, self).setUp(ansible_mock=False)
class FakeNeutronNetwork(dict):
def __init__(self, **attrs):
NETWORK_ATTRS = ['id',
'name',
'status',
'tenant_id',
'is_admin_state_up',
'mtu',
'segments',
'is_shared',
'subnet_ids',
'provider:network_type',
'provider:physical_network',
'provider:segmentation_id',
'router:external',
'availability_zones',
'availability_zone_hints',
'is_default',
'tags']
raw = dict.fromkeys(NETWORK_ATTRS)
raw.update(attrs)
raw.update({
'provider_physical_network': attrs.get(
'provider:physical_network', None),
'provider_network_type': attrs.get(
'provider:network_type', None),
'provider_segmentation_id': attrs.get(
'provider:segmentation_id', None)
})
super(FakeNeutronNetwork, self).__init__(raw)
def __getattr__(self, key):
try:
return self[key]
except KeyError:
raise AttributeError(key)
def __setattr__(self, key, value):
if key in self:
self[key] = value
else:
raise AttributeError(key)
class FakeNeutronSubnet(dict):
def __init__(self, **attrs):
SUBNET_ATTRS = ['id',
'name',
'network_id',
'cidr',
'tenant_id',
'is_dhcp_enabled',
'dns_nameservers',
'allocation_pools',
'host_routes',
'ip_version',
'gateway_ip',
'ipv6_address_mode',
'ipv6_ra_mode',
'subnetpool_id',
'segment_id',
'tags']
raw = dict.fromkeys(SUBNET_ATTRS)
raw.update(attrs)
super(FakeNeutronSubnet, self).__init__(raw)
def __getattr__(self, key):
try:
return self[key]
except KeyError:
raise AttributeError(key)
def __setattr__(self, key, value):
if key in self:
self[key] = value
else:
raise AttributeError(key)

78
tripleoclient/tests/v1/overcloud_deploy/test_overcloud_deploy.py

@ -21,7 +21,6 @@ import tempfile
import yaml
import mock
import openstack
from osc_lib import exceptions as oscexc
from osc_lib.tests import utils
@ -129,8 +128,8 @@ class TestDeployOvercloud(fakes.TestDeployOvercloud):
@mock.patch('heatclient.common.template_utils.'
'process_environment_and_files', autospec=True)
@mock.patch('tripleoclient.utils.check_nic_config_with_ansible')
@mock.patch('tripleoclient.v1.overcloud_deploy.DeployOvercloud.'
'_get_ctlplane_attrs', autospec=True, return_value={})
@mock.patch('tripleoclient.utils.get_ctlplane_attrs', autospec=True,
return_value={})
@mock.patch('tripleoclient.utils.copy_clouds_yaml')
@mock.patch('tripleoclient.v1.overcloud_deploy.DeployOvercloud.'
'_get_undercloud_host_entry', autospec=True,
@ -247,8 +246,8 @@ class TestDeployOvercloud(fakes.TestDeployOvercloud):
autospec=True, return_value={})
@mock.patch('heatclient.common.template_utils.'
'process_environment_and_files', autospec=True)
@mock.patch('tripleoclient.v1.overcloud_deploy.DeployOvercloud.'
'_get_ctlplane_attrs', autospec=True, return_value={})
@mock.patch('tripleoclient.utils.get_ctlplane_attrs', autospec=True,
return_value={})
@mock.patch('tripleoclient.workflows.deployment.create_overcloudrc',
autospec=True)
@mock.patch('tripleoclient.utils.copy_clouds_yaml')
@ -864,8 +863,8 @@ class TestDeployOvercloud(fakes.TestDeployOvercloud):
return_value={})
@mock.patch('tripleoclient.workflows.roles.get_roles_data',
autospec=True, return_value={})
@mock.patch('tripleoclient.v1.overcloud_deploy.DeployOvercloud.'
'_get_ctlplane_attrs', autospec=True, return_value={})
@mock.patch('tripleoclient.utils.get_ctlplane_attrs', autospec=True,
return_value={})
@mock.patch('tripleoclient.utils.copy_clouds_yaml')
@mock.patch('tripleoclient.v1.overcloud_deploy.DeployOvercloud.'
'_get_undercloud_host_entry', autospec=True,
@ -1246,8 +1245,8 @@ class TestDeployOvercloud(fakes.TestDeployOvercloud):
@mock.patch('tripleoclient.utils.process_multiple_environments',
autospec=True)
@mock.patch('tripleoclient.utils.check_nic_config_with_ansible')
@mock.patch('tripleoclient.v1.overcloud_deploy.DeployOvercloud.'
'_get_ctlplane_attrs', autospec=True, return_value={})
@mock.patch('tripleoclient.utils.get_ctlplane_attrs', autospec=True,
return_value={})
@mock.patch('tripleoclient.utils.copy_clouds_yaml')
@mock.patch('tripleoclient.v1.overcloud_deploy.DeployOvercloud.'
'_get_undercloud_host_entry', autospec=True,
@ -1543,67 +1542,6 @@ class TestDeployOvercloud(fakes.TestDeployOvercloud):
'specified.')
mock_warning.assert_called_once_with(expected_message)
@mock.patch('openstack.connect', autospec=True)
def test__get_ctlplane_attrs_no_config(self, mock_connect):
mock_connect.side_effect = openstack.exceptions.ConfigException
function = overcloud_deploy.DeployOvercloud._get_ctlplane_attrs
expected = dict()
self.assertEqual(expected, function(mock.ANY))
@mock.patch('openstack.connect', autospec=True)
@mock.patch.object(openstack.connection, 'Connection', autospec=True)
def test__get_ctlplane_attrs_no_network(self, mock_conn, mock_connect):
mock_connect.return_value = mock_conn
function = overcloud_deploy.DeployOvercloud._get_ctlplane_attrs
mock_conn.network.find_network.return_value = None
expected = dict()
self.assertEqual(expected, function(mock.ANY))
@mock.patch('openstack.connect', autospec=True)
@mock.patch.object(openstack.connection, 'Connection', autospec=True)
def test__get_ctlplane_attrs(self, mock_conn, mock_connect):
mock_connect.return_value = mock_conn
function = overcloud_deploy.DeployOvercloud._get_ctlplane_attrs
fake_network = fakes.FakeNeutronNetwork(
name='net_name',
mtu=1440,
dns_domain='ctlplane.localdomain.',
tags=[],
subnet_ids=['subnet_id'])
fake_subnet = fakes.FakeNeutronSubnet(
id='subnet_id',
name='subnet_name',
cidr='192.168.24.0/24',
gateway_ip='192.168.24.1',
host_routes=[
{'destination': '192.168.25.0/24', 'nexthop': '192.168.24.1'}],
dns_nameservers=['192.168.24.254'],
ip_version=4
)
mock_conn.network.find_network.return_value = fake_network
mock_conn.network.get_subnet.return_value = fake_subnet
expected = {
'network': {
'dns_domain': 'ctlplane.localdomain.',
'mtu': 1440,
'name': 'net_name',
'tags': []},
'subnets': {
'subnet_name': {
'cidr': '192.168.24.0/24',
'dns_nameservers': ['192.168.24.254'],
'gateway_ip': '192.168.24.1',
'host_routes': [{'destination': '192.168.25.0/24',
'nexthop': '192.168.24.1'}],
'ip_version': 4,
'name': 'subnet_name'}
}
}
self.assertEqual(expected, function(mock.ANY))
class TestArgumentValidation(fakes.TestDeployOvercloud):

56
tripleoclient/tests/v1/overcloud_update/test_overcloud_update.py

@ -36,73 +36,73 @@ class TestOvercloudUpdatePrepare(fakes.TestOvercloudUpdatePrepare):
self.mock_uuid4 = uuid4_patcher.start()
self.addCleanup(self.mock_uuid4.stop)
@mock.patch('tripleoclient.v1.overcloud_deploy.DeployOvercloud.'
'_get_ctlplane_attrs', autospec=True, return_value={})
@mock.patch('tripleoclient.utils.ensure_run_as_normal_user')
@mock.patch('tripleoclient.utils.prompt_user_for_confirmation',
return_value=True)
@mock.patch('tripleoclient.v1.overcloud_deploy.DeployOvercloud.'
'_get_undercloud_host_entry', autospec=True,
return_value='192.168.0.1 uc.ctlplane.localhost uc.ctlplane')
@mock.patch('tripleoclient.v1.overcloud_update.UpdatePrepare.log',
autospec=True)
@mock.patch('six.moves.builtins.open')
@mock.patch('os.path.abspath')
@mock.patch('yaml.safe_load')
@mock.patch('shutil.copytree', autospec=True)
@mock.patch('six.moves.builtins.open')
@mock.patch('tripleoclient.v1.overcloud_deploy.DeployOvercloud.'
'take_action', autospec=True)
def test_update_out(self, mock_deploy, mock_open, mock_copy, mock_yaml,
mock_abspath, mock_logger,
mock_get_undercloud_host_entry,
mock_confirm, mock_usercheck,
mock_get_ctlplane_attrs):
def test_update_failed(self, mock_deploy, mock_copy, mock_yaml,
mock_abspath, mock_open,
mock_confirm, mock_usercheck):
mock_deploy.side_effect = exceptions.DeploymentError()
mock_yaml.return_value = {'fake_container': 'fake_value'}
argslist = ['--stack', 'overcloud', '--templates']
argslist = ['--stack', 'overcloud', '--templates', ]
verifylist = [
('stack', 'overcloud'),
('templates', constants.TRIPLEO_HEAT_TEMPLATES),
]
parsed_args = self.check_parser(self.cmd, argslist, verifylist)
with mock.patch('os.path.exists') as mock_exists, \
mock.patch('os.path.isfile') as mock_isfile:
mock_exists.return_value = True
mock_isfile.return_value = True
self.cmd.take_action(parsed_args)
self.assertRaises(exceptions.DeploymentError,
self.cmd.take_action, parsed_args)
mock_usercheck.assert_called_once()
mock_deploy.assert_called_once()
@mock.patch('tripleoclient.utils.get_ctlplane_attrs', autospec=True,
return_value={})
@mock.patch('tripleoclient.utils.ensure_run_as_normal_user')
@mock.patch('tripleoclient.utils.prompt_user_for_confirmation',
return_value=True)
@mock.patch('six.moves.builtins.open')
@mock.patch('tripleoclient.v1.overcloud_deploy.DeployOvercloud.'
'_get_undercloud_host_entry', autospec=True,
return_value='192.168.0.1 uc.ctlplane.localhost uc.ctlplane')
@mock.patch('tripleoclient.v1.overcloud_update.UpdatePrepare.log',
autospec=True)
@mock.patch('os.path.abspath')
@mock.patch('yaml.safe_load')
@mock.patch('shutil.copytree', autospec=True)
@mock.patch('six.moves.builtins.open')
@mock.patch('tripleoclient.v1.overcloud_deploy.DeployOvercloud.'
'take_action', autospec=True)
def test_update_failed(self, mock_deploy, mock_copy, mock_yaml,
mock_abspath, mock_open,
mock_confirm, mock_usercheck):
mock_deploy.side_effect = exceptions.DeploymentError()
def test_update_out(self, mock_deploy, mock_open, mock_copy, mock_yaml,
mock_abspath, mock_logger,
mock_get_undercloud_host_entry,
mock_confirm, mock_usercheck,
mock_get_ctlplane_attrs):
mock_yaml.return_value = {'fake_container': 'fake_value'}
argslist = ['--stack', 'overcloud', '--templates', ]
argslist = ['--stack', 'overcloud', '--templates']
verifylist = [
('stack', 'overcloud'),
('templates', constants.TRIPLEO_HEAT_TEMPLATES),
]
parsed_args = self.check_parser(self.cmd, argslist, verifylist)
parsed_args = self.check_parser(self.cmd, argslist, verifylist)
with mock.patch('os.path.exists') as mock_exists, \
mock.patch('os.path.isfile') as mock_isfile:
mock_exists.return_value = True
mock_isfile.return_value = True
self.assertRaises(exceptions.DeploymentError,
self.cmd.take_action, parsed_args)
self.cmd.take_action(parsed_args)
mock_usercheck.assert_called_once()
mock_deploy.assert_called_once()
class TestOvercloudUpdateRun(fakes.TestOvercloudUpdateRun):

39
tripleoclient/utils.py

@ -30,6 +30,7 @@ import logging
import multiprocessing
import netaddr
import openstack
import os
import os.path
import pwd
@ -2675,3 +2676,41 @@ def export_overcloud(heat, stack, excludes, should_filter,
data.update({'AddVipsToEtcHosts': False})
data = dict(parameter_defaults=data)
return data
def get_ctlplane_attrs():
try:
conn = openstack.connect('undercloud')
except openstack.exceptions.ConfigException:
return dict()
if not conn.endpoint_for('network'):
return dict()
network = conn.network.find_network('ctlplane')
if network is None:
return dict()
net_attributes_map = {'network': dict(), 'subnets': dict()}
net_attributes_map['network'].update({
'name': network.name,
'mtu': network.mtu,
'dns_domain': network.dns_domain,
'tags': network.tags,
})
for subnet_id in network.subnet_ids:
subnet = conn.network.get_subnet(subnet_id)
net_attributes_map['subnets'].update({
subnet.name: {
'name': subnet.name,
'cidr': subnet.cidr,
'gateway_ip': subnet.gateway_ip,
'host_routes': subnet.host_routes,
'dns_nameservers': subnet.dns_nameservers,
'ip_version': subnet.ip_version,
}
})
return net_attributes_map

40
tripleoclient/v1/overcloud_deploy.py

@ -31,7 +31,6 @@ import yaml
from heatclient.common import template_utils
from keystoneauth1.exceptions.catalog import EndpointNotFound
import openstack
from osc_lib import exceptions as oscexc
from osc_lib.i18n import _
from tripleo_common.image import kolla_builder
@ -124,47 +123,10 @@ class DeployOvercloud(command.Command):
parameters[
'UndercloudHostsEntries'] = [self._get_undercloud_host_entry()]
parameters['CtlplaneNetworkAttributes'] = self._get_ctlplane_attrs()
parameters['CtlplaneNetworkAttributes'] = utils.get_ctlplane_attrs()
return parameters
def _get_ctlplane_attrs(self):
try:
conn = openstack.connect('undercloud')
except openstack.exceptions.ConfigException:
return dict()
if not conn.endpoint_for('network'):
return dict()
network = conn.network.find_network('ctlplane')
if network is None:
return dict()
net_attributes_map = {'network': dict(), 'subnets': dict()}
net_attributes_map['network'].update({
'name': network.name,
'mtu': network.mtu,
'dns_domain': network.dns_domain,
'tags': network.tags,
})
for subnet_id in network.subnet_ids:
subnet = conn.network.get_subnet(subnet_id)
net_attributes_map['subnets'].update({
subnet.name: {
'name': subnet.name,
'cidr': subnet.cidr,
'gateway_ip': subnet.gateway_ip,
'host_routes': subnet.host_routes,
'dns_nameservers': subnet.dns_nameservers,
'ip_version': subnet.ip_version,
}
})
return net_attributes_map
def _cleanup_host_entry(self, entry):
# remove any tab or space excess
entry_stripped = re.sub('[ \t]+', ' ', str(entry).rstrip())

Loading…
Cancel
Save