michael@0: #!/usr/bin/env python
michael@0: # This Source Code Form is subject to the terms of the Mozilla Public
michael@0: # License, v. 2.0. If a copy of the MPL was not distributed with this
michael@0: # file, You can obtain one at http://mozilla.org/MPL/2.0/.
michael@0: 
michael@0: import os, tempfile, unittest, shutil, struct, platform, subprocess, multiprocessing.dummy
michael@0: import mock
michael@0: from mock import patch
michael@0: import symbolstore
michael@0: 
michael@0: # Some simple functions to mock out files that the platform-specific dumpers will accept.
michael@0: # dump_syms itself will not be run (we mock that call out), but we can't override
michael@0: # the ShouldProcessFile method since we actually want to test that.
michael@0: def write_elf(filename):
michael@0:     open(filename, "wb").write(struct.pack("<7B45x", 0x7f, ord("E"), ord("L"), ord("F"), 1, 1, 1))
michael@0: 
michael@0: def write_macho(filename):
michael@0:     open(filename, "wb").write(struct.pack("<I28x", 0xfeedface))
michael@0: 
michael@0: def write_pdb(filename):
michael@0:     open(filename, "w").write("aaa")
michael@0:     # write out a fake DLL too
michael@0:     open(os.path.splitext(filename)[0] + ".dll", "w").write("aaa")
michael@0: 
michael@0: writer = {'Windows': write_pdb,
michael@0:           'Microsoft': write_pdb,
michael@0:           'Linux': write_elf,
michael@0:           'Sunos5': write_elf,
michael@0:           'Darwin': write_macho}[platform.system()]
michael@0: extension = {'Windows': ".pdb",
michael@0:              'Microsoft': ".pdb",
michael@0:              'Linux': ".so",
michael@0:              'Sunos5': ".so",
michael@0:              'Darwin': ".dylib"}[platform.system()]
michael@0: 
michael@0: def add_extension(files):
michael@0:     return [f + extension for f in files]
michael@0: 
michael@0: class HelperMixin(object):
michael@0:     """
michael@0:     Test that passing filenames to exclude from processing works.
michael@0:     """
michael@0:     def setUp(self):
michael@0:         self.test_dir = tempfile.mkdtemp()
michael@0:         if not self.test_dir.endswith("/"):
michael@0:             self.test_dir += "/"
michael@0:         symbolstore.srcdirRepoInfo = {}
michael@0:         symbolstore.vcsFileInfoCache = {}
michael@0: 
michael@0:     def tearDown(self):
michael@0:         shutil.rmtree(self.test_dir)
michael@0:         symbolstore.srcdirRepoInfo = {}
michael@0:         symbolstore.vcsFileInfoCache = {}
michael@0: 
michael@0:     def add_test_files(self, files):
michael@0:         for f in files:
michael@0:             f = os.path.join(self.test_dir, f)
michael@0:             d = os.path.dirname(f)
michael@0:             if d and not os.path.exists(d):
michael@0:                 os.makedirs(d)
michael@0:             writer(f)
michael@0: 
michael@0: class TestExclude(HelperMixin, unittest.TestCase):
michael@0:     def test_exclude_wildcard(self):
michael@0:         """
michael@0:         Test that using an exclude list with a wildcard pattern works.
michael@0:         """
michael@0:         processed = []
michael@0:         def mock_process_file(filenames):
michael@0:             for filename in filenames:
michael@0:                 processed.append((filename[len(self.test_dir):] if filename.startswith(self.test_dir) else filename).replace('\\', '/'))
michael@0:             return True
michael@0:         self.add_test_files(add_extension(["foo", "bar", "abc/xyz", "abc/fooxyz", "def/asdf", "def/xyzfoo"]))
michael@0:         d = symbolstore.GetPlatformSpecificDumper(dump_syms="dump_syms",
michael@0:                                                   symbol_path="symbol_path",
michael@0:                                                   exclude=["*foo*"])
michael@0:         d.ProcessFiles = mock_process_file
michael@0:         d.Process(self.test_dir)
michael@0:         d.Finish(stop_pool=False)
michael@0:         processed.sort()
michael@0:         expected = add_extension(["bar", "abc/xyz", "def/asdf"])
michael@0:         expected.sort()
michael@0:         self.assertEqual(processed, expected)
michael@0: 
michael@0:     def test_exclude_filenames(self):
michael@0:         """
michael@0:         Test that excluding a filename without a wildcard works.
michael@0:         """
michael@0:         processed = []
michael@0:         def mock_process_file(filenames):
michael@0:             for filename in filenames:
michael@0:                 processed.append((filename[len(self.test_dir):] if filename.startswith(self.test_dir) else filename).replace('\\', '/'))
michael@0:             return True
michael@0:         self.add_test_files(add_extension(["foo", "bar", "abc/foo", "abc/bar", "def/foo", "def/bar"]))
michael@0:         d = symbolstore.GetPlatformSpecificDumper(dump_syms="dump_syms",
michael@0:                                                   symbol_path="symbol_path",
michael@0:                                                   exclude=add_extension(["foo"]))
michael@0:         d.ProcessFiles = mock_process_file
michael@0:         d.Process(self.test_dir)
michael@0:         d.Finish(stop_pool=False)
michael@0:         processed.sort()
michael@0:         expected = add_extension(["bar", "abc/bar", "def/bar"])
michael@0:         expected.sort()
michael@0:         self.assertEqual(processed, expected)
michael@0: 
michael@0: def popen_factory(stdouts):
michael@0:     """
michael@0:     Generate a class that can mock subprocess.Popen. |stdouts| is an iterable that
michael@0:     should return an iterable for the stdout of each process in turn.
michael@0:     """
michael@0:     class mock_popen(object):
michael@0:         def __init__(self, args, *args_rest, **kwargs):
michael@0:             self.stdout = stdouts.next()
michael@0: 
michael@0:         def wait(self):
michael@0:             return 0
michael@0:     return mock_popen
michael@0: 
michael@0: def mock_dump_syms(module_id, filename):
michael@0:     return ["MODULE os x86 %s %s" % (module_id, filename),
michael@0:             "FILE 0 foo.c",
michael@0:             "PUBLIC xyz 123"]
michael@0: 
michael@0: class TestCopyDebugUniversal(HelperMixin, unittest.TestCase):
michael@0:     """
michael@0:     Test that CopyDebug does the right thing when dumping multiple architectures.
michael@0:     """
michael@0:     def setUp(self):
michael@0:         HelperMixin.setUp(self)
michael@0:         self.symbol_dir = tempfile.mkdtemp()
michael@0:         self._subprocess_call = subprocess.call
michael@0:         subprocess.call = self.mock_call
michael@0:         self._subprocess_popen = subprocess.Popen
michael@0:         subprocess.Popen = popen_factory(self.next_mock_stdout())
michael@0:         self.stdouts = []
michael@0:         self._shutil_rmtree = shutil.rmtree
michael@0:         shutil.rmtree = self.mock_rmtree
michael@0: 
michael@0:     def tearDown(self):
michael@0:         HelperMixin.tearDown(self)
michael@0:         shutil.rmtree = self._shutil_rmtree
michael@0:         shutil.rmtree(self.symbol_dir)
michael@0:         subprocess.call = self._subprocess_call
michael@0:         subprocess.Popen = self._subprocess_popen
michael@0: 
michael@0:     def mock_rmtree(self, path):
michael@0:         pass
michael@0: 
michael@0:     def mock_call(self, args, **kwargs):
michael@0:         if args[0].endswith("dsymutil"):
michael@0:             filename = args[-1]
michael@0:             os.makedirs(filename + ".dSYM")
michael@0:         return 0
michael@0: 
michael@0:     def next_mock_stdout(self):
michael@0:         if not self.stdouts:
michael@0:             yield iter([])
michael@0:         for s in self.stdouts:
michael@0:             yield iter(s)
michael@0: 
michael@0:     def test_copy_debug_universal(self):
michael@0:         """
michael@0:         Test that dumping symbols for multiple architectures only copies debug symbols once
michael@0:         per file.
michael@0:         """
michael@0:         copied = []
michael@0:         def mock_copy_debug(filename, debug_file, guid):
michael@0:             copied.append(filename[len(self.symbol_dir):] if filename.startswith(self.symbol_dir) else filename)
michael@0:         self.add_test_files(add_extension(["foo"]))
michael@0:         self.stdouts.append(mock_dump_syms("X" * 33, add_extension(["foo"])[0]))
michael@0:         self.stdouts.append(mock_dump_syms("Y" * 33, add_extension(["foo"])[0]))
michael@0:         d = symbolstore.GetPlatformSpecificDumper(dump_syms="dump_syms",
michael@0:                                                   symbol_path=self.symbol_dir,
michael@0:                                                   copy_debug=True,
michael@0:                                                   archs="abc xyz")
michael@0:         d.CopyDebug = mock_copy_debug
michael@0:         d.Process(self.test_dir)
michael@0:         d.Finish(stop_pool=False)
michael@0:         self.assertEqual(1, len(copied))
michael@0: 
michael@0: class TestGetVCSFilename(HelperMixin, unittest.TestCase):
michael@0:     def setUp(self):
michael@0:         HelperMixin.setUp(self)
michael@0: 
michael@0:     def tearDown(self):
michael@0:         HelperMixin.tearDown(self)
michael@0: 
michael@0:     @patch("subprocess.Popen")
michael@0:     def testVCSFilenameHg(self, mock_Popen):
michael@0:         # mock calls to `hg parent` and `hg showconfig paths.default`
michael@0:         mock_communicate = mock_Popen.return_value.communicate
michael@0:         mock_communicate.side_effect = [("abcd1234", ""),
michael@0:                                         ("http://example.com/repo", "")]
michael@0:         os.mkdir(os.path.join(self.test_dir, ".hg"))
michael@0:         filename = os.path.join(self.test_dir, "foo.c")
michael@0:         self.assertEqual("hg:example.com/repo:foo.c:abcd1234",
michael@0:                          symbolstore.GetVCSFilename(filename, [self.test_dir])[0])
michael@0: 
michael@0:     @patch("subprocess.Popen")
michael@0:     def testVCSFilenameHgMultiple(self, mock_Popen):
michael@0:         # mock calls to `hg parent` and `hg showconfig paths.default`
michael@0:         mock_communicate = mock_Popen.return_value.communicate
michael@0:         mock_communicate.side_effect = [("abcd1234", ""),
michael@0:                                         ("http://example.com/repo", ""),
michael@0:                                         ("0987ffff", ""),
michael@0:                                         ("http://example.com/other", "")]
michael@0:         srcdir1 = os.path.join(self.test_dir, "one")
michael@0:         srcdir2 = os.path.join(self.test_dir, "two")
michael@0:         os.makedirs(os.path.join(srcdir1, ".hg"))
michael@0:         os.makedirs(os.path.join(srcdir2, ".hg"))
michael@0:         filename1 = os.path.join(srcdir1, "foo.c")
michael@0:         filename2 = os.path.join(srcdir2, "bar.c")
michael@0:         self.assertEqual("hg:example.com/repo:foo.c:abcd1234",
michael@0:                          symbolstore.GetVCSFilename(filename1, [srcdir1, srcdir2])[0])
michael@0:         self.assertEqual("hg:example.com/other:bar.c:0987ffff",
michael@0:                          symbolstore.GetVCSFilename(filename2, [srcdir1, srcdir2])[0])
michael@0: 
michael@0: class TestRepoManifest(HelperMixin, unittest.TestCase):
michael@0:     def testRepoManifest(self):
michael@0:         manifest = os.path.join(self.test_dir, "sources.xml")
michael@0:         open(manifest, "w").write("""<?xml version="1.0" encoding="UTF-8"?>
michael@0: <manifest>
michael@0: <remote fetch="http://example.com/foo/" name="foo"/>
michael@0: <remote fetch="git://example.com/bar/" name="bar"/>
michael@0: <default remote="bar"/>
michael@0: <project name="projects/one" revision="abcd1234"/>
michael@0: <project name="projects/two" path="projects/another" revision="ffffffff" remote="foo"/>
michael@0: <project name="something_else" revision="00000000" remote="bar"/>
michael@0: </manifest>
michael@0: """)
michael@0:         # Use a source file from each of the three projects
michael@0:         file1 = os.path.join(self.test_dir, "projects", "one", "src1.c")
michael@0:         file2 = os.path.join(self.test_dir, "projects", "another", "src2.c")
michael@0:         file3 = os.path.join(self.test_dir, "something_else", "src3.c")
michael@0:         d = symbolstore.Dumper("dump_syms", "symbol_path",
michael@0:                                repo_manifest=manifest)
michael@0:         self.assertEqual("git:example.com/bar/projects/one:src1.c:abcd1234",
michael@0:                          symbolstore.GetVCSFilename(file1, d.srcdirs)[0])
michael@0:         self.assertEqual("git:example.com/foo/projects/two:src2.c:ffffffff",
michael@0:                          symbolstore.GetVCSFilename(file2, d.srcdirs)[0])
michael@0:         self.assertEqual("git:example.com/bar/something_else:src3.c:00000000",
michael@0:                          symbolstore.GetVCSFilename(file3, d.srcdirs)[0])
michael@0: 
michael@0: if platform.system() in ("Windows", "Microsoft"):
michael@0:     class TestSourceServer(HelperMixin, unittest.TestCase):
michael@0:         @patch("subprocess.call")
michael@0:         @patch("subprocess.Popen")
michael@0:         def test_HGSERVER(self, mock_Popen, mock_call):
michael@0:             """
michael@0:             Test that HGSERVER gets set correctly in the source server index.
michael@0:             """
michael@0:             symbolpath = os.path.join(self.test_dir, "symbols")
michael@0:             os.makedirs(symbolpath)
michael@0:             srcdir = os.path.join(self.test_dir, "srcdir")
michael@0:             os.makedirs(os.path.join(srcdir, ".hg"))
michael@0:             sourcefile = os.path.join(srcdir, "foo.c")
michael@0:             test_files = add_extension(["foo"])
michael@0:             self.add_test_files(test_files)
michael@0:             # srcsrv needs PDBSTR_PATH set
michael@0:             os.environ["PDBSTR_PATH"] = "pdbstr"
michael@0:             # mock calls to `dump_syms`, `hg parent` and
michael@0:             # `hg showconfig paths.default`
michael@0:             mock_Popen.return_value.stdout = iter([
michael@0:                     "MODULE os x86 %s %s" % ("X" * 33, test_files[0]),
michael@0:                     "FILE 0 %s" % sourcefile,
michael@0:                     "PUBLIC xyz 123"
michael@0:                     ])
michael@0:             mock_communicate = mock_Popen.return_value.communicate
michael@0:             mock_communicate.side_effect = [("abcd1234", ""),
michael@0:                                             ("http://example.com/repo", ""),
michael@0:                                             ]
michael@0:             # And mock the call to pdbstr to capture the srcsrv stream data.
michael@0:             global srcsrv_stream
michael@0:             srcsrv_stream = None
michael@0:             def mock_pdbstr(args, cwd="", **kwargs):
michael@0:                 for arg in args:
michael@0:                     if arg.startswith("-i:"):
michael@0:                         global srcsrv_stream
michael@0:                         srcsrv_stream = open(os.path.join(cwd, arg[3:]), 'r').read()
michael@0:                 return 0
michael@0:             mock_call.side_effect = mock_pdbstr
michael@0:             d = symbolstore.GetPlatformSpecificDumper(dump_syms="dump_syms",
michael@0:                                                       symbol_path=symbolpath,
michael@0:                                                       srcdirs=[srcdir],
michael@0:                                                       vcsinfo=True,
michael@0:                                                       srcsrv=True,
michael@0:                                                       copy_debug=True)
michael@0:             # stub out CopyDebug
michael@0:             d.CopyDebug = lambda *args: True
michael@0:             d.Process(self.test_dir)
michael@0:             d.Finish(stop_pool=False)
michael@0:             self.assertNotEqual(srcsrv_stream, None)
michael@0:             hgserver = [x.rstrip() for x in srcsrv_stream.splitlines() if x.startswith("HGSERVER=")]
michael@0:             self.assertEqual(len(hgserver), 1)
michael@0:             self.assertEqual(hgserver[0].split("=")[1], "http://example.com/repo")
michael@0: 
michael@0: if __name__ == '__main__':
michael@0:     # use the multiprocessing.dummy module to use threading wrappers so
michael@0:     # that our mocking/module-patching works
michael@0:     symbolstore.Dumper.GlobalInit(module=multiprocessing.dummy)
michael@0: 
michael@0:     unittest.main()
michael@0: 
michael@0:     symbolstore.Dumper.pool.close()
michael@0:     symbolstore.Dumper.pool.join()