DXR is a code search and navigation tool aimed at making sense of large projects. It supports full-text and regex searches as well as structural queries.

Mercurial (5b81998bb7ab)

VCS Links

Line Code
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172
#!/usr/bin/env python
#
# This Source Code Form is subject to the terms of the Mozilla Public
# License, v. 2.0. If a copy of the MPL was not distributed with this file,
# You can obtain one at http://mozilla.org/MPL/2.0/.

import os, unittest, subprocess, tempfile, shutil, urlparse, zipfile, StringIO
import mozcrash, mozlog, mozhttpd

# Make logs go away
log = mozlog.getLogger("mozcrash", os.devnull)

def popen_factory(stdouts):
    """
    Generate a class that can mock subprocess.Popen. |stdouts| is an iterable that
    should return an iterable for the stdout of each process in turn.
    """
    class mock_popen(object):
        def __init__(self, args, *args_rest, **kwargs):
            self.stdout = stdouts.next()
            self.returncode = 0

        def wait(self):
            return 0

        def communicate(self):
            return (self.stdout.next(), "")

    return mock_popen

class TestCrash(unittest.TestCase):
    def setUp(self):
        self.tempdir = tempfile.mkdtemp()
        # a fake file to use as a stackwalk binary
        self.stackwalk = os.path.join(self.tempdir, "stackwalk")
        open(self.stackwalk, "w").write("fake binary")
        self._subprocess_popen = subprocess.Popen
        subprocess.Popen = popen_factory(self.next_mock_stdout())
        self.stdouts = []

    def tearDown(self):
        subprocess.Popen = self._subprocess_popen
        shutil.rmtree(self.tempdir)

    def next_mock_stdout(self):
        if not self.stdouts:
            yield iter([])
        for s in self.stdouts:
            yield iter(s)

    def test_nodumps(self):
        """
        Test that check_for_crashes returns False if no dumps are present.
        """
        self.stdouts.append(["this is some output"])
        self.assertFalse(mozcrash.check_for_crashes(self.tempdir,
                                                    'symbols_path',
                                                    stackwalk_binary=self.stackwalk,
                                                    quiet=True))

    def test_simple(self):
        """
        Test that check_for_crashes returns True if a dump is present.
        """
        open(os.path.join(self.tempdir, "test.dmp"), "w").write("foo")
        self.stdouts.append(["this is some output"])
        self.assert_(mozcrash.check_for_crashes(self.tempdir,
                                                'symbols_path',
                                                stackwalk_binary=self.stackwalk,
                                                quiet=True))

    def test_stackwalk_envvar(self):
        """
        Test that check_for_crashes uses the MINIDUMP_STACKWALK environment var.
        """
        open(os.path.join(self.tempdir, "test.dmp"), "w").write("foo")
        self.stdouts.append(["this is some output"])
        os.environ['MINIDUMP_STACKWALK'] = self.stackwalk
        self.assert_(mozcrash.check_for_crashes(self.tempdir,
                                                'symbols_path',
                                                quiet=True))
        del os.environ['MINIDUMP_STACKWALK']

    def test_save_path(self):
        """
        Test that dump_save_path works.
        """
        open(os.path.join(self.tempdir, "test.dmp"), "w").write("foo")
        save_path = os.path.join(self.tempdir, "saved")
        os.mkdir(save_path)
        self.stdouts.append(["this is some output"])
        self.assert_(mozcrash.check_for_crashes(self.tempdir,
                                                'symbols_path',
                                                stackwalk_binary=self.stackwalk,
                                                dump_save_path=save_path,
                                                quiet=True))
        self.assert_(os.path.isfile(os.path.join(save_path, "test.dmp")))

    def test_save_path_not_present(self):
        """
        Test that dump_save_path works when the directory doesn't exist.
        """
        open(os.path.join(self.tempdir, "test.dmp"), "w").write("foo")
        save_path = os.path.join(self.tempdir, "saved")
        self.stdouts.append(["this is some output"])
        self.assert_(mozcrash.check_for_crashes(self.tempdir,
                                                'symbols_path',
                                                stackwalk_binary=self.stackwalk,
                                                dump_save_path=save_path,
                                                quiet=True))
        self.assert_(os.path.isfile(os.path.join(save_path, "test.dmp")))

    def test_save_path_isfile(self):
        """
        Test that dump_save_path works when the directory doesn't exist,
        but a file with the same name exists.
        """
        open(os.path.join(self.tempdir, "test.dmp"), "w").write("foo")
        save_path = os.path.join(self.tempdir, "saved")
        open(save_path, "w").write("junk")
        self.stdouts.append(["this is some output"])
        self.assert_(mozcrash.check_for_crashes(self.tempdir,
                                                'symbols_path',
                                                stackwalk_binary=self.stackwalk,
                                                dump_save_path=save_path,
                                                quiet=True))
        self.assert_(os.path.isfile(os.path.join(save_path, "test.dmp")))

    def test_save_path_envvar(self):
        """
        Test that the MINDUMP_SAVE_PATH environment variable works.
        """
        open(os.path.join(self.tempdir, "test.dmp"), "w").write("foo")
        save_path = os.path.join(self.tempdir, "saved")
        os.mkdir(save_path)
        self.stdouts.append(["this is some output"])
        os.environ['MINIDUMP_SAVE_PATH'] = save_path
        self.assert_(mozcrash.check_for_crashes(self.tempdir,
                                                'symbols_path',
                                                stackwalk_binary=self.stackwalk,
                                                quiet=True))
        del os.environ['MINIDUMP_SAVE_PATH']
        self.assert_(os.path.isfile(os.path.join(save_path, "test.dmp")))

    def test_symbol_path_url(self):
        """
        Test that passing a URL as symbols_path correctly fetches the URL.
        """
        open(os.path.join(self.tempdir, "test.dmp"), "w").write("foo")
        self.stdouts.append(["this is some output"])

        def make_zipfile():
            data = StringIO.StringIO()
            z = zipfile.ZipFile(data, 'w')
            z.writestr("symbols.txt", "abc/xyz")
            z.close()
            return data.getvalue()
        def get_symbols(req):
            headers = {}
            return (200, headers, make_zipfile())
        httpd = mozhttpd.MozHttpd(port=0,
                                  urlhandlers=[{'method':'GET', 'path':'/symbols', 'function':get_symbols}])
        httpd.start()
        symbol_url = urlparse.urlunsplit(('http', '%s:%d' % httpd.httpd.server_address,
                                        '/symbols','',''))
        self.assert_(mozcrash.check_for_crashes(self.tempdir,
                                                symbol_url,
                                                stackwalk_binary=self.stackwalk,
                                                quiet=True))

if __name__ == '__main__':
    unittest.main()