path: root/test/unittests/test_report_collect.py

from subprocess import TimeoutExpired
from crmsh.report import collect, constants
import crmsh.log

import unittest
from unittest import mock


class TestCollect(unittest.TestCase):

    @mock.patch('logging.Logger.warning')
    @mock.patch('os.path.isfile')
    def test_get_pcmk_log_no_config(self, mock_isfile, mock_warning):
        mock_isfile.side_effect = [False, False, False]
        res = collect.get_pcmk_log()
        self.assertEqual(res, "")
        mock_isfile.assert_has_calls([
            mock.call(constants.PCMKCONF),
            mock.call("/var/log/pacemaker/pacemaker.log"),
            mock.call("/var/log/pacemaker.log")
            ])
        mock_warning.assert_called_once_with("No valid pacemaker log file found")

    @mock.patch('logging.Logger.warning')
    @mock.patch('crmsh.utils.read_from_file')
    @mock.patch('os.path.isfile')
    def test_get_pcmk_log(self, mock_isfile, mock_read, mock_warning):
        mock_isfile.return_value = True
        mock_read.return_value = """
# has been enabled, those as well). This log is of more use to developers and
# advanced system administrators, and when reporting problems.
PCMK_logfile=/var/log/pacemaker/pacemaker.log

# Set the permissions on the above log file to owner/group read/write
        """
        res = collect.get_pcmk_log()
        self.assertEqual(res, "/var/log/pacemaker/pacemaker.log")
        mock_isfile.assert_has_calls([
            mock.call(constants.PCMKCONF),
            mock.call("/var/log/pacemaker/pacemaker.log")
            ])
        mock_read.assert_called_once_with(constants.PCMKCONF)

    @mock.patch('crmsh.report.utils.dump_logset')
    @mock.patch('os.path.isfile')
    @mock.patch('crmsh.report.collect.get_pcmk_log')
    @mock.patch('crmsh.report.collect.get_corosync_log')
    def test_collect_ha_logs(self, mock_corosync_log, mock_get_log, mock_isfile, mock_dump):
        mock_corosync_log.return_value = "/var/log/cluster/corosync.log"
        mock_get_log.return_value = "/var/pacemaker.log"
        mock_isfile.side_effect = [True, True]
        mock_ctx_inst = mock.Mock(extra_log_list=[])

        collect.collect_ha_logs(mock_ctx_inst)

        mock_get_log.assert_called_once_with()
        mock_isfile.assert_has_calls([
            mock.call(mock_get_log.return_value),
            mock.call(mock_corosync_log.return_value)
            ])
        mock_dump.assert_has_calls([
            mock.call(mock_ctx_inst, mock_get_log.return_value),
            mock.call(mock_ctx_inst, mock_corosync_log.return_value)
            ])

    @mock.patch('logging.Logger.warning')
    @mock.patch('os.path.exists')
    @mock.patch('crmsh.corosync.conf')
    def test_get_corosync_log_not_exist(self, mock_conf, mock_exists, mock_warning):
        mock_conf.return_value = "/etc/corosync/corosync.conf"
        mock_exists.return_value = False
        self.assertEqual(collect.get_corosync_log(), "")

    @mock.patch('crmsh.corosync.get_value')
    @mock.patch('os.path.exists')
    @mock.patch('crmsh.corosync.conf')
    def test_get_corosync_log(self, mock_conf, mock_exists, mock_get_value):
        mock_conf.return_value = "/etc/corosync/corosync.conf"
        mock_get_value.return_value = "/var/log/cluster/corosync.log"
        mock_exists.return_value = True
        self.assertEqual(collect.get_corosync_log(), mock_get_value.return_value)

    @mock.patch('crmsh.report.utils.real_path')
    @mock.patch('crmsh.report.collect.logger', spec=crmsh.log.DEBUG2Logger)
    @mock.patch('crmsh.utils.str2file')
    @mock.patch('crmsh.report.utils.get_cmd_output')
    @mock.patch('crmsh.report.utils.ts_to_str')
    def test_collect_journal_logs(self, mock_ts_to_str, mock_get_cmd_output,
                                  mock_str2file, mock_logger, mock_real_path):
        mock_real_path.side_effect = [
                constants.JOURNAL_F,
                constants.JOURNAL_PCMK_F,
                constants.JOURNAL_COROSYNC_F,
                constants.JOURNAL_SBD_F
        ]
        mock_ctx_inst = mock.Mock(from_time=1234, to_time=5678, work_dir="/opt/work")
        mock_ts_to_str.side_effect = ["10.10", "10.12"]
        mock_get_cmd_output.side_effect = ["data_default", "data_pacemaker", "data_corosync", "data_sbd"]
        collect.collect_journal_logs(mock_ctx_inst)
        mock_ts_to_str.assert_has_calls([
            mock.call(mock_ctx_inst.from_time),
            mock.call(mock_ctx_inst.to_time)
            ])
        cmd_list = [
                'journalctl -o short-iso-precise --since "10.10" --until "10.12" --no-pager | tail -n +2',
                'journalctl -u pacemaker -o short-iso-precise --since "10.10" --until "10.12" --no-pager | tail -n +2',
                'journalctl -u corosync -o short-iso-precise --since "10.10" --until "10.12" --no-pager | tail -n +2',
                'journalctl -u sbd -o short-iso-precise --since "10.10" --until "10.12" --no-pager | tail -n +2'
                ]
        mock_get_cmd_output.assert_has_calls([
            mock.call(cmd_list[0]),
            mock.call(cmd_list[1]),
            mock.call(cmd_list[2]),
            mock.call(cmd_list[3]),
            ])
        mock_logger.debug2.assert_has_calls([
            mock.call("Collect journal logs since: 10.10 until: 10.12"),
            mock.call(f"Running command: {cmd_list[0]}"),
            mock.call(f"Running command: {cmd_list[1]}"),
            mock.call(f"Running command: {cmd_list[2]}"),
            mock.call(f"Running command: {cmd_list[3]}"),
            ])
        mock_logger.debug.assert_has_calls([
            mock.call(f"Dump jounal log for default into {constants.JOURNAL_F}"),
            mock.call(f"Dump jounal log for pacemaker into {constants.JOURNAL_PCMK_F}"),
            mock.call(f"Dump jounal log for corosync into {constants.JOURNAL_COROSYNC_F}"),
            mock.call(f"Dump jounal log for sbd into {constants.JOURNAL_SBD_F}")
            ])

    @mock.patch('crmsh.report.collect.ShellUtils')
    def test_dump_D_process_empty(self, mock_run):
        mock_run_inst = mock.Mock()
        mock_run.return_value = mock_run_inst
        mock_run_inst.get_stdout_stderr.return_value = (0, None, None)
        res = collect.dump_D_process()
        self.assertEqual(res, "Dump D-state process stack: 0\n")

    @mock.patch('crmsh.report.collect.ShellUtils')
    def test_dump_D_process(self, mock_run):
        mock_run_inst = mock.Mock()
        mock_run.return_value = mock_run_inst
        mock_run_inst.get_stdout_stderr.side_effect = [
            (0, "1000", None),
            (0, "data1", None),
            (0, "data2", None)
        ]
        res = collect.dump_D_process()
        self.assertEqual(res, "Dump D-state process stack: 1\npid: 1000     comm: data1\ndata2\n\n")
        mock_run_inst.get_stdout_stderr.assert_has_calls([
            mock.call("ps aux|awk '$8 ~ /^D/{print $2}'"),
            mock.call('cat /proc/1000/comm'),
            mock.call('cat /proc/1000/stack')
            ])

    @mock.patch('logging.Logger.debug')
    @mock.patch('os.path.exists')
    def test_collect_sbd_info_no_config(self, mock_exists, mock_debug):
        mock_exists.return_value = False
        mock_ctx_inst = mock.Mock()
        collect.collect_sbd_info(mock_ctx_inst)
        mock_exists.assert_called_once_with(constants.SBDCONF)
        mock_debug.assert_called_once_with(f"SBD config file {constants.SBDCONF} does not exist")

    @mock.patch('shutil.which')
    @mock.patch('shutil.copy2')
    @mock.patch('os.path.exists')
    def test_collect_sbd_info_no_cmd(self, mock_exists, mock_copy, mock_which):
        mock_exists.return_value = True
        mock_which.return_value = False
        mock_ctx_inst = mock.Mock(work_dir="/opt")
        collect.collect_sbd_info(mock_ctx_inst)
        mock_exists.assert_called_once_with(constants.SBDCONF)
        mock_copy.assert_called_once_with(constants.SBDCONF, mock_ctx_inst.work_dir)
        mock_which.assert_called_once_with("sbd")

    @mock.patch('crmsh.report.utils.real_path')
    @mock.patch('builtins.open', create=True)
    @mock.patch('logging.Logger.debug')
    @mock.patch('crmsh.report.utils.get_cmd_output')
    @mock.patch('shutil.which')
    @mock.patch('shutil.copy2')
    @mock.patch('os.path.exists')
    def test_collect_sbd_info(self, mock_exists, mock_copy, mock_which, mock_run, mock_debug, mock_open_file, mock_real_path):
        mock_real_path.return_value = constants.SBD_F
        mock_exists.return_value = True
        mock_which.return_value = True
        mock_open_write = mock.mock_open()
        file_handle = mock_open_write.return_value.__enter__.return_value
        mock_open_file.return_value = mock_open_write.return_value
        mock_run.return_value = "data"
        mock_ctx_inst = mock.Mock(work_dir="/opt")

        collect.collect_sbd_info(mock_ctx_inst)

        mock_exists.assert_called_once_with(constants.SBDCONF)
        mock_copy.assert_called_once_with(constants.SBDCONF, mock_ctx_inst.work_dir)
        mock_which.assert_called_once_with("sbd")
        mock_open_file.assert_called_once_with(f"{mock_ctx_inst.work_dir}/{constants.SBD_F}", "w")
        file_handle.write.assert_has_calls([
            mock.call("\n\n#=====[ Command ] ==========================#\n"),
            mock.call("# . /etc/sysconfig/sbd;export SBD_DEVICE;sbd dump;sbd list\n"),
            mock.call("data")
            ])
        mock_debug.assert_called_once_with(f"Dump SBD config file into {constants.SBD_F}")

    @mock.patch('logging.Logger.warning')
    @mock.patch('crmsh.report.collect.ShellUtils')
    def test_pe_to_dot(self, mock_run, mock_warning):
        mock_run_inst = mock.Mock()
        mock_run.return_value = mock_run_inst
        mock_run_inst.get_stdout_stderr.return_value = (1, None, None)
        collect.pe_to_dot("/opt/pe-input-0.bz2")
        mock_run_inst.get_stdout_stderr.assert_called_once_with("crm_simulate -D /opt/pe-input-0.dot -x /opt/pe-input-0.bz2")
        mock_warning.assert_called_once_with('pe_to_dot: %s -> %s failed', '/opt/pe-input-0.bz2', '/opt/pe-input-0.dot')

    @mock.patch('crmsh.report.utils.find_files_in_timespan')
    @mock.patch('crmsh.report.collect.logger', spec=crmsh.log.DEBUG2Logger)
    def test_collect_pe_inputs_no_found(self, mock_logger, mock_find_files):
        mock_ctx_inst = mock.Mock(pe_dir="/opt/pe_dir")
        mock_find_files.return_value = []
        collect.collect_pe_inputs(mock_ctx_inst)
        mock_find_files.assert_called_once_with(mock_ctx_inst, [mock_ctx_inst.pe_dir])
        mock_logger.debug2.assert_has_calls([
            mock.call(f"Looking for PE files in {mock_ctx_inst.pe_dir}"),
            mock.call("No PE file found for the giving time")
            ])

    @mock.patch('crmsh.report.utils.real_path')
    @mock.patch('crmsh.report.collect.pe_to_dot')
    @mock.patch('os.symlink')
    @mock.patch('crmsh.utils.mkdirp')
    @mock.patch('crmsh.report.utils.find_files_in_timespan')
    @mock.patch('crmsh.report.collect.logger', spec=crmsh.log.DEBUG2Logger)
    def test_collect_pe_inputs(self, mock_logger, mock_find_files, mock_mkdir, mock_symlink, mock_to_dot, mock_real_path):
        mock_real_path.return_value = "pe_dir"
        mock_ctx_inst = mock.Mock(pe_dir="/opt/pe_dir", work_dir="/opt/work_dir", speed_up=False)
        mock_find_files.return_value = ["/opt/pe_dir/pe_input1", "/opt/pe_dir/pe_input2"]

        collect.collect_pe_inputs(mock_ctx_inst)

        mock_find_files.assert_called_once_with(mock_ctx_inst, [mock_ctx_inst.pe_dir])
        mock_logger.debug2.assert_has_calls([
            mock.call(f"Looking for PE files in {mock_ctx_inst.pe_dir}"),
            mock.call(f"Found 2 PE files in {mock_ctx_inst.pe_dir}"),
            ])
        mock_logger.debug.assert_called_once_with(f"Dump PE files into pe_dir")

    @mock.patch('crmsh.report.utils.real_path')
    @mock.patch('crmsh.report.collect.logger', spec=crmsh.log.DEBUG2Logger)
    @mock.patch('crmsh.utils.str2file')
    @mock.patch('crmsh.report.utils.get_cmd_output')
    def test_collect_sys_stats(self, mock_run, mock_str2file, mock_logger, mock_real_path):
        mock_real_path.return_value = constants.SYSSTATS_F
        mock_run.side_effect = [
                "data_hostname", "data_uptime", "data_ps_axf", "data_ps_auxw",
                "data_top", "data_ip_addr", "data_ip_link", "data_ip_show", "data_iscsi",
                "data_lspci", "data_mount", "data_cpuinfo", TimeoutExpired("df", 5)
                ]
        mock_ctx_inst = mock.Mock(work_dir="/opt")
        collect.collect_sys_stats(mock_ctx_inst)
        mock_logger.warning.assert_called_once_with(f"Timeout while running command: df")
        mock_run.assert_has_calls([
            mock.call("hostname", timeout=5),
            mock.call("uptime", timeout=5),
            mock.call("ps axf", timeout=5),
            mock.call("ps auxw", timeout=5),
            mock.call("top -b -n 1", timeout=5),
            mock.call("ip addr", timeout=5),
            mock.call("ip -s link", timeout=5),
            mock.call("ip n show", timeout=5),
            mock.call("lsscsi", timeout=5),
            mock.call("lspci", timeout=5),
            mock.call("mount", timeout=5),
            mock.call("cat /proc/cpuinfo", timeout=5),
            mock.call("df", timeout=5)
            ])

    @mock.patch('crmsh.report.utils.real_path')
    @mock.patch('logging.Logger.debug')
    @mock.patch('crmsh.report.utils.get_distro_info')
    @mock.patch('crmsh.utils.str2file')
    @mock.patch('os.uname')
    @mock.patch('crmsh.report.utils.Package')
    def test_collect_sys_info(self, mock_package, mock_uname, mock_str2file, mock_get_distro, mock_debug, mock_real_path):
        mock_real_path.return_value = constants.SYSINFO_F
        mock_package_inst = mock.Mock()
        mock_package.return_value = mock_package_inst
        mock_package_inst.version = mock.Mock(return_value="version_data\n")
        mock_package_inst.verify = mock.Mock(return_value="verify_data\n")
        mock_ctx_inst = mock.Mock(speed_up=False, work_dir="/opt/work")
        mock_uname.return_value = ("Linux", None, "4.5", None, "x86_64")
        mock_get_distro.return_value = "suse"

        collect.collect_sys_info(mock_ctx_inst)

        mock_package.assert_called_once_with(constants.PACKAGES)
        mock_str2file.assert_called_once_with('##### System info #####\nPlatform: Linux\nKernel release: 4.5\nArchitecture: x86_64\nDistribution: suse\n\n##### Installed cluster related packages #####\nversion_data\n\n\n##### Verification output of packages #####\nverify_data\n', '/opt/work/sysinfo.txt')
        mock_debug.assert_called_once_with(f"Dump packages and platform info into {constants.SYSINFO_F}")

    @mock.patch('crmsh.report.utils.real_path')
    @mock.patch('crmsh.report.collect.dump_configurations')
    @mock.patch('crmsh.report.collect.consume_cib_in_workdir')
    @mock.patch('crmsh.report.collect.logger', spec=crmsh.log.DEBUG2Logger)
    @mock.patch('crmsh.utils.str2file')
    @mock.patch('crmsh.report.collect.dump_runtime_state')
    @mock.patch('crmsh.report.collect.ServiceManager')
    def test_collect_config_running(self, mock_service, mock_dump_state, mock_write, mock_debug2, mock_cib, mock_dump_config, mock_real_path):
        mock_real_path.return_value = "workdir"
        mock_service_inst = mock.Mock()
        mock_service.return_value = mock_service_inst
        mock_service_inst.service_is_active.return_value = True
        mock_ctx_inst = mock.Mock(work_dir="/opt/workdir")
        collect.collect_config(mock_ctx_inst)

    @mock.patch('crmsh.report.utils.real_path')
    @mock.patch('crmsh.report.collect.dump_configurations')
    @mock.patch('crmsh.report.collect.consume_cib_in_workdir')
    @mock.patch('crmsh.report.collect.logger', spec=crmsh.log.DEBUG2Logger)
    @mock.patch('crmsh.utils.str2file')
    @mock.patch('shutil.copy2')
    @mock.patch('crmsh.report.collect.ServiceManager')
    def test_collect_config_stopped(self, mock_service, mock_copy2, mock_write, mock_debug2, mock_cib, mock_dump_config, mock_real_path):
        mock_real_path.return_value = "workdir"
        mock_service_inst = mock.Mock()
        mock_service.return_value = mock_service_inst
        mock_service_inst.service_is_active.return_value = False
        mock_ctx_inst = mock.Mock(work_dir="/opt/workdir", cib_dir="/var/log/pacemaker/cib")
        collect.collect_config(mock_ctx_inst)

    @mock.patch('crmsh.utils.str2file')
    @mock.patch('crmsh.report.collect.sh.cluster_shell')
    @mock.patch('os.path.isfile')
    def test_consume_cib_in_workdir(self, mock_isfile, mock_run, mock_str2file):
        mock_isfile.return_value = True
        mock_run_inst = mock.Mock()
        mock_run.return_value = mock_run_inst
        mock_run_inst.get_stdout_or_raise_error.side_effect = ["data1", "data2"]
        collect.consume_cib_in_workdir("/workdir")
        mock_isfile.assert_called_once_with(f"/workdir/{constants.CIB_F}")
        mock_run_inst.get_stdout_or_raise_error.assert_has_calls([
            mock.call('CIB_file=/workdir/cib.xml crm configure show'),
            mock.call('crm_verify -V -x /workdir/cib.xml')
        ])
        mock_str2file.assert_has_calls([
            mock.call("data1", f"/workdir/{constants.CONFIGURE_SHOW_F}"),
            mock.call("data2", f"/workdir/{constants.CRM_VERIFY_F}")
        ])

    @mock.patch('crmsh.report.collect.logger', spec=crmsh.log.DEBUG2Logger)
    @mock.patch('crmsh.report.collect.sh.cluster_shell')
    def test_collect_ratraces_return(self, mock_run, mock_logger):
        mock_run_inst = mock.Mock()
        mock_run.return_value = mock_run_inst
        mock_run_inst.get_rc_stdout_stderr_without_input.return_value = (0, "data", None)
        mock_ctx_inst = mock.Mock(node_list=["node1"])
        collect.collect_ratraces(mock_ctx_inst)
        mock_logger.debug2.assert_not_called()

    @mock.patch('crmsh.report.utils.real_path')
    @mock.patch('crmsh.report.collect.logger', spec=crmsh.log.DEBUG2Logger)
    @mock.patch('shutil.copy2')
    @mock.patch('crmsh.utils.mkdirp')
    @mock.patch('crmsh.report.utils.find_files_in_timespan')
    @mock.patch('crmsh.report.collect.sh.cluster_shell')
    def test_collect_ratraces(self, mock_run, mock_find, mock_mkdirp, mock_copy, mock_logger, mock_real_path):
        mock_real_path.return_value = "/var/log"
        mock_run_inst = mock.Mock()
        mock_run.return_value = mock_run_inst
        data = "INFO: Trace for .* is written to /var/log/cluster/pacemaker.log"
        mock_run_inst.get_rc_stdout_stderr_without_input.return_value = (0, data, None)
        mock_ctx_inst = mock.Mock(node_list=["node1"], work_dir="/opt/work")
        mock_find.return_value = ["/var/log/cluster"]

        collect.collect_ratraces(mock_ctx_inst)

        mock_logger.debug2.assert_called_once_with('Looking for RA trace files in "%s"', '/var/log/cluster')
        mock_logger.debug.assert_called_once_with(f'Dump RA trace files into {mock_real_path.return_value}')

    @mock.patch('crmsh.report.collect.ShellUtils')
    def test_lsof_ocfs2_device(self, mock_run):
        mock_run_inst = mock.Mock()
        mock_run.return_value = mock_run_inst
        mount_data = """
/dev/vda3 on /home type xfs (rw,relatime,attr2,inode64,logbufs=8,logbsize=32k,noquota)
tmpfs on /run/user/0 type tmpfs (rw,nosuid,nodev,relatime,size=169544k,nr_inodes=42386,mode=700,inode64)
/dev/sda7 on /srv/clusterfs type ocfs2 (rw,relatime,heartbeat=non
        """
        mock_run_inst.get_stdout_stderr.side_effect = [(0, mount_data, None), (0, "data", None)]
        res = collect.lsof_ocfs2_device()
        self.assertEqual(res, "\n\n#=====[ Command ] ==========================#\n# lsof /dev/sda7\ndata")
        mock_run_inst.get_stdout_stderr.assert_has_calls([
            mock.call("mount"),
            mock.call("lsof /dev/sda7")
        ])

    @mock.patch('crmsh.report.utils.get_cmd_output')
    @mock.patch('os.path.exists')
    @mock.patch('shutil.which')
    def test_ocfs2_commands_output(self, mock_which, mock_exists, mock_run):
        mock_which.side_effect = [False for i in range(5)] + [True, True]
        mock_exists.return_value = False
        mock_run.return_value = "data"
        res = collect.ocfs2_commands_output()
        self.assertEqual(res, "\n\n#===== [ Command ] ==========================#\n# mount\ndata")

    @mock.patch('crmsh.report.collect.logger', spec=crmsh.log.DEBUG2Logger)
    @mock.patch('crmsh.utils.str2file')
    @mock.patch('crmsh.report.collect.ShellUtils')
    def test_collect_ocfs2_info_error(self, mock_run, mock_str2file, mock_debug2):
        mock_run_inst = mock.Mock()
        mock_run.return_value = mock_run_inst
        mock_run_inst.get_stdout_stderr.return_value = (1, None, "error")
        mock_ctx_inst = mock.Mock(work_dir="/opt/workdir")
        collect.collect_ocfs2_info(mock_ctx_inst)
        mock_str2file.assert_called_once_with('Failed to run "mounted.ocfs2 -d": error', '/opt/workdir/ocfs2.txt')

    @mock.patch('crmsh.report.collect.logger', spec=crmsh.log.DEBUG2Logger)
    @mock.patch('crmsh.utils.str2file')
    @mock.patch('crmsh.report.collect.ShellUtils')
    def test_collect_ocfs2_info_no_found(self, mock_run, mock_str2file, mock_debug2):
        mock_run_inst = mock.Mock()
        mock_run.return_value = mock_run_inst
        mock_run_inst.get_stdout_stderr.return_value = (0, "data", None)
        mock_ctx_inst = mock.Mock(work_dir="/opt/workdir")
        collect.collect_ocfs2_info(mock_ctx_inst)
        mock_str2file.assert_called_once_with('No ocfs2 partitions found', '/opt/workdir/ocfs2.txt')

    @mock.patch('crmsh.report.utils.real_path')
    @mock.patch('crmsh.report.collect.ocfs2_commands_output')
    @mock.patch('crmsh.report.collect.lsof_ocfs2_device')
    @mock.patch('crmsh.report.collect.dump_D_process')
    @mock.patch('crmsh.report.collect.logger', spec=crmsh.log.DEBUG2Logger)
    @mock.patch('crmsh.utils.str2file')
    @mock.patch('crmsh.report.collect.ShellUtils')
    def test_collect_ocfs2_info(self, mock_run, mock_str2file, mock_debug2, mock_D, mock_lsof, mock_output, mock_real_path):
        mock_real_path.return_value = constants.OCFS2_F
        mock_run_inst = mock.Mock()
        mock_run.return_value = mock_run_inst
        mock_run_inst.get_stdout_stderr.return_value = (0, "line1\nline2", None)
        mock_D.return_value = "data_D\n"
        mock_lsof.return_value = "data_lsof\n"
        mock_output.return_value = "data_output\n"
        mock_ctx_inst = mock.Mock(work_dir="/opt/workdir")
        collect.collect_ocfs2_info(mock_ctx_inst)
        mock_str2file.assert_called_once_with('data_D\ndata_lsof\ndata_output\n', '/opt/workdir/ocfs2.txt')

    @mock.patch('crmsh.report.utils.real_path')
    @mock.patch('logging.Logger.debug')
    @mock.patch('crmsh.utils.str2file')
    @mock.patch('crmsh.report.utils.get_cmd_output')
    @mock.patch('shutil.which')
    def test_collect_dlm_info(self, mock_which, mock_get_output, mock_str2file, mock_debug, mock_real_path):
        mock_real_path.return_value = constants.DLM_DUMP_F
        mock_which.return_value = True
        ls_data = """
dlm lockspaces
name          08BB5A6A38EE491DBF63627EEB57E558
id            0x19041a12
        """
        mock_get_output.side_effect = [ls_data, "lockdebug data", "dump data"]
        mock_ctx_inst = mock.Mock(work_dir="/opt/work_dir")
        collect.collect_dlm_info(mock_ctx_inst)
        mock_debug.assert_called_once_with(f"Dump DLM information into {constants.DLM_DUMP_F}")

    @mock.patch('crmsh.report.collect.dump_core_info')
    @mock.patch('logging.Logger.warning')
    @mock.patch('os.path.basename')
    @mock.patch('crmsh.report.utils.find_files_in_timespan')
    def test_collect_coredump_info(self, mock_find, mock_basename, mock_warning, mock_dump):
        mock_ctx_inst = mock.Mock(cores_dir_list=['/var/lib/pacemaker/cores'], work_dir="/opt/work_dir")
        mock_find.return_value = ["/var/lib/pacemaker/cores/core.1"]
        mock_basename.return_value = "core.1"
        collect.collect_coredump_info(mock_ctx_inst)
        mock_dump.assert_called_once_with("/opt/work_dir", mock_find.return_value)
        mock_warning.assert_called_once_with(f"Found coredump file: {mock_find.return_value}")

    @mock.patch('crmsh.report.collect.ShellUtils')
    def test_find_binary_path_for_core_not_found(self, mock_run):
        mock_run().get_stdout_stderr.return_value = (0, "Core not found", None)
        res = collect.find_binary_path_for_core("core.1")
        self.assertEqual("Cannot find the program path for core core.1", res)

    @mock.patch('crmsh.report.collect.ShellUtils')
    def test_find_binary_path_for_core(self, mock_run):
        mock_run_inst = mock.Mock()
        mock_run.return_value = mock_run_inst
        mock_run_inst.get_stdout_stderr.return_value = (0, "Core was generated by `/usr/sbin/crm_mon'", None)
        res = collect.find_binary_path_for_core("core.1")
        self.assertEqual("Core core.1 was generated by /usr/sbin/crm_mon", res)

    @mock.patch('crmsh.report.utils.real_path')
    @mock.patch('crmsh.report.collect.logger', spec=crmsh.log.DEBUG2Logger)
    @mock.patch('crmsh.utils.str2file')
    @mock.patch('shutil.which')
    def test_dump_core_info_no_gdb(self, mock_which, mock_str2file, mock_logger, mock_real_path):
        mock_real_path.return_value = constants.COREDUMP_F
        mock_which.return_value = False
        collect.dump_core_info("/opt/workdir", ["core.1"])
        mock_logger.warning.assert_called_once_with("Please install gdb to get more info for coredump files")

    @mock.patch('crmsh.report.utils.real_path')
    @mock.patch('crmsh.report.collect.logger', spec=crmsh.log.DEBUG2Logger)
    @mock.patch('crmsh.utils.str2file')
    @mock.patch('crmsh.report.collect.find_binary_path_for_core')
    @mock.patch('shutil.which')
    def test_dump_core_info(self, mock_which, mock_find_binary, mock_str2file, mock_debug2, mock_real_path):
        mock_real_path.return_value = constants.COREDUMP_F
        mock_which.return_value = True
        mock_find_binary.return_value = "data"
        collect.dump_core_info("/opt/workdir", ["core.1"])
        mock_str2file.assert_called_once_with("data\n\nPlease utilize the gdb and debuginfo packages to obtain more detailed information locally", f"/opt/workdir/{constants.COREDUMP_F}")
        mock_debug2(f"Dump coredump info into {constants.COREDUMP_F}")

    @mock.patch('crmsh.utils.str2file')
    @mock.patch('pwd.getpwnam')
    @mock.patch('os.stat')
    @mock.patch('os.path.isdir')
    def test_collect_perms_state(self, mock_isdir, mock_stat, mock_getpwnam, mock_str2file):
        mock_ctx_inst = mock.Mock(
            pcmk_lib_dir="/var/lib/pacemaker",
            pe_dir="/var/lib/pacemaker/pe",
            cib_dir="/var/lib/pacemaker/cib",
            work_dir="/opt/work_dir"
        )
        mock_isdir.side_effect = [False, True, True]
        mock_stat_inst_pe = mock.Mock(st_uid=1000, st_gid=1000, st_mode=0o750)
        mock_stat_inst_cib = mock.Mock(st_uid=1000, st_gid=1000, st_mode=0o750)
        mock_stat.side_effect = [mock_stat_inst_pe, mock_stat_inst_cib]
        mock_getpwnam_inst_pe = mock.Mock(pw_uid=1000, pw_gid=1000)
        mock_getpwnam_inst_cib = mock.Mock(pw_uid=1001, pw_gid=1000)
        mock_getpwnam.side_effect = [mock_getpwnam_inst_pe, mock_getpwnam_inst_cib]

        collect.collect_perms_state(mock_ctx_inst)

        data = "##### Check perms for /var/lib/pacemaker: /var/lib/pacemaker is not a directory or does not exist\n##### Check perms for /var/lib/pacemaker/pe: OK\n##### Check perms for /var/lib/pacemaker/cib: Permissions or ownership for /var/lib/pacemaker/cib are incorrect\n"
        mock_str2file.assert_called_once_with(data, f"/opt/work_dir/{constants.PERMISSIONS_F}")

    @mock.patch('crmsh.report.utils.real_path')
    @mock.patch('crmsh.utils.this_node')
    @mock.patch('crmsh.utils.get_dc')
    @mock.patch('logging.Logger.debug')
    @mock.patch('crmsh.utils.str2file')
    @mock.patch('crmsh.report.collect.sh.cluster_shell')
    def test_dump_runtime_state(self, mock_run, mock_str2file, mock_debug, mock_get_dc, mock_this_node, mock_real_path):
        mock_real_path.side_effect = [
            constants.CRM_MON_F,
            constants.CIB_F,
            constants.MEMBERSHIP_F,
            "workdir"
        ]
        mock_run_inst = mock.Mock()
        mock_run.return_value = mock_run_inst
        mock_run_inst.get_stdout_or_raise_error.side_effect = ["crm_mon_data", "cib_data", "crm_node_data"]
        mock_get_dc.return_value = "node1"
        mock_this_node.return_value = "node1"
        collect.dump_runtime_state("/opt/workdir")
        mock_debug.assert_has_calls([
            mock.call(f"Dump cluster state into {constants.CRM_MON_F}"),
            mock.call(f"Dump CIB contents into {constants.CIB_F}"),
            mock.call(f"Dump members of this partition into {constants.MEMBERSHIP_F}"),
            mock.call(f"Current DC is node1; Touch file 'DC' in workdir")
        ])

    @mock.patch('shutil.copytree')
    @mock.patch('os.path.basename')
    @mock.patch('os.path.isdir')
    @mock.patch('shutil.copy2')
    @mock.patch('os.path.isfile')
    @mock.patch('crmsh.corosync.conf')
    def test_dump_configurations(self, mock_corosync_conf, mock_isfile, mock_copy2, mock_isdir, mock_basename, mock_copytree):
        mock_corosync_conf.return_value = "/etc/corosync/corosync.conf"
        mock_isfile.side_effect = [True, True, False, True]
        mock_isdir.return_value = True
        mock_basename.return_value = "drbd.d"
        collect.dump_configurations("/opt/workdir")

    @mock.patch('crmsh.report.utils.real_path')
    @mock.patch('logging.Logger.debug')
    @mock.patch('crmsh.utils.str2file')
    @mock.patch('crmsh.report.utils.get_cmd_output')
    @mock.patch('crmsh.report.utils.find_files_in_timespan')
    def test_collect_corosync_blackbox(self, mock_find_files, mock_get_cmd_output, mock_str2file, mock_debug, mock_real_path):
        mock_real_path.return_value = constants.COROSYNC_RECORDER_F
        mock_ctx_inst = mock.Mock(work_dir="/opt/workdir")
        mock_find_files.return_value = ["/var/lib/corosync/fdata.1"]
        mock_get_cmd_output.return_value = "data"
        collect.collect_corosync_blackbox(mock_ctx_inst)
        mock_debug.assert_called_once_with(f"Dump corosync blackbox info into {constants.COROSYNC_RECORDER_F}")