How to use the ffpuppet.minidump_parser.MinidumpParser function in ffpuppet

def test_minidump_parser_01(mocker, tmp_path):
    """test MinidumpParser() with missing and empty scan path"""
    with pytest.raises(IOError):
        MinidumpParser("/path/does/not/exist/")
    mdp = MinidumpParser(str(tmp_path))
    assert not mdp.dump_files
    callback = mocker.Mock()
    with pytest.raises(IOError):
        mdp.collect_logs(callback, "/path/does/not/exist/")
    assert callback.call_count == 0
    mdp.collect_logs(callback, str(tmp_path))
    assert callback.call_count == 0

out_fp.write(b"0|0|blah|foo|a/bar.c|123|0x0\n")
        out_fp.write(b"0|1|blat|foo|a/bar.c|223|0x0\n")
        out_fp.write(b"junk\n")
        out_fp.write(b"0|2|blas|foo|a/bar.c|423|0x0\n")
        out_fp.write(b"0|3|blas|foo|a/bar.c|423|0x0\n")
        out_fp.write(b"1|0|libpthread-2.23.so||||0xd360\n")
        out_fp.write(b"junk\n")
        out_fp.write(b"1|1|swrast_dri.so||||0x7237f3\n")
        out_fp.write(b"1|2|libplds4.so|_fini|||0x163\n")
        out_fp.write(b"2|0|swrast_dri.so||||0x723657\n")
        out_fp.write(b"junk\n")
        out_fp.write(b"2|1|libpthread-2.23.so||||0x76ba\n")
        out_fp.write(b"2|3|libc-2.23.so||||0x1073dd\n\n")
        out_fp.seek(0)
    mocker.patch.object(MinidumpParser, '_call_mdsw', side_effect=fake_call_mdsw)
    mdp = MinidumpParser(str(tmp_path))
    MinidumpParser.MDSW_MAX_STACK = 7
    with tempfile.TemporaryFile() as log_fp:
        mdp._read_stacktrace("fake.dmp", log_fp)
        log_fp.seek(0)
        md_lines = log_fp.readlines()
    assert len(md_lines) == 8  # only the interesting stack info should be in here
    assert md_lines[-1].startswith(b"WARNING: Hit line output limit!")
    assert md_lines[-2].startswith(b"0|2|")
    # test raw_fp set
    MinidumpParser.MDSW_MAX_STACK = 150
    with tempfile.TemporaryFile() as log_fp, tempfile.TemporaryFile() as raw_fp:
        mdp._read_stacktrace("fake.dmp", log_fp, raw_fp=raw_fp)
        raw_size = raw_fp.tell()
        log_fp.seek(0)
        md_lines = log_fp.readlines()
    with tempfile.TemporaryFile() as log_fp:

import threading
import time

from psutil import AccessDenied, NoSuchProcess, Process, wait_procs
import pytest

from .core import FFPuppet
from .exceptions import BrowserTimeoutError, BrowserTerminatedError, LaunchError
from .helpers import get_processes
from .minidump_parser import MinidumpParser

CWD = os.path.realpath(os.path.dirname(__file__))
TESTFF_BIN = os.path.join(CWD, "resources", "testff.py")
TESTMDSW_BIN = os.path.join(CWD, "resources", "testmdsw.py")

MinidumpParser.MDSW_BIN = TESTMDSW_BIN
MinidumpParser.MDSW_MAX_STACK = 8


class ReqHandler(BaseHTTPRequestHandler):
    def do_GET(self):
        self.send_response(200)
        self.end_headers()
        self.wfile.write(b"hello world")


class HTTPTestServer(object):
    def __init__(self, handler=None):
        self._handler = handler if handler is not None else ReqHandler
        while True:
            try:
                self._httpd = HTTPServer(("127.0.0.1", 0), self._handler)

def test_minidump_parser_01(mocker, tmp_path):
    """test MinidumpParser() with missing and empty scan path"""
    with pytest.raises(IOError):
        MinidumpParser("/path/does/not/exist/")
    mdp = MinidumpParser(str(tmp_path))
    assert not mdp.dump_files
    callback = mocker.Mock()
    with pytest.raises(IOError):
        mdp.collect_logs(callback, "/path/does/not/exist/")
    assert callback.call_count == 0
    mdp.collect_logs(callback, str(tmp_path))
    assert callback.call_count == 0

def test_minidump_parser_02(mocker, tmp_path):
    """test MinidumpParser() with empty minidumps (ignore mdsw failures)"""
    md_path = tmp_path / "minidumps"
    md_path.mkdir()
    (md_path / "not_a_dmp.txt").touch()
    (md_path / "test.dmp").touch()
    callback = mocker.mock_open()
    callback.return_value.tell.return_value = 0
    log_path = tmp_path / "logs"
    log_path.mkdir()
    mdp = MinidumpParser(str(md_path), record_failures=False)
    assert len(mdp.dump_files) == 1
    mdp.collect_logs(callback, str(log_path))
    assert callback.return_value.tell.call_count == 1
    assert callback.return_value.write.call_count == 1

def test_minidump_parser_05(mocker, tmp_path):
    """test MinidumpParser.collect_logs()"""
    (tmp_path / "dummy.dmp").touch()
    (tmp_path / "dummy.txt").touch()
    with (tmp_path / "test.dmp").open("wb") as out_fp:
        out_fp.write(b"Crash reason:  SIGSEGV\n")
        out_fp.write(b"Crash address: 0x0\n")
        out_fp.write(b"Thread 0 (crashed)\n")
        out_fp.write(b" 0  libxul.so + 0x123456788\n")
        out_fp.write(b"    rax = 0xe5423423423fffe8   rdx = 0x0000000000000000\n")
        out_fp.write(b"OS|Linux|0.0.0 sys info...\n")
        out_fp.write(b"Crash|SIGSEGV|0x7fff27aaeff8|0\n")
        out_fp.write(b"0|0|blah|foo|a/bar.c|123|0x0\n")
    fake_subproc = mocker.patch("ffpuppet.minidump_parser.subprocess", autospec=True)
    fake_subproc.call.return_value = 0
    mdp = MinidumpParser(str(tmp_path))
    callback = mocker.mock_open()
    callback.return_value.tell.return_value = 0
    mdp.collect_logs(callback, str(tmp_path))
    assert callback.call_count == 2

import time

from psutil import AccessDenied, NoSuchProcess, Process, wait_procs
import pytest

from .core import FFPuppet
from .exceptions import BrowserTimeoutError, BrowserTerminatedError, LaunchError
from .helpers import get_processes
from .minidump_parser import MinidumpParser

CWD = os.path.realpath(os.path.dirname(__file__))
TESTFF_BIN = os.path.join(CWD, "resources", "testff.py")
TESTMDSW_BIN = os.path.join(CWD, "resources", "testmdsw.py")

MinidumpParser.MDSW_BIN = TESTMDSW_BIN
MinidumpParser.MDSW_MAX_STACK = 8


class ReqHandler(BaseHTTPRequestHandler):
    def do_GET(self):
        self.send_response(200)
        self.end_headers()
        self.wfile.write(b"hello world")


class HTTPTestServer(object):
    def __init__(self, handler=None):
        self._handler = handler if handler is not None else ReqHandler
        while True:
            try:
                self._httpd = HTTPServer(("127.0.0.1", 0), self._handler)
            except socket.error as soc_e:

out_fp.write(b"Thread 0 (crashed)\n")
        out_fp.write(b" 0  libxul.so + 0x123456788\n")
        out_fp.write(b"    rax = 0xe5423423423fffe8   rdx = 0x0000000000000000\n")
        out_fp.write(b"    rcx = 0x0000000000000000   rbx = 0xe54234234233e5e5\n")
        out_fp.write(b"    rsi = 0x0000000000000000   rdi = 0x00007fedc31fe308\n")
        out_fp.write(b"    rbp = 0x00007fffca0dab00   rsp = 0x00007fffca0daad0\n")
        out_fp.write(b"     r8 = 0x0000000000000000    r9 = 0x0000000000000008\n")
        out_fp.write(b"    r10 = 0xffff00ffffffffff   r11 = 0xffffff00ffffffff\n")
        out_fp.write(b"    r12 = 0x0000743564566308   r13 = 0x00007fedce9d8000\n")
        out_fp.write(b"    r14 = 0x0000000000000001   r15 = 0x0000000000000000\n")
        out_fp.write(b"    rip = 0x0000745666666ac\n")
        out_fp.write(b"    Found by: given as instruction pointer in context\n")
        out_fp.write(b" 1  libxul.so + 0x1f4361c]\n\n")
        out_fp.seek(0)
    mocker.patch.object(MinidumpParser, '_call_mdsw', side_effect=fake_call_mdsw)
    mdp = MinidumpParser(str(tmp_path))
    md_lines = list()
    with tempfile.TemporaryFile() as log_fp:
        mdp._read_registers("fake.dmp", log_fp)
        log_fp.seek(0)
        for line in log_fp:  # pylint: disable=not-an-iterable
            if b"=" not in line:
                break
            md_lines.append(line)
    assert len(md_lines) == 9   # only register info should be in here

    @type cb_create_log: callback
    @param cb_create_log: A callback to the add_log() of a PuppetLogger

    @rtype: None
    @return: None
    """
    assert isinstance(scan_path, str)
    assert isinstance(symbols_path, str)
    assert callable(cb_create_log)

    if not os.path.isdir(scan_path):
        log.debug("scan_path %r does not exist", scan_path)
        return

    md_parser = MinidumpParser(scan_path)
    if not md_parser.dump_files:
        log.debug("scan_path %r did not contain '.dmp' files", scan_path)
        return

    if not os.path.isdir(symbols_path):
        log.warning("symbols_path not found: %r", symbols_path)
        return

    if not md_parser.mdsw_available():
        log.warning("Found a minidump, but can't process it without minidump_stackwalk."
                    " See README.md for how to obtain it.")
        return

    md_parser.collect_logs(cb_create_log, symbols_path)