--- /dev/null
+stdout file data
\ No newline at end of file
--- /dev/null
+test_ssh_cmd1
+test_ssh_cmd2
+test_ssh_cmd3
--- /dev/null
+test_ssh_cmd1
+test_ssh_cmd2
+test_ssh_cmd3
\ No newline at end of file
--- /dev/null
+test_ssh_cmd1
+test_ssh_cmd2
+test_ssh_cmd3 option1
\ No newline at end of file
[DEFAULT]
#The mangement ip of daisy
#When Daisy will be installed in the virtual machine, this option is required.
-daisy_management_ip=10.20.11.2
+
[BACKEND]
#Default backend types of daisy, including tecs, zenic, proton, kolla.
import os
import pytest
import mock
+import paramiko
from deploy import daisy_server
from deploy.daisy_server import (
- DaisyServer
+ DaisyServer,
+ log_from_stream,
+ log_scp
)
+from deploy.utils import WORKSPACE
-@pytest.fixture(scope="session")
+@pytest.fixture(scope="module")
def conf_file_dir(data_root):
return os.path.join(data_root, 'lab_conf')
+@pytest.fixture(scope="module")
+def common_test_file_dir(data_root):
+ return os.path.join(data_root, 'common')
+
+
+def ssh_test_file_dir():
+ return os.path.join(WORKSPACE, 'tests/data/common')
+
+
+def get_ssh_test_command_from_file(dir, file_name):
+ file_path = os.path.join(dir, file_name)
+ with open(file_path) as f:
+ return f.read()
+
+
+data1 = get_ssh_test_command_from_file(ssh_test_file_dir(), 'ssh_stream_data1.txt')
+res1 = None
+expected_ret1 = None
+res2 = 'test_res_commd'
+data2 = get_ssh_test_command_from_file(ssh_test_file_dir(), 'ssh_stream_data2.txt')
+expected_ret2 = 'test_ssh_cmd3'
+data3 = get_ssh_test_command_from_file(ssh_test_file_dir(), 'ssh_stream_data3.txt')
+
+
+@pytest.mark.parametrize('data, res, expected', [
+ (data1, res1, expected_ret1),
+ (data1, res2, expected_ret1),
+ (data2, res1, expected_ret2),
+ (data2, res2, expected_ret2),
+ (data3, res1, expected_ret1),
+ (data3, res2, expected_ret1)])
+def test_log_from_stream(data, res, expected):
+ def log_func(str):
+ print str
+ pre_val = daisy_server.BLOCK_SIZE
+ daisy_server.BLOCK_SIZE = 16
+ ret = log_from_stream(res, data, log_func)
+ daisy_server.BLOCK_SIZE = pre_val
+ assert expected == ret
+
+
+@pytest.mark.parametrize('filename, size, send', [
+ ('test_file_name', 1024, 1000),
+ ('test_file_name', 2048, 2048),
+ ('test_file_name_1234', 2097152, 2097152)])
+@mock.patch('deploy.daisy_server.LD')
+def test_log_scp(mock_LD, filename, size, send):
+ pre_val = daisy_server.LEN_OF_NAME_PART
+ daisy_server.LEN_OF_NAME_PART = 24
+ log_scp(filename, size, send)
+ daisy_server.LEN_OF_NAME_PART = pre_val
+ if size != send:
+ mock_LD.assert_not_called()
+ elif len(filename) <= 18:
+ mock_LD.assert_called_once()
+ else:
+ assert mock_LD.call_count == 2
+
+
daisy_server_info = {
'name': 'daisy',
'image': 'daisy.qcow2',
tmpdir.remove()
+@mock.patch.object(daisy_server.paramiko.SSHClient, 'connect')
+@mock.patch.object(daisy_server.DaisyServer, 'ssh_run')
+def test_connect_DaisyServer(mock_ssh_run, mock_connect, tmpdir):
+ bin_file = os.path.join(tmpdir.dirname, tmpdir.basename, bin_file_name)
+ mock_connect.return_value = 0
+ DaisyServerInst = DaisyServer(daisy_server_info['name'],
+ daisy_server_info['address'],
+ daisy_server_info['password'],
+ remote_dir,
+ bin_file,
+ adapter,
+ scenario,
+ deploy_file_name,
+ net_file_name)
+ DaisyServerInst.connect()
+ mock_ssh_run.assert_called_once_with('ls -al', check=True)
+ tmpdir.remove()
+
+
+@mock.patch.object(daisy_server.paramiko.SSHClient, 'close')
+@mock.patch.object(daisy_server.paramiko.SSHClient, 'connect')
+@mock.patch.object(daisy_server.DaisyServer, 'ssh_run')
+def test_close_DaisyServer(mock_ssh_run, mock_connect,
+ mock_close, tmpdir):
+ bin_file = os.path.join(tmpdir.dirname, tmpdir.basename, bin_file_name)
+ mock_connect.return_value = 0
+ mock_ssh_run.return_valule = 0
+ mock_close.return_value = 0
+ DaisyServerInst = DaisyServer(daisy_server_info['name'],
+ daisy_server_info['address'],
+ daisy_server_info['password'],
+ remote_dir,
+ bin_file,
+ adapter,
+ scenario,
+ deploy_file_name,
+ net_file_name)
+ DaisyServerInst.connect()
+ DaisyServerInst.close()
+ mock_close.assert_called_once_with()
+ tmpdir.remove()
+
+
+stdout1 = open(os.path.join(ssh_test_file_dir(), 'sim_stdout_file'))
+stdin1 = open(os.path.join(ssh_test_file_dir(), 'sim_stdout_file'))
+stderr1 = open(os.path.join(ssh_test_file_dir(), 'sim_stderr_file'))
+stderr2 = open(os.path.join(ssh_test_file_dir(), 'sim_stdout_file'))
+
+
+@pytest.mark.parametrize('stdout, stdin, stderr', [
+ (stdout1, stdin1, stderr1),
+ (stdout1, stdin1, stderr2)])
+@mock.patch('deploy.daisy_server.err_exit')
+@mock.patch.object(daisy_server.paramiko.SSHClient, 'connect')
+@mock.patch.object(daisy_server.paramiko.SSHClient, 'exec_command')
+@mock.patch.object(daisy_server.DaisyServer, 'ssh_run')
+def test_ssh_exec_cmd_DaisyServer(mock_ssh_run, mock_exec_command,
+ mock_connect, mock_err_exit,
+ stdout, stdin, stderr, tmpdir):
+ bin_file = os.path.join(tmpdir.dirname, tmpdir.basename, bin_file_name)
+ cmd = 'ls -l'
+ mock_connect.return_value = 0
+ mock_ssh_run.return_valule = 0
+ expect = 'stdout file data'
+ mock_exec_command.return_value = (stdin, stdout, stderr)
+ DaisyServerInst = DaisyServer(daisy_server_info['name'],
+ daisy_server_info['address'],
+ daisy_server_info['password'],
+ remote_dir,
+ bin_file,
+ adapter,
+ scenario,
+ deploy_file_name,
+ net_file_name)
+ DaisyServerInst.connect()
+ ret = DaisyServerInst.ssh_exec_cmd(cmd)
+ mock_exec_command.assert_called_once()
+ if stderr == stderr1:
+ if stdout == stdout1:
+ assert ret == expect
+ elif stderr == stderr2:
+ mock_err_exit.assert_called_once_with('SSH client error occurred')
+ tmpdir.remove()
+
+
+@pytest.mark.parametrize('check, is_recv_exit_status, expect', [
+ (False, 0, 0),
+ (True, 1, 1)])
+@mock.patch('deploy.daisy_server.err_exit')
+@mock.patch.object(daisy_server.paramiko.SSHClient, 'get_transport')
+def test_ssh_run_DaisyServer(mock_get_transport, mock_err_exit,
+ check, is_recv_exit_status,
+ expect, tmpdir):
+ class TestSession():
+
+ def __init__(self, is_recv_exit_status):
+ self.recv_data = 'recv_test_data'
+ self.recv_data_total_len = len(self.recv_data)
+ self.recv_data_read_index = 0
+ self.recv_err_data = 'recv_test_err_data'
+ self.recv_err_data_total_len = len(self.recv_err_data)
+ self.recv_err_data_read_index = 0
+ self.is_recv_exit_status = is_recv_exit_status
+ return None
+
+ def exec_command(self, cmd):
+ return 0
+
+ def recv_ready(self):
+ return True
+
+ def recv(self, size):
+ if self.recv_data_read_index < self.recv_data_total_len:
+ if size <= self.recv_data_total_len - self.recv_data_read_index:
+ cur_index = self.recv_data_read_index
+ self.recv_data_read_index += size
+ return self.recv_data[cur_index:self.recv_data_read_index]
+ else:
+ cur_index = self.recv_data_read_index
+ self.recv_data_read_index = self.recv_data_total_len
+ return self.recv_data[cur_index:]
+ else:
+ return None
+
+ def recv_stderr_ready(self):
+ return True
+
+ def recv_stderr(self, size):
+ if self.recv_err_data_read_index < self.recv_err_data_total_len:
+ if size <= self.recv_err_data_total_len - self.recv_err_data_read_index:
+ cur_index = self.recv_err_data_read_index
+ self.recv_err_data_read_index += size
+ return self.recv_err_data[cur_index:self.recv_err_data_read_index]
+ else:
+ cur_index = self.recv_err_data_read_index
+ self.recv_err_data_read_index = self.recv_err_data_total_len
+ return self.recv_err_data[cur_index:]
+ else:
+ return None
+
+ def exit_status_ready(self):
+ return True
+
+ def recv_exit_status(self):
+ return self.is_recv_exit_status
+
+ class TestTransport():
+ def __init__(self, is_recv_exit_status):
+ self.is_recv_exit_status = is_recv_exit_status
+
+ def set_keepalive(self, time):
+ self.time = time
+
+ def open_session(self):
+ return TestSession(is_recv_exit_status)
+
+ bin_file = os.path.join(tmpdir.dirname, tmpdir.basename, bin_file_name)
+ cmd = 'ls -l'
+ mock_get_transport.return_value = TestTransport(is_recv_exit_status)
+ DaisyServerInst = DaisyServer(daisy_server_info['name'],
+ daisy_server_info['address'],
+ daisy_server_info['password'],
+ remote_dir,
+ bin_file,
+ adapter,
+ scenario,
+ deploy_file_name,
+ net_file_name)
+ DaisyServerInst.ssh_client = paramiko.SSHClient()
+ DaisyServerInst.ssh_client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
+ ret = DaisyServerInst.ssh_run(cmd, check=check)
+ if check and is_recv_exit_status:
+ mock_err_exit.assert_called_once()
+ assert ret == expect
+ tmpdir.remove()
+
+
+@mock.patch.object(daisy_server.scp.SCPClient, 'get')
+@mock.patch.object(daisy_server.paramiko.SSHClient, 'get_transport')
+@mock.patch.object(daisy_server.paramiko.SSHClient, 'connect')
+@mock.patch.object(daisy_server.paramiko.SSHClient, 'exec_command')
+@mock.patch.object(daisy_server.DaisyServer, 'ssh_run')
+def test_scp_get_DaisyServer(mock_ssh_run, mock_exec_command,
+ mock_connect, mock_get_transport,
+ mock_get, tmpdir):
+ bin_file = os.path.join(tmpdir.dirname, tmpdir.basename, bin_file_name)
+ mock_connect.return_value = 0
+ mock_ssh_run.return_valule = 0
+ remote = '/remote_dir'
+ local = '.'
+ DaisyServerInst = DaisyServer(daisy_server_info['name'],
+ daisy_server_info['address'],
+ daisy_server_info['password'],
+ remote_dir,
+ bin_file,
+ adapter,
+ scenario,
+ deploy_file_name,
+ net_file_name)
+ DaisyServerInst.connect()
+ DaisyServerInst.scp_get(remote, local)
+ mock_get.assert_called_once_with(remote, local_path=local, recursive=True)
+ tmpdir.remove()
+
+
+@mock.patch.object(daisy_server.scp.SCPClient, 'put')
+@mock.patch.object(daisy_server.paramiko.SSHClient, 'get_transport')
+@mock.patch.object(daisy_server.paramiko.SSHClient, 'connect')
+@mock.patch.object(daisy_server.paramiko.SSHClient, 'exec_command')
+@mock.patch.object(daisy_server.DaisyServer, 'ssh_run')
+def test_scp_put_DaisyServer(mock_ssh_run, mock_exec_command,
+ mock_connect, mock_get_transport,
+ mock_put, tmpdir):
+ bin_file = os.path.join(tmpdir.dirname, tmpdir.basename, bin_file_name)
+ mock_connect.return_value = 0
+ mock_ssh_run.return_valule = 0
+ remote = '.'
+ local = '/tmp'
+ DaisyServerInst = DaisyServer(daisy_server_info['name'],
+ daisy_server_info['address'],
+ daisy_server_info['password'],
+ remote_dir,
+ bin_file,
+ adapter,
+ scenario,
+ deploy_file_name,
+ net_file_name)
+ DaisyServerInst.connect()
+ DaisyServerInst.scp_put(local, remote)
+ mock_put.assert_called_once_with(local, remote_path=remote, recursive=True)
+ tmpdir.remove()
+
+
@mock.patch.object(daisy_server.DaisyServer, 'ssh_exec_cmd')
def test_create_dir_DaisyServer(mock_ssh_exec_cmd, tmpdir):
remote_dir_test = '/home/daisy/test'
tmpdir.remove()
+bin_file_path1 = os.path.join('/tmp', bin_file_name)
+bin_file_path2 = os.path.join(WORKSPACE, bin_file_name)
+
+
+@pytest.mark.parametrize('bin_file', [
+ (bin_file_path1),
+ (bin_file_path2)])
@mock.patch.object(daisy_server.DaisyServer, 'delete_dir')
@mock.patch.object(daisy_server.DaisyServer, 'scp_put')
@mock.patch.object(daisy_server.DaisyServer, 'create_dir')
mock_create_dir,
mock_scp_put,
mock_delete_dir,
+ bin_file,
tmpdir):
- bin_file = os.path.join(tmpdir.dirname, tmpdir.basename, bin_file_name)
DaisyServerInst = DaisyServer(daisy_server_info['name'],
daisy_server_info['address'],
daisy_server_info['password'],
DaisyServerInst.prepare_files()
DaisyServerInst.delete_dir.assert_called_once_with(remote_dir)
DaisyServerInst.create_dir.assert_called_once_with('/home/daisy_install')
- assert DaisyServerInst.scp_put.call_count == 3
+ if bin_file == bin_file_path1:
+ assert DaisyServerInst.scp_put.call_count == 3
+ else:
+ assert DaisyServerInst.scp_put.call_count == 2
tmpdir.remove()
tmpdir.remove()
+@pytest.mark.parametrize('adapter', [
+ ('libvirt'), ('ipmi')])
@mock.patch.object(daisy_server.DaisyServer, 'ssh_run')
-def test_prepare_configurations_DaisyServer(mock_ssh_run, tmpdir):
+def test_prepare_configurations_DaisyServer(mock_ssh_run, adapter, tmpdir):
bin_file = os.path.join(tmpdir.dirname, tmpdir.basename, bin_file_name)
- adapter = 'libvirt'
DaisyServerInst = DaisyServer(daisy_server_info['name'],
daisy_server_info['address'],
daisy_server_info['password'],
net_file=os.path.join(remote_dir, net_file_name),
is_bare=1 if adapter == 'ipmi' else 0)
DaisyServerInst.prepare_configurations()
- DaisyServerInst.ssh_run.assert_called_once_with(cmd)
+ if adapter == 'libvirt':
+ DaisyServerInst.ssh_run.assert_called_once_with(cmd)
+ else:
+ DaisyServerInst.ssh_run.assert_not_called()
tmpdir.remove()
deploy_struct = get_conf_info_from_file(get_conf_file_dir(), 'deploy_baremetal.yml')
deploy_virtual_struct = get_conf_info_from_file(get_conf_file_dir(), 'deploy_virtual1.yml')
+deploy_virtual2_struct = get_conf_info_from_file(get_conf_file_dir(), 'deploy_virtual2.yml')
net_struct = get_conf_info_from_file(get_conf_file_dir(), 'network_baremetal.yml')
adapter = 'ipmi'
adapter_virtual = 'libvirt'
@pytest.mark.parametrize('deploy_struct_info', [
(deploy_struct)])
-@mock.patch('deploy.environment.err_exit')
@mock.patch('deploy.environment.ipmi_reboot_node')
-def test_reboot_nodes_BareMetalEnvironment(mock_ipmi_reboot_node, mock_err_exit,
+def test_reboot_nodes_BareMetalEnvironment(mock_ipmi_reboot_node,
deploy_struct_info, tmpdir):
work_dir = os.path.join(tmpdir.dirname, tmpdir.basename, work_dir_name)
storage_dir = os.path.join(tmpdir.dirname, tmpdir.basename, storage_dir_name)
daisy_server = copy.deepcopy(daisy_server_info)
daisy_server['image'] = os.path.join(storage_dir, daisy_server['image'])
mock_ipmi_reboot_node.return_value = True
- mock_err_exit.return_value = 0
BareMetalEnvironmentInst = BareMetalEnvironment(
deploy_struct_info, net_struct, adapter, pxe_bridge,
daisy_server, work_dir, storage_dir, scenario)
tmpdir.remove()
+@pytest.mark.parametrize('deploy_struct_info', [
+ (deploy_virtual_struct)])
+def test_reboot_nodes_err_BareMetalEnvironment(deploy_struct_info, tmpdir):
+ work_dir = os.path.join(tmpdir.dirname, tmpdir.basename, work_dir_name)
+ storage_dir = os.path.join(tmpdir.dirname, tmpdir.basename, storage_dir_name)
+ daisy_server = copy.deepcopy(daisy_server_info)
+ daisy_server['image'] = os.path.join(storage_dir, daisy_server['image'])
+ BareMetalEnvironmentInst = BareMetalEnvironment(
+ deploy_struct_info, net_struct, adapter, pxe_bridge,
+ daisy_server, work_dir, storage_dir, scenario)
+ with pytest.raises(SystemExit):
+ BareMetalEnvironmentInst.reboot_nodes()
+ tmpdir.remove()
+
+
@mock.patch.object(environment.DaisyEnvironmentBase, 'create_daisy_server_image')
@mock.patch.object(environment.BareMetalEnvironment, 'create_daisy_server_vm')
def test_create_daisy_server_BareMetalEnvironment(mock_create_daisy_server_vm, mock_create_daisy_server_image, tmpdir):
environment.create_vm.assert_called_once_with(VMDEPLOY_DAISY_SERVER_VM,
name=daisy_server['name'],
disks=[daisy_server['image']])
+ tmpdir.remove()
@mock.patch.object(environment.DaisyEnvironmentBase, 'create_daisy_server_image')
tmpdir.remove()
+@pytest.mark.parametrize('deploy_info_struct, node_num', [
+ (deploy_virtual_struct, 0),
+ (deploy_virtual_struct, 3),
+ (deploy_virtual2_struct, 0)])
@mock.patch('deploy.environment.create_vm')
@mock.patch('deploy.environment.create_virtual_disk')
-def test_create_virtual_node_VirtualEnvironment(mock_create_virtual_disk,
- mock_create_vm,
- tmpdir):
+def test_create_virtual_node_VirtualEnvironment(mock_create_virtual_disk, mock_create_vm,
+ deploy_info_struct, node_num, tmpdir):
work_dir = os.path.join(tmpdir.dirname, tmpdir.basename, work_dir_name)
storage_dir = os.path.join(tmpdir.dirname, tmpdir.basename, storage_dir_name)
daisy_server = copy.deepcopy(daisy_server_info)
daisy_server['image'] = os.path.join(storage_dir, daisy_server['image'])
VirtualEnvironmentInst = VirtualEnvironment(
- deploy_virtual_struct, net_struct, adapter_virtual, pxe_bridge_virtual,
+ deploy_info_struct, net_struct, adapter_virtual, pxe_bridge_virtual,
daisy_server, work_dir, storage_dir, scenario)
- VirtualEnvironmentInst.create_virtual_node(deploy_virtual_struct['hosts'][0])
- environment.create_virtual_disk.call_count == 2
- name = deploy_virtual_struct['hosts'][0]['name']
- template = deploy_virtual_struct['hosts'][0]['template']
- files = [os.path.join(storage_dir, name + '.qcow2'), os.path.join(storage_dir, name + '_data.qcow2')]
+ VirtualEnvironmentInst.create_virtual_node(deploy_info_struct['hosts'][node_num])
+ name = deploy_info_struct['hosts'][node_num]['name']
+ template = deploy_info_struct['hosts'][node_num]['template']
+ if deploy_info_struct == deploy_virtual_struct:
+ files = [os.path.join(storage_dir, name + '.qcow2'), os.path.join(storage_dir, name + '_data.qcow2')]
+ assert environment.create_virtual_disk.call_count == 2
+ elif deploy_info_struct == deploy_virtual2_struct:
+ files = [os.path.join(storage_dir, name + '.qcow2')]
+ assert environment.create_virtual_disk.call_count == 1
environment.create_vm.assert_called_once_with(template, name, files)
tmpdir.remove()
daisy_server, work_dir, storage_dir, scenario)
VirtualEnvironmentInst.delete_nodes()
assert environment.delete_vm_and_disk.call_count == 5
+ tmpdir.remove()
@mock.patch('deploy.environment.reboot_vm')
daisy_server, work_dir, storage_dir, scenario)
VirtualEnvironmentInst.reboot_nodes()
assert environment.reboot_vm.call_count == 5
+ tmpdir.remove()
+@pytest.mark.parametrize('isdir', [
+ (True),
+ (False)])
@mock.patch('deploy.environment.delete_virtual_network')
-def test_delete_networks_VirtualEnvironment(mock_delete_virtual_network, tmpdir):
+@mock.patch('deploy.environment.os.path.isdir')
+def test_delete_networks_VirtualEnvironment(mock_isdir, mock_delete_virtual_network, isdir, tmpdir):
work_dir = os.path.join(tmpdir.dirname, tmpdir.basename, work_dir_name)
storage_dir = os.path.join(tmpdir.dirname, tmpdir.basename, storage_dir_name)
daisy_server = copy.deepcopy(daisy_server_info)
daisy_server['image'] = os.path.join(storage_dir, daisy_server['image'])
+ mock_isdir.return_value = isdir
VirtualEnvironmentInst = VirtualEnvironment(
deploy_virtual_struct, net_struct, adapter_virtual, pxe_bridge_virtual,
daisy_server, work_dir, storage_dir, scenario)
VirtualEnvironmentInst.delete_networks()
- assert environment.delete_virtual_network.call_count == 3
+ if isdir is True:
+ assert mock_delete_virtual_network.call_count == 3
+ else:
+ mock_delete_virtual_network.assert_not_called()
+ tmpdir.remove()
@mock.patch.object(environment.DaisyEnvironmentBase, 'delete_daisy_server')
tmpdir.remove()
+@mock.patch.object(environment.DaisyServer, 'post_deploy')
+@mock.patch.object(environment.DaisyServer, 'check_openstack_installation')
+@mock.patch.object(environment.DaisyServer, 'check_os_installation')
+@mock.patch.object(environment.DaisyServer, 'install_virtual_nodes')
+@mock.patch.object(environment.DaisyServer, 'prepare_host_and_pxe')
+@mock.patch.object(environment.DaisyServer, 'copy_new_deploy_config')
+@mock.patch.object(environment.DaisyServer, 'prepare_cluster')
+@mock.patch.object(environment.VirtualEnvironment, '_post_deploy')
+@mock.patch.object(environment.VirtualEnvironment, 'reboot_nodes')
+@mock.patch.object(environment.VirtualEnvironment, 'create_nodes')
+def test_deploy_VirtualEnvironment(mock_create_nodes, mock_reboot_nodes,
+ mock__post_deploy, mock_prepare_cluster,
+ mock_copy_new_deploy_config, mock_prepare_host_and_pxe,
+ mock_install_virtual_nodes, mock_check_os_installation,
+ mock_check_openstack_installation, mock_post_deploy,
+ tmpdir):
+ work_dir = os.path.join(tmpdir.dirname, tmpdir.basename, work_dir_name)
+ storage_dir = os.path.join(tmpdir.dirname, tmpdir.basename, storage_dir_name)
+ daisy_server = copy.deepcopy(daisy_server_info)
+ daisy_server['image'] = os.path.join(storage_dir, daisy_server['image'])
+ remote_dir = '/home/daisy'
+ bin_file = os.path.join(tmpdir.dirname, tmpdir.basename, 'opnfv.bin')
+ deploy_file_name = 'final_deploy.yml'
+ net_file_name = 'network_virtual1.yml'
+ deploy_file = os.path.join(get_conf_file_dir(), 'deploy_virtual1.yml')
+ net_file = os.path.join(get_conf_file_dir(), 'network_virtual1.yml')
+ VirtualEnvironmentInst = VirtualEnvironment(
+ deploy_virtual_struct, net_struct, adapter_virtual, pxe_bridge_virtual,
+ daisy_server, work_dir, storage_dir, scenario)
+ VirtualEnvironmentInst.server = DaisyServer(
+ daisy_server['name'],
+ daisy_server['address'],
+ daisy_server['password'],
+ remote_dir,
+ bin_file,
+ adapter,
+ scenario,
+ deploy_file_name,
+ net_file_name)
+ VirtualEnvironmentInst.deploy(deploy_file, net_file)
+ mock_create_nodes.assert_called_once()
+ assert mock_reboot_nodes.call_count == 2
+ mock__post_deploy.assert_called_once()
+ mock_prepare_cluster.assert_called_once()
+ mock_copy_new_deploy_config.assert_called_once()
+ mock_prepare_host_and_pxe.assert_called_once()
+ mock_install_virtual_nodes.assert_called_once()
+ mock_check_os_installation.assert_called_once()
+ mock_check_openstack_installation.assert_called_once()
+ mock_post_deploy.assert_called_once()
+ tmpdir.remove()
+
+
+@pytest.mark.parametrize('status', [
+ (True),
+ (False)])
@mock.patch('deploy.environment.commands.getstatusoutput')
-def test_post_deploy_VirtualEnvironment(mock_getstatusoutput, tmpdir):
+@mock.patch('deploy.environment.LW')
+@mock.patch('deploy.environment.LI')
+def test__post_deploy_VirtualEnvironment(mock_LI, mock_LW,
+ mock_getstatusoutput, status, tmpdir):
work_dir = os.path.join(tmpdir.dirname, tmpdir.basename, work_dir_name)
storage_dir = os.path.join(tmpdir.dirname, tmpdir.basename, storage_dir_name)
daisy_server = copy.deepcopy(daisy_server_info)
daisy_server['image'] = os.path.join(storage_dir, daisy_server['image'])
- mock_getstatusoutput.return_value = (0, 'sucess')
+ mock_getstatusoutput.return_value = (status, 'success')
VirtualEnvironmentInst = VirtualEnvironment(
deploy_virtual_struct, net_struct, adapter_virtual, pxe_bridge_virtual,
daisy_server, work_dir, storage_dir, scenario)
VirtualEnvironmentInst._daisy_server_net = 'daisy1'
VirtualEnvironmentInst._daisy_os_net = 'daisy2'
assert environment.commands.getstatusoutput.call_count == 4
+ if status:
+ mock_LW.assert_called()
+ else:
+ mock_LI.assert_called()
+ tmpdir.remove()
import os
import pytest
import mock
+import shutil
from deploy import utils
from deploy.utils import (
+ get_logger,
+ save_log_to_file,
err_exit,
+ log_bar,
check_sudo_privilege,
check_file_exists,
make_file_executable,
confirm_dir_exists,
- check_scenario_valid
+ update_config,
+ ipmi_reboot_node,
+ run_shell,
+ check_scenario_valid,
+ LI
)
+@pytest.fixture(scope="module")
+def daisy_conf_file_dir(data_root):
+ return os.path.join(data_root, 'daisy_conf')
+
+
+@mock.patch.object(utils.logging.Logger, 'addHandler')
+def test_get_logger(mock_addHandler):
+ get_logger()
+ mock_addHandler.assert_called_once()
+
+
+@mock.patch.object(utils.logging.Logger, 'addHandler')
+def test_save_log_to_file(mock_addHandler, tmpdir):
+ log_file = os.path.join(tmpdir.dirname, tmpdir.basename, 'test_log.txt')
+ save_log_to_file(log_file)
+ mock_addHandler.assert_called_once()
+ tmpdir.remove()
+
+
def test_err_exit():
message = 'test error msg!'
with pytest.raises(SystemExit):
err_exit(message)
+def test_log_bar():
+ log_bar('test_messgae', log_func=LI)
+
+
@mock.patch('deploy.utils.err_exit')
@mock.patch('deploy.utils.os.getuid')
def test_check_sudo_privilege(mock_getuid, mock_err_exit):
mock_getuid.return_value = 1
check_sudo_privilege()
- utils.err_exit.assert_called_once_with('You need run this script with sudo privilege')
+ mock_err_exit.assert_called_once_with('You need run this script with sudo privilege')
-@pytest.mark.parametrize('test_file_name', [
- ('no_exist_file'),
- ('exist_file')])
-def test_check_file_exists(tmpdir, test_file_name):
- try:
+@pytest.mark.parametrize('test_file_name, include_dirname', [
+ ('no_exist_file', True),
+ ('exist_file', True),
+ ('no_exist_file', False),
+ ('exist_file', False)])
+@mock.patch('deploy.utils.err_exit')
+def test_check_file_exists(mock_err_exit, tmpdir, test_file_name, include_dirname):
+ if include_dirname:
file_path = os.path.join(tmpdir.dirname, tmpdir.basename, test_file_name)
if test_file_name == 'exist_file':
os.mknod(file_path)
- check_file_exists(file_path)
- except SystemExit:
- if test_file_name == 'exist_file':
- assert 0
else:
- if test_file_name == 'no_exist_file':
- assert 0
- finally:
- tmpdir.remove()
-
-
-@pytest.mark.parametrize('test_file_name', [
- ('no_exist_file'),
- ('no_exe_file'),
- ('exe_file')])
-def test_make_file_executable(tmpdir, test_file_name):
- try:
+ file_path = test_file_name
+ check_file_exists(file_path)
+ if include_dirname is True:
+ if test_file_name == 'exist_file':
+ mock_err_exit.assert_not_called()
+ else:
+ mock_err_exit.assert_called_once()
+ if include_dirname is False:
+ mock_err_exit.assert_called_once()
+ tmpdir.remove()
+
+
+@pytest.mark.parametrize('test_file_name, status, include_dir', [
+ ('no_exist_file', False, False),
+ ('no_exe_file', False, True),
+ ('no_exe_file', True, True),
+ ('exe_file', False, True)])
+@mock.patch('deploy.utils.commands.getstatusoutput')
+@mock.patch('deploy.utils.err_exit')
+def test_make_file_executable(mock_err_exit, mock_getstatusoutput,
+ tmpdir, test_file_name,
+ status, include_dir):
+ if include_dir:
file_path = os.path.join(tmpdir.dirname, tmpdir.basename, test_file_name)
- if test_file_name == 'no_exe_file':
- os.mknod(file_path)
- if test_file_name == 'exe_file':
- os.mknod(file_path, 0700)
- make_file_executable(file_path)
- except SystemExit:
- if test_file_name == 'no_exe_file' or test_file_name == 'exe_file':
- assert 0
else:
- if test_file_name == 'no_exist_file':
- assert 0
- finally:
- if test_file_name == 'no_exe_file' or test_file_name == 'exe_file':
- assert os.access(file_path, os.X_OK)
- tmpdir.remove()
+ file_path = test_file_name
+ if test_file_name == 'no_exe_file':
+ os.mknod(file_path)
+ if test_file_name == 'exe_file':
+ os.mknod(file_path, 0700)
+ output = 'test_out'
+ mock_getstatusoutput.return_value = (status, output)
+ make_file_executable(file_path)
+ if test_file_name == 'exe_file':
+ mock_err_exit.assert_not_called()
+ assert os.access(file_path, os.X_OK)
+ if test_file_name == 'no_exe_file' and status is False:
+ mock_err_exit.assert_not_called()
+ if test_file_name == 'no_exe_file' and status is True:
+ mock_err_exit.assert_called()
+ if test_file_name == 'no_exist_file':
+ mock_err_exit.assert_called()
+ tmpdir.remove()
@pytest.mark.parametrize('test_dir_name', [
tmpdir.remove()
+def test_update_config(daisy_conf_file_dir, tmpdir):
+ src_daisy_conf_file = os.path.join(daisy_conf_file_dir, 'daisy.conf')
+ dst_daisy_conf_file = os.path.join(tmpdir.dirname, tmpdir.basename, 'daisy.conf')
+ shutil.copyfile(src_daisy_conf_file, dst_daisy_conf_file)
+ key = 'daisy_management_ip'
+ value = '10.20.11.2'
+ update_line = 'daisy_management_ip = 10.20.11.2'
+ is_match = False
+ update_config(dst_daisy_conf_file, key, value, section='DEFAULT')
+ with open(dst_daisy_conf_file) as f:
+ lines = f.readlines()
+ for line in lines:
+ line_content = line.strip()
+ if update_line in line_content:
+ is_match = True
+ break
+ assert is_match
+ tmpdir.remove()
+
+
+@pytest.mark.parametrize('boot_source, status', [
+ (None, 0),
+ (None, 1),
+ ('test_source', 0),
+ ('test_source', 1)])
+@mock.patch('deploy.utils.err_exit')
+@mock.patch.object(utils.commands, 'getstatusoutput')
+def test_ipmi_reboot_node(mock_getstatusoutput, mock_err_exit,
+ boot_source, status):
+ host = '192.168.1.11'
+ user = 'testuser'
+ passwd = 'testpass'
+ output = 'test_out'
+ mock_getstatusoutput.return_value = (status, output)
+ ipmi_reboot_node(host, user, passwd, boot_source=boot_source)
+ if boot_source:
+ assert mock_getstatusoutput.call_count == 2
+ if status:
+ assert mock_err_exit.call_count == 2
+ else:
+ mock_getstatusoutput.called_once()
+ if status:
+ mock_err_exit.called_once()
+
+
+@pytest.mark.parametrize('cmd, check, expect', [
+ ('cd /home', False, 0),
+ ('cd /home', True, 0),
+ ('test_command', False, 127),
+ ('test_command', True, 127)])
+@mock.patch('deploy.utils.err_exit')
+def test_run_shell(mock_err_exit, cmd, check, expect):
+ ret = run_shell(cmd, check=check)
+ if check:
+ if cmd == 'cd /home':
+ mock_err_exit.assert_not_called()
+ elif cmd == 'test_command':
+ mock_err_exit.assert_called_once()
+ assert ret == expect
+
+
@pytest.mark.parametrize('scenario', [
('os-nosdn-nofeature-ha')])
@mock.patch("deploy.utils.err_exit")
Code Review / daisy.git / commitdiff