python utility to manage a tripleo based cloud
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.

603 lines
25 KiB

# Copyright 2021 Red Hat, Inc.
#
# 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 fixtures
import mock
import os
from pathlib import Path
import shutil
import subprocess
import time
from tripleoclient import heat_launcher
from tripleoclient.exceptions import HeatPodMessageQueueException
from tripleoclient.tests import base
from tripleoclient import utils
class TestHeatPodLauncher(base.TestCase):
def setUp(self):
super(TestHeatPodLauncher, self).setUp()
self.run = mock.patch('subprocess.run').start()
self.call = mock.patch('subprocess.call').start()
self.check_call = mock.patch('subprocess.check_call').start()
self.check_output = mock.patch('subprocess.check_output').start()
self.templates_dir = mock.patch(
'tripleoclient.heat_launcher.DEFAULT_TEMPLATES_DIR',
os.path.join(os.path.dirname(__file__),
'..', '..', 'templates')).start()
self.heat_dir = self.useFixture(fixtures.TempDir()).path
self.addCleanup(mock.patch.stopall)
def get_launcher(self, **kwargs):
return heat_launcher.HeatPodLauncher(
heat_dir=self.heat_dir,
use_tmp_dir=False,
**kwargs)
def check_calls(self, check_call, mock_obj):
for call in mock_obj.call_args_list:
call_str = ' '.join(call.args[0])
if check_call in call_str:
return True
return False
def test_rm_heat_launcher(self):
self.assertIsInstance(self.get_launcher(rm_heat=True),
heat_launcher.HeatPodLauncher)
def test_chcon(self):
launcher = self.get_launcher()
launcher._chcon()
self.check_calls('chcon', self.check_call)
self.check_calls(launcher.heat_dir, self.check_call)
def test_fetch_container_image(self):
launcher = self.get_launcher(skip_heat_pull=True)
launcher._fetch_container_image()
self.assertFalse(self.check_calls('podman pull', self.check_output))
launcher = self.get_launcher(skip_heat_pull=False)
launcher._fetch_container_image()
self.assertTrue(self.check_calls('podman pull', self.check_output))
@mock.patch('tripleoclient.heat_launcher.HeatPodLauncher._decode')
def test_get_pod_state(self, mock_decode):
launcher = self.get_launcher()
launcher.get_pod_state()
self.check_calls('podman pod inspect', self.run)
self.assertTrue(mock_decode.called)
@mock.patch(
'tripleoclient.heat_launcher.HeatPodLauncher._write_heat_config')
@mock.patch('tripleoclient.heat_launcher.HeatPodLauncher._write_heat_pod')
@mock.patch('tripleoclient.heat_launcher.HeatPodLauncher.get_pod_state')
def test_lauch_heat(
self, mock_get_pod_state, mock_write_heat_pod,
mock_write_heat_config):
launcher = self.get_launcher()
mock_get_pod_state.return_value = 'Running'
launcher.launch_heat()
self.assertFalse(mock_write_heat_pod.called)
self.assertFalse(self.check_calls('podman play kube', self.check_call))
mock_get_pod_state.return_value = 'Exited'
launcher.launch_heat()
self.assertTrue(mock_write_heat_pod.called)
self.assertTrue(self.check_calls('podman play kube', self.check_call))
mock_get_pod_state.return_value = ''
launcher.launch_heat()
self.assertTrue(mock_write_heat_pod.called)
self.assertTrue(self.check_calls('podman play kube', self.check_call))
@mock.patch('tripleoclient.heat_launcher.HeatPodLauncher.do_restore_db')
@mock.patch('tripleoclient.heat_launcher.HeatPodLauncher.database_exists')
def test_heat_db_sync(
self, mock_db_exists, mock_do_restore_db):
launcher = self.get_launcher()
mock_db_exists.return_value = True
launcher.heat_db_sync(restore_db=False)
self.assertFalse(self.check_calls('create database', self.check_call))
self.assertFalse(self.check_calls('create user', self.check_call))
self.assertFalse(self.check_calls('grant all', self.check_call))
self.assertFalse(self.check_calls('flush priv', self.check_call))
self.assertTrue(self.check_calls('heat-manage', self.check_call))
self.assertFalse(mock_do_restore_db.called)
self.check_call.reset_mock()
mock_db_exists.return_value = True
launcher.heat_db_sync(restore_db=True)
self.assertFalse(self.check_calls('create database', self.check_call))
self.assertFalse(self.check_calls('create user', self.check_call))
self.assertFalse(self.check_calls('grant all', self.check_call))
self.assertFalse(self.check_calls('flush priv', self.check_call))
self.assertTrue(self.check_calls('heat-manage', self.check_call))
self.assertTrue(mock_do_restore_db.called)
self.check_call.reset_mock()
mock_db_exists.return_value = False
launcher.heat_db_sync(restore_db=True)
self.assertTrue(self.check_calls('create database', self.check_call))
self.assertTrue(self.check_calls('create user', self.check_call))
self.assertTrue(self.check_calls('grant all', self.check_call))
self.assertTrue(self.check_calls('flush priv', self.check_call))
self.assertTrue(self.check_calls('heat-manage', self.check_call))
self.assertTrue(mock_do_restore_db.called)
self.check_call.reset_mock()
mock_do_restore_db.reset_mock()
mock_db_exists.return_value = False
launcher.heat_db_sync(restore_db=False)
self.assertTrue(self.check_calls('create database', self.check_call))
self.assertTrue(self.check_calls('create user', self.check_call))
self.assertTrue(self.check_calls('grant all', self.check_call))
self.assertTrue(self.check_calls('flush priv', self.check_call))
self.assertTrue(self.check_calls('heat-manage', self.check_call))
self.assertFalse(mock_do_restore_db.called)
@mock.patch('os.unlink')
@mock.patch('tripleoclient.heat_launcher.HeatPodLauncher.untar_file')
@mock.patch('glob.glob')
def test_do_restore_db(
self, mock_glob, mock_untar, mock_unlink):
launcher = self.get_launcher()
one = Path(os.path.join(launcher.heat_dir,
'heat-db-dump-one.tar.bz2'))
two = Path(os.path.join(launcher.heat_dir,
'heat-db-dump-two.tar.bz2'))
three = Path(os.path.join(launcher.heat_dir,
'heat-db-dump-three.tar.bz2'))
now = time.time()
one.touch()
two.touch()
three.touch()
os.utime(str(one), (now, 1000))
os.utime(str(two), (now, 2000))
os.utime(str(three), (now, 3000))
mock_glob.return_value = [str(one), str(two), str(three)]
def untar(path, dir):
p = Path(path.rstrip('.tar.bz2'))
p.touch()
mock_untar.side_effect = untar
mock_open = mock.mock_open()
with mock.patch('six.moves.builtins.open', mock_open):
# pylint: disable=bad-str-strip-call
launcher.do_restore_db()
self.assertEqual(mock.call(str(three), launcher.heat_dir),
mock_untar.call_args)
self.assertEqual(mock.call(launcher.heat_dir + '/heat-db.sql'),
mock_unlink.call_args)
mock_open.assert_called_with(launcher.heat_dir + '/heat-db.sql') # noqa
self.assertTrue(self.check_call('mysql heat', self.run))
mock_unlink.reset_mock()
self.run.reset_mock()
two.touch()
mock_open = mock.mock_open()
with mock.patch('six.moves.builtins.open', mock_open):
# pylint: disable=bad-str-strip-call
launcher.do_restore_db()
self.assertEqual(mock.call(str(two), launcher.heat_dir),
mock_untar.call_args)
self.assertEqual(mock.call(launcher.heat_dir + '/heat-db.sql'),
mock_unlink.call_args)
mock_open.assert_called_with(launcher.heat_dir + '/heat-db.sql') # noqa
self.assertTrue(self.check_call('mysql heat', self.run))
@mock.patch('tripleoclient.heat_launcher.HeatPodLauncher.tar_file')
def test_do_backup_db(self, mock_tar):
launcher = self.get_launcher()
p = Path(os.path.join(launcher.heat_dir, 'heat-db.sql'))
p.touch()
self.assertRaises(Exception, launcher.do_backup_db, str(p))
p.unlink()
launcher.do_backup_db()
mock_tar.assert_called_with(str(p))
self.assertTrue(self.check_calls('mysqldump heat', self.run))
def test_pod_exists(self):
launcher = self.get_launcher()
self.assertTrue(launcher.pod_exists())
self.check_calls('pod inspect', self.check_call)
self.check_call.reset_mock()
self.check_call.side_effect = subprocess.CalledProcessError(1, 'test')
self.assertFalse(launcher.pod_exists())
self.check_calls('pod inspect', self.check_call)
@mock.patch('os.path.exists')
@mock.patch('tripleoclient.heat_launcher.HeatPodLauncher.tar_file')
@mock.patch(
'tripleoclient.heat_launcher.HeatPodLauncher._read_heat_config')
@mock.patch('tripleoclient.heat_launcher.HeatPodLauncher.pod_exists')
@mock.patch('tripleoclient.heat_launcher.HeatPodLauncher.do_backup_db')
@mock.patch('tripleoclient.heat_launcher.HeatPodLauncher.database_exists')
def test_rm_heat(self, mock_db_exists, mock_backup_db, mock_pod_exists,
mock_read_heat_config, mock_tar, mock_exists):
launcher = self.get_launcher()
launcher.log_dir = '/log'
mock_db_exists.return_value = True
mock_pod_exists.return_value = True
mock_exists.return_value = True
mock_read_heat_config.return_value = {
'DEFAULT': {
'log_file': 'heat-log'}}
launcher.rm_heat()
mock_backup_db.assert_called()
self.check_calls('drop database heat', self.check_call)
self.check_calls('drop user', self.check_call)
mock_pod_exists.assert_called()
self.check_calls('podman pod rm -f', self.call)
mock_read_heat_config.assert_called()
mock_tar.assert_called_with('/log/heat-log')
mock_backup_db.reset_mock()
self.call.reset_mock()
mock_tar.reset_mock()
mock_db_exists.return_value = False
mock_pod_exists.return_value = False
mock_exists.return_value = False
launcher.rm_heat()
mock_backup_db.assert_not_called()
self.call.assert_not_called()
mock_tar.assert_not_called()
mock_backup_db.reset_mock()
self.call.reset_mock()
mock_tar.reset_mock()
mock_exists.reset_mock()
mock_db_exists.return_value = False
mock_pod_exists.return_value = True
mock_exists.return_value = True
launcher.rm_heat(backup_db=False)
mock_backup_db.assert_not_called()
self.check_calls('podman pod rm -f', self.call)
@mock.patch('tripleoclient.heat_launcher.HeatPodLauncher.get_pod_state')
@mock.patch('tripleoclient.heat_launcher.HeatPodLauncher.pod_exists')
def test_stop_heat(self, mock_pod_exists, mock_pod_state):
launcher = self.get_launcher()
mock_pod_exists.return_value = True
mock_pod_state.return_value = 'Running'
launcher.stop_heat()
mock_pod_exists.assert_called()
mock_pod_state.assert_called()
self.check_calls('podman pod stop', self.check_call)
self.check_call.reset_mock()
mock_pod_exists.reset_mock()
mock_pod_state.reset_mock()
mock_pod_state.return_value = 'Exited'
mock_pod_exists.return_value = True
launcher.stop_heat()
mock_pod_exists.assert_called()
mock_pod_state.assert_called()
self.check_call.assert_not_called()
self.check_call.reset_mock()
mock_pod_exists.reset_mock()
mock_pod_state.reset_mock()
mock_pod_state.return_value = 'Exited'
mock_pod_exists.return_value = False
launcher.stop_heat()
mock_pod_exists.assert_called()
mock_pod_state.assert_not_called()
self.check_call.assert_not_called()
def test_check_message_bus(self):
launcher = self.get_launcher()
launcher.check_message_bus()
self.check_calls('rabbitmqctl list_queues', self.check_call)
self.check_call.reset_mock()
self.check_call.side_effect = subprocess.CalledProcessError(1, 'test')
self.assertRaises(subprocess.CalledProcessError,
launcher.check_message_bus)
@mock.patch(
'tripleoclient.heat_launcher.HeatPodLauncher._get_ctlplane_ip')
def test_check_database(self, mock_ctlplane_ip):
launcher = self.get_launcher()
mock_ctlplane_ip.return_value = '1.1.1.1'
self.assertTrue(launcher.check_database())
mock_ctlplane_ip.assert_called()
self.check_calls('show databases', self.check_call)
self.check_call.reset_mock()
mock_ctlplane_ip.reset_mock()
self.check_call.side_effect = subprocess.CalledProcessError(1, '/test')
self.assertRaises(subprocess.CalledProcessError,
launcher.check_database)
def test_database_exists(self):
launcher = self.get_launcher()
self.check_output.return_value = 'heat'
self.assertTrue(launcher.database_exists())
self.check_calls('show databases like "heat"', self.check_output)
self.check_output.reset_mock()
self.check_output.return_value = 'nova'
self.assertFalse(launcher.database_exists())
self.check_calls('show databases like "heat"', self.check_output)
@mock.patch('tripleoclient.heat_launcher.HeatPodLauncher.pod_exists')
def test_kill_heat(self, mock_pod_exists):
launcher = self.get_launcher()
mock_pod_exists.return_value = True
launcher.kill_heat(0)
self.check_calls('podman pod kill', self.call)
mock_pod_exists.assert_called()
mock_pod_exists.reset_mock()
self.call.reset_mock()
mock_pod_exists.return_value = False
launcher.kill_heat(0)
mock_pod_exists.assert_called()
self.call.assert_not_called()
def test_decode(self):
launcher = self.get_launcher()
mock_encoded = mock.Mock()
mock_decoded = mock.Mock()
mock_encoded.decode.return_value = mock_decoded
mock_decoded.endswith.return_value = False
launcher._decode(mock_encoded)
mock_encoded.decode.assert_called_with('utf-8')
self.assertEqual('test', launcher._decode(b'test\n'))
@mock.patch('tripleoclient.heat_launcher.HeatPodLauncher._decode')
def test_get_transport_url(self, mock_decode):
launcher = self.get_launcher()
mock_decode.side_effect = ['user', 'password', 'fqdn_ctlplane', 'port']
self.assertEqual("rabbit://user:password@fqdn_ctlplane:port/?ssl=0",
launcher._get_transport_url())
@mock.patch(
'tripleoclient.heat_launcher.HeatPodLauncher._get_ctlplane_vip')
def test_get_db_connection(self, mock_ctlplane_vip):
launcher = self.get_launcher()
mock_ctlplane_vip.return_value = '1.1.1.1'
self.assertEqual(
'mysql+pymysql://'
'heat:heat@1.1.1.1/heat?read_default_file='
'/etc/my.cnf.d/tripleo.cnf&read_default_group=tripleo',
launcher._get_db_connection())
@mock.patch('tripleoclient.heat_launcher.HeatPodLauncher._decode')
def test_get_ctlplane_vip(self, mock_decode):
launcher = self.get_launcher()
self.check_output.return_value = '1.1.1.1'
launcher._get_ctlplane_vip()
self.check_calls('sudo hiera controller_virtual_ip', self.check_output)
mock_decode.assert_called_with('1.1.1.1')
@mock.patch('tripleoclient.heat_launcher.HeatPodLauncher._decode')
def test_get_ctlplane_ip(self, mock_decode):
launcher = self.get_launcher()
self.check_output.return_value = '1.1.1.1'
launcher._get_ctlplane_ip()
self.check_calls('sudo hiera ctlplane', self.check_output)
mock_decode.assert_called_with('1.1.1.1')
@mock.patch('multiprocessing.cpu_count')
def test_get_num_engine_workers(self, mock_cpu_count):
launcher = self.get_launcher()
mock_cpu_count.return_value = 4
self.assertEqual(2, launcher._get_num_engine_workers())
def test_wait_for_message_queue(self):
launcher = self.get_launcher()
wait_mq = launcher.wait_for_message_queue.__wrapped__
self.check_output.return_value = 'heat'
wait_mq(launcher)
self.check_output.reset_mock()
self.check_output.return_value = 'test'
self.assertRaises(HeatPodMessageQueueException, wait_mq, launcher)
def test_get_log_file_path(self):
launcher = self.get_launcher()
launcher.timestamp = '1111'
self.assertEqual('heat-1111.log', launcher._get_log_file_path())
@mock.patch('configparser.ConfigParser')
def test_read_heat_config(self, mock_config_parser):
launcher = self.get_launcher()
mock_cp = mock.Mock()
mock_cp.read.return_value = 'test'
mock_config_parser.return_value = mock_cp
self.assertEqual(mock_cp, launcher._read_heat_config())
mock_config_parser.assert_called()
mock_cp.read.assert_called_with(launcher.config_file)
@mock.patch('tripleoclient.heat_launcher.'
'HeatPodLauncher._get_num_engine_workers')
@mock.patch(
'tripleoclient.heat_launcher.HeatPodLauncher._get_db_connection')
@mock.patch(
'tripleoclient.heat_launcher.HeatPodLauncher._get_transport_url')
def test_write_heat_config(self, mock_get_transport_url, mock_get_db_conn,
mock_num_engine_workers):
launcher = self.get_launcher()
launcher.api_port = '1234'
launcher.log_file = '/log/heat'
mock_get_transport_url.return_value = 'transport-url'
mock_get_db_conn.return_value = 'db-connection'
mock_num_engine_workers.return_value = 'num-engine-workers'
launcher._write_heat_config()
with open(launcher.config_file) as f:
config = f.read()
self.assertIn('num_engine_workers = num-engine-workers\n', config)
self.assertIn('connection = db-connection\n', config)
self.assertIn('transport_url=transport-url\n', config)
self.assertIn('bind_port = 1234\n', config)
self.assertIn('log_file = /log/heat\n', config)
def test_write_heat_pod(self):
launcher = self.get_launcher()
launcher.install_dir = 'install-dir'
launcher.host = '1.1.1.1'
launcher.api_port = '1234'
launcher.api_container_image = 'api-image'
launcher.engine_container_image = 'engine-image'
launcher._write_heat_pod()
with open(os.path.join(launcher.heat_dir, 'heat-pod.yaml')) as f:
pod = f.read()
self.assertIn('hostPort: 1234', pod)
self.assertIn('hostIP: 1.1.1.1', pod)
self.assertIn('image: api-image', pod)
self.assertIn('image: engine-image', pod)
class TestHeatPodLauncherUtils(base.TestCase):
def setUp(self):
super(TestHeatPodLauncherUtils, self).setUp()
def test_rm_heat(self):
launcher = mock.Mock()
utils.rm_heat(launcher)
launcher.rm_heat.assert_called_once_with(False)
launcher.reset_mock()
utils.rm_heat(launcher, True)
launcher.rm_heat.assert_called_once_with(True)
launcher.reset_mock()
utils.rm_heat(launcher, False)
launcher.rm_heat.assert_called_once_with(False)
def test_kill_heat(self):
launcher = mock.Mock()
utils.kill_heat(launcher)
launcher.kill_heat.assert_called_once_with(None)
launcher.reset_mock()
utils._heat_pid = 111
utils.kill_heat(launcher)
launcher.kill_heat.assert_called_once_with(111)
launcher.reset_mock()
utils.kill_heat(launcher)
launcher.kill_heat.assert_called_once_with(111)
launcher.reset_mock()
utils.kill_heat(launcher)
launcher.kill_heat.assert_called_once_with(111)
@mock.patch('tripleoclient.heat_launcher.HeatPodLauncher')
@mock.patch('tripleoclient.heat_launcher.HeatNativeLauncher')
@mock.patch('tripleoclient.heat_launcher.HeatContainerLauncher')
def test_get_heat_launcher(self, mock_container, mock_native, mock_pod):
utils.get_heat_launcher('pod', 1, 2, 3, a='a', b='b', c='c')
mock_pod.assert_called_once_with(1, 2, 3, a='a', b='b', c='c')
utils.get_heat_launcher('native', 1, 2, 3, a='a', b='b', c='c')
mock_native.assert_called_once_with(1, 2, 3, a='a', b='b', c='c')
utils.get_heat_launcher('container', 1, 2, 3, a='a', b='b', c='c')
mock_container.assert_called_once_with(1, 2, 3, a='a', b='b', c='c')
def test_heat_api_port(self):
test_port = utils.test_heat_api_port.__wrapped__
mock_socket = mock.Mock()
host = '1.1.1.1'
port = 1234
test_port(mock_socket, host, port)
mock_socket.connect.assert_called_once_with((host, port))
@mock.patch('tripleoclient.utils.test_heat_api_port')
@mock.patch('tripleo_common.utils.heat.local_orchestration_client')
@mock.patch('socket.socket')
@mock.patch('tripleoclient.utils.get_heat_launcher')
def test_launch_heat(self, mock_get_heat_launcher, mock_socket,
mock_local_client, mock_test_port):
utils._local_orchestration_client = 'client'
self.assertEqual('client', utils.launch_heat())
mock_get_heat_launcher.assert_not_called()
utils._local_orchestration_client = None
mock_launcher = mock.Mock()
mock_launcher.api_port = 1234
mock_get_heat_launcher.return_value = mock_launcher
mock_socket.return_value = 'socket'
utils.launch_heat()
mock_get_heat_launcher.assert_called_once()
mock_launcher.check_database.assert_called_once_with()
mock_launcher.check_message_bus.assert_called_once_with()
mock_launcher.heat_db_sync.assert_called_once_with(False)
mock_launcher.launch_heat.assert_called_once_with()
mock_test_port.assert_called_once_with(
'socket', mock_launcher.host,
int(mock_launcher.api_port))
mock_launcher.wait_for_message_queue.assert_called_once_with()
mock_local_client.assert_called_once_with(
mock_launcher.host,
mock_launcher.api_port)
class TestHeatNativeLauncher(base.TestCase):
def setUp(self):
super(TestHeatNativeLauncher, self).setUp()
self.run = mock.patch('subprocess.run').start()
self.popen = mock.patch('subprocess.Popen').start()
self.mock_popen = mock.Mock()
self.mock_popen.communicate.return_value = ("", "")
self.popen.return_value = self.mock_popen
self.getpwnam = mock.patch('pwd.getpwnam').start()
self.getgrnam = mock.patch('grp.getgrnam').start()
self.chown = mock.patch('os.chown').start()
self.templates_dir = mock.patch(
'tripleoclient.heat_launcher.DEFAULT_TEMPLATES_DIR',
os.path.join(os.path.dirname(__file__),
'..', '..', 'templates')).start()
self.heat_dir = self.useFixture(fixtures.TempDir()).path
self.tmp_dir = self.useFixture(fixtures.TempDir()).path
self.addCleanup(mock.patch.stopall)
def get_launcher(self, **kwargs):
return heat_launcher.HeatNativeLauncher(
heat_dir=self.heat_dir,
use_tmp_dir=True,
use_root=True,
**kwargs)
def test_heat_dir_no_exist(self):
shutil.rmtree(self.heat_dir)
launcher = self.get_launcher()
self.assertNotEqual(self.heat_dir, launcher.install_dir)
@mock.patch('tempfile.mkdtemp')
def test_get_launcher(self, mock_mkdtemp):
mock_mkdtemp.return_value = self.tmp_dir
def test_install_dir():
mock_mkdtemp.assert_not_called()
return ("", "")
# Test that tempfile.mkdtemp is *not* called before the tmpfs is setup,
# otherwise the tmpfs will cause the temp dir to be lost
self.mock_popen.communicate.side_effect = test_install_dir
self.get_launcher()