| # Copyright 2013 The ChromiumOS Authors |
| # Use of this source code is governed by a BSD-style license that can be |
| # found in the LICENSE file. |
| |
| # TODO: We removed --network integration tests. |
| |
| """Unittests for upload_symbols.py""" |
| |
| import errno |
| import http.server |
| import itertools |
| import os |
| import signal |
| import socket |
| import socketserver |
| import sys |
| import time |
| from unittest import mock |
| import urllib.request |
| |
| import pytest # pylint: disable=import-error |
| |
| |
| # We specifically set up a local server to connect to, so make sure we |
| # delete any proxy settings that might screw that up. We also need to |
| # do it here because modules that are imported below will implicitly |
| # initialize with this proxy setting rather than dynamically pull it |
| # on the fly :(. |
| # pylint: disable=wrong-import-position |
| os.environ.pop("http_proxy", None) |
| |
| from chromite.lib import constants |
| |
| |
| # The isolateserver includes a bunch of third_party python packages that clash |
| # with chromite's bundled third_party python packages (like oauth2client). |
| # Since upload_symbols is not imported in to other parts of chromite, and there |
| # are no deps in third_party we care about, purge the chromite copy. This way |
| # we can use isolateserver for deduping. |
| # TODO: If we ever sort out third_party/ handling and make it per-script opt-in, |
| # we can purge this logic. |
| third_party = str(constants.CHROMITE_DIR / "third_party") |
| while True: |
| try: |
| sys.path.remove(third_party) |
| except ValueError: |
| break |
| del third_party |
| |
| from chromite.cbuildbot import cbuildbot_alerts |
| from chromite.lib import cros_test_lib |
| from chromite.lib import osutils |
| from chromite.lib import remote_access |
| from chromite.scripts import cros_generate_breakpad_symbols |
| from chromite.scripts import upload_symbols |
| |
| |
| class SymbolsTestBase(cros_test_lib.MockTempDirTestCase): |
| """Base class for most symbols tests.""" |
| |
| SLIM_CONTENT = """ |
| some junk |
| """ |
| |
| FAT_CONTENT = """ |
| STACK CFI 1234 |
| some junk |
| STACK CFI 1234 |
| """ |
| |
| def setUp(self) -> None: |
| # Make certain we don't use the network. |
| self.urlopen_mock = self.PatchObject(urllib.request, "urlopen") |
| self.request_mock = self.PatchObject( |
| upload_symbols, |
| "ExecRequest", |
| return_value={"uploadUrl": "testurl", "uploadKey": "asdgasgas"}, |
| ) |
| |
| # Make 'uploads' go fast. |
| self.PatchObject(upload_symbols, "SLEEP_DELAY", 0) |
| self.PatchObject(upload_symbols, "INITIAL_RETRY_DELAY", 0) |
| |
| # So our symbol file content doesn't have to be real. |
| self.PatchObject( |
| cros_generate_breakpad_symbols, |
| "ReadSymsHeader", |
| return_value=cros_generate_breakpad_symbols.SymbolHeader( |
| os="os", cpu="cpu", id="id", name="name" |
| ), |
| ) |
| |
| self.working = os.path.join(self.tempdir, "expand") |
| osutils.SafeMakedirs(self.working) |
| |
| self.data = os.path.join(self.tempdir, "data") |
| osutils.SafeMakedirs(self.data) |
| |
| def createSymbolFile( |
| self, filename, content=FAT_CONTENT, size=0, status=None, dedupe=False |
| ): |
| fullname = os.path.join(self.data, filename) |
| osutils.SafeMakedirs(os.path.dirname(fullname)) |
| |
| # If a file size is given, force that to be the minimum file size. |
| # Create a sparse file so large files are practical. |
| with open(fullname, "w+b") as f: |
| f.truncate(size) |
| f.seek(0) |
| f.write(content.encode("utf-8")) |
| |
| result = upload_symbols.SymbolFile( |
| display_path=filename, file_name=fullname |
| ) |
| |
| if status: |
| result.status = status |
| |
| if dedupe: |
| result.dedupe_item = upload_symbols.DedupeItem(result) |
| result.dedupe_push_state = "push_state" |
| |
| return result |
| |
| |
| class SymbolServerRequestHandler(http.server.BaseHTTPRequestHandler): |
| """HTTP handler for symbol POSTs""" |
| |
| RESP_CODE = None |
| RESP_MSG = None |
| |
| def do_POST(self) -> None: |
| """Handle a POST request""" |
| # Drain the data from the client. If we don't, we might write the |
| # response and close the socket before the client finishes, so they die |
| # with EPIPE. |
| clen = int(self.headers.get("Content-Length", "0")) |
| self.rfile.read(clen) |
| |
| self.send_response(self.RESP_CODE, self.RESP_MSG) |
| self.end_headers() |
| |
| # pylint: disable=arguments-differ |
| def log_message(self, *args, **kwargs) -> None: |
| """Stub the logger as it writes to stderr""" |
| |
| |
| class SymbolServer(socketserver.ThreadingTCPServer, http.server.HTTPServer): |
| """Simple HTTP server that forks each request""" |
| |
| |
| @pytest.mark.usefixtures("singleton_manager") |
| class UploadSymbolsServerTest(cros_test_lib.MockTempDirTestCase): |
| """Tests for UploadSymbols() and a local HTTP server""" |
| |
| SYM_CONTENTS = """MODULE Linux arm 123-456 blkid |
| PUBLIC 1471 0 main""" |
| |
| def SpawnServer(self, RequestHandler) -> None: |
| """Spawn a new http server""" |
| while True: |
| try: |
| port = remote_access.GetUnusedPort() |
| address = ("", port) |
| self.httpd = SymbolServer(address, RequestHandler) |
| break |
| except socket.error as e: |
| if e.errno == errno.EADDRINUSE: |
| continue |
| raise |
| self.server_url = "http://localhost:%i/post/path" % port |
| self.httpd_pid = os.fork() |
| if self.httpd_pid == 0: |
| self.httpd.serve_forever(poll_interval=0.1) |
| sys.exit(0) |
| # The child runs the server, so close the socket in the parent. |
| self.httpd.server_close() |
| |
| def setUp(self) -> None: |
| self.httpd_pid = None |
| self.httpd = None |
| self.server_url = None |
| self.sym_file = os.path.join(self.tempdir, "test.sym") |
| osutils.WriteFile(self.sym_file, self.SYM_CONTENTS) |
| |
| # Stop sleeps and retries for these tests. |
| self.PatchObject(upload_symbols, "SLEEP_DELAY", 0) |
| self.PatchObject(upload_symbols, "INITIAL_RETRY_DELAY", 0) |
| self.PatchObject(upload_symbols, "MAX_RETRIES", 0) |
| |
| def tearDown(self) -> None: |
| # Only kill the server if we forked one. |
| if self.httpd_pid: |
| os.kill(self.httpd_pid, signal.SIGUSR1) |
| |
| def testSuccess(self) -> None: |
| """The server returns success for all uploads""" |
| |
| class Handler(SymbolServerRequestHandler): |
| """Always return 200""" |
| |
| RESP_CODE = 200 |
| self.PatchObject( |
| upload_symbols, |
| "ExecRequest", |
| return_value={ |
| "uploadUrl": "testurl", |
| "uploadKey": "testSuccess", |
| }, |
| ) |
| |
| self.SpawnServer(Handler) |
| ret = upload_symbols.UploadSymbols( |
| sym_paths=[self.sym_file] * 10, |
| upload_url=self.server_url, |
| api_key="testSuccess", |
| ) |
| self.assertEqual(ret, 0) |
| |
| def testError(self) -> None: |
| """The server returns errors for all uploads""" |
| |
| class Handler(SymbolServerRequestHandler): |
| """All connections error""" |
| |
| RESP_CODE = 500 |
| RESP_MSG = "Internal Server Error" |
| |
| self.SpawnServer(Handler) |
| ret = upload_symbols.UploadSymbols( |
| sym_paths=[self.sym_file] * 10, |
| upload_url=self.server_url, |
| api_key="testkey", |
| ) |
| self.assertEqual(ret, 10) |
| |
| def testHungServer(self) -> None: |
| """The server chokes, but we recover""" |
| |
| class Handler(SymbolServerRequestHandler): |
| """All connections choke forever""" |
| |
| self.PatchObject( |
| upload_symbols, "ExecRequest", return_value={"pairs": []} |
| ) |
| |
| def do_POST(self) -> None: |
| while True: |
| time.sleep(1000) |
| |
| self.SpawnServer(Handler) |
| with mock.patch.object(upload_symbols, "GetUploadTimeout") as m: |
| m.return_value = 0.01 |
| ret = upload_symbols.UploadSymbols( |
| sym_paths=[self.sym_file] * 10, |
| upload_url=self.server_url, |
| timeout=m.return_value, |
| api_key="testkey", |
| ) |
| self.assertEqual(ret, 10) |
| |
| |
| class UploadSymbolsHelpersTest(cros_test_lib.TestCase): |
| """Test assorted helper functions and classes.""" |
| |
| def testIsTarball(self) -> None: |
| notTar = [ |
| "/foo/bar/test.bin", |
| "/foo/bar/test.tar.bin", |
| "/foo/bar/test.faketar.gz", |
| "/foo/bar/test.nottgz", |
| ] |
| |
| isTar = [ |
| "/foo/bar/test.tar", |
| "/foo/bar/test.bin.tar", |
| "/foo/bar/test.bin.tar.bz2", |
| "/foo/bar/test.bin.tar.gz", |
| "/foo/bar/test.bin.tar.xz", |
| "/foo/bar/test.tbz2", |
| "/foo/bar/test.tbz", |
| "/foo/bar/test.tgz", |
| "/foo/bar/test.txz", |
| ] |
| |
| for p in notTar: |
| self.assertFalse(upload_symbols.IsTarball(p)) |
| |
| for p in isTar: |
| self.assertTrue(upload_symbols.IsTarball(p)) |
| |
| def testBatchGenerator(self) -> None: |
| result = upload_symbols.BatchGenerator([], 2) |
| self.assertEqual(list(result), []) |
| |
| result = upload_symbols.BatchGenerator(range(6), 2) |
| self.assertEqual(list(result), [[0, 1], [2, 3], [4, 5]]) |
| |
| result = upload_symbols.BatchGenerator(range(7), 2) |
| self.assertEqual(list(result), [[0, 1], [2, 3], [4, 5], [6]]) |
| |
| # Prove that we are streaming the results, not generating them all at |
| # once. |
| result = upload_symbols.BatchGenerator(itertools.repeat(0), 2) |
| self.assertEqual(next(result), [0, 0]) |
| |
| |
| class FindSymbolFilesTest(SymbolsTestBase): |
| """Test FindSymbolFiles.""" |
| |
| def setUp(self) -> None: |
| self.symfile = self.createSymbolFile("root.sym").file_name |
| self.innerfile = self.createSymbolFile( |
| os.path.join("nested", "inner.sym") |
| ).file_name |
| |
| # CreateTarball is having issues outside the chroot from open file |
| # tests. |
| # |
| # self.tarball = os.path.join(self.tempdir, 'syms.tar.gz') |
| # cros_build_lib.CreateTarball( |
| # 'syms.tar.gz', self.tempdir, inputs=(self.data)) |
| |
| def testEmpty(self) -> None: |
| symbols = list(upload_symbols.FindSymbolFiles(self.working, [])) |
| self.assertEqual(symbols, []) |
| |
| def testFile(self) -> None: |
| symbols = list( |
| upload_symbols.FindSymbolFiles(self.working, [self.symfile]) |
| ) |
| |
| self.assertEqual(len(symbols), 1) |
| sf = symbols[0] |
| |
| self.assertEqual(sf.display_name, "root.sym") |
| self.assertEqual(sf.display_path, self.symfile) |
| self.assertEqual(sf.file_name, self.symfile) |
| self.assertEqual(sf.status, upload_symbols.SymbolFile.INITIAL) |
| self.assertEqual(sf.FileSize(), len(self.FAT_CONTENT)) |
| |
| def testDir(self) -> None: |
| symbols = list( |
| upload_symbols.FindSymbolFiles(self.working, [self.data]) |
| ) |
| |
| self.assertEqual(len(symbols), 2) |
| root = symbols[0] |
| nested = symbols[1] |
| |
| self.assertEqual(root.display_name, "root.sym") |
| self.assertEqual(root.display_path, "root.sym") |
| self.assertEqual(root.file_name, self.symfile) |
| self.assertEqual(root.status, upload_symbols.SymbolFile.INITIAL) |
| self.assertEqual(root.FileSize(), len(self.FAT_CONTENT)) |
| |
| self.assertEqual(nested.display_name, "inner.sym") |
| self.assertEqual(nested.display_path, "nested/inner.sym") |
| self.assertEqual(nested.file_name, self.innerfile) |
| self.assertEqual(nested.status, upload_symbols.SymbolFile.INITIAL) |
| self.assertEqual(nested.FileSize(), len(self.FAT_CONTENT)) |
| |
| |
| class AdjustSymbolFileSizeTest(SymbolsTestBase): |
| """Test AdjustSymbolFileSize.""" |
| |
| def setUp(self) -> None: |
| self.slim = self.createSymbolFile("slim.sym", self.SLIM_CONTENT) |
| self.fat = self.createSymbolFile("fat.sym", self.FAT_CONTENT) |
| |
| self.warn_mock = self.PatchObject( |
| cbuildbot_alerts, "PrintBuildbotStepWarnings" |
| ) |
| |
| def _testNotStripped(self, symbol, size=None, content=None) -> None: |
| start_file = symbol.file_name |
| after = upload_symbols.AdjustSymbolFileSize(symbol, self.working, size) |
| self.assertIs(after, symbol) |
| self.assertEqual(after.file_name, start_file) |
| if content is not None: |
| self.assertEqual(osutils.ReadFile(after.file_name), content) |
| |
| def _testStripped(self, symbol, size=None, content=None) -> None: |
| after = upload_symbols.AdjustSymbolFileSize(symbol, self.working, size) |
| self.assertIs(after, symbol) |
| self.assertTrue(after.file_name.startswith(self.working)) |
| if content is not None: |
| self.assertEqual(osutils.ReadFile(after.file_name), content) |
| |
| def testSmall(self) -> None: |
| """Ensure that files smaller than the limit are not modified.""" |
| self._testNotStripped(self.slim, 1024, self.SLIM_CONTENT) |
| self._testNotStripped(self.fat, 1024, self.FAT_CONTENT) |
| |
| def testLarge(self) -> None: |
| """Ensure that files larger than the limit are modified.""" |
| self._testStripped(self.slim, 1, self.SLIM_CONTENT) |
| self._testStripped(self.fat, 1, self.SLIM_CONTENT) |
| |
| def testMixed(self) -> None: |
| """Test mix of large and small.""" |
| strip_size = len(self.SLIM_CONTENT) + 1 |
| |
| self._testNotStripped(self.slim, strip_size, self.SLIM_CONTENT) |
| self._testStripped(self.fat, strip_size, self.SLIM_CONTENT) |
| |
| def testSizeWarnings(self) -> None: |
| large = self.createSymbolFile( |
| "large.sym", |
| content=self.SLIM_CONTENT, |
| size=upload_symbols.CRASH_SERVER_FILE_LIMIT * 2, |
| ) |
| |
| # Would like to Strip as part of this test, but that really copies all |
| # of the sparse file content, which is too expensive for a unittest. |
| self._testNotStripped(large, None, None) |
| |
| self.assertEqual(self.warn_mock.call_count, 1) |
| |
| |
| class DeduplicateTest(SymbolsTestBase): |
| """Test server Deduplication.""" |
| |
| def setUp(self) -> None: |
| self.PatchObject( |
| upload_symbols, |
| "ExecRequest", |
| return_value={ |
| "pairs": [ |
| { |
| "status": "FOUND", |
| "symbolId": { |
| "debugFile": "sym1_sym", |
| "debugId": "BEAA9BE", |
| }, |
| }, |
| { |
| "status": "FOUND", |
| "symbolId": { |
| "debugFile": "sym2_sym", |
| "debugId": "B6B1A36", |
| }, |
| }, |
| { |
| "status": "MISSING", |
| "symbolId": { |
| "debugFile": "sym3_sym", |
| "debugId": "D4FC0FC", |
| }, |
| }, |
| ] |
| }, |
| ) |
| |
| def testFindDuplicates(self) -> None: |
| # The first two symbols will be duplicate, the third new. |
| sym1 = self.createSymbolFile("sym1.sym") |
| sym1.header = cros_generate_breakpad_symbols.SymbolHeader( |
| "cpu", "BEAA9BE", "sym1_sym", "os" |
| ) |
| sym2 = self.createSymbolFile("sym2.sym") |
| sym2.header = cros_generate_breakpad_symbols.SymbolHeader( |
| "cpu", "B6B1A36", "sym2_sym", "os" |
| ) |
| sym3 = self.createSymbolFile("sym3.sym") |
| sym3.header = cros_generate_breakpad_symbols.SymbolHeader( |
| "cpu", "D4FC0FC", "sym3_sym", "os" |
| ) |
| |
| result = upload_symbols.FindDuplicates( |
| (sym1, sym2, sym3), "fake_url", api_key="testkey" |
| ) |
| self.assertEqual(list(result), [sym1, sym2, sym3]) |
| |
| self.assertEqual(sym1.status, upload_symbols.SymbolFile.DUPLICATE) |
| self.assertEqual(sym2.status, upload_symbols.SymbolFile.DUPLICATE) |
| self.assertEqual(sym3.status, upload_symbols.SymbolFile.INITIAL) |
| |
| |
| class PerformSymbolFilesUploadTest(SymbolsTestBase): |
| """Test PerformSymbolFile, and it's helper methods.""" |
| |
| def setUp(self) -> None: |
| self.sym_initial = self.createSymbolFile("initial.sym") |
| self.sym_error = self.createSymbolFile( |
| "error.sym", status=upload_symbols.SymbolFile.ERROR |
| ) |
| self.sym_duplicate = self.createSymbolFile( |
| "duplicate.sym", status=upload_symbols.SymbolFile.DUPLICATE |
| ) |
| self.sym_uploaded = self.createSymbolFile( |
| "uploaded.sym", status=upload_symbols.SymbolFile.UPLOADED |
| ) |
| |
| def testGetUploadTimeout(self) -> None: |
| """Test GetUploadTimeout helper function.""" |
| # Timeout for small file. |
| self.assertEqual( |
| upload_symbols.GetUploadTimeout(self.sym_initial), |
| upload_symbols.UPLOAD_MIN_TIMEOUT, |
| ) |
| |
| # Timeout for 512M file. |
| large = self.createSymbolFile("large.sym", size=(512 * 1024 * 1024)) |
| self.assertEqual(upload_symbols.GetUploadTimeout(large), 15 * 60) |
| |
| def testUploadSymbolFile(self) -> None: |
| upload_symbols.UploadSymbolFile( |
| "fake_url", self.sym_initial, api_key="testkey" |
| ) |
| # TODO: Examine mock in more detail to make sure request is correct. |
| self.assertEqual(self.request_mock.call_count, 3) |
| |
| def testPerformSymbolsFileUpload(self) -> None: |
| """We upload on first try.""" |
| symbols = [self.sym_initial] |
| |
| result = upload_symbols.PerformSymbolsFileUpload( |
| symbols, "fake_url", api_key="testkey" |
| ) |
| |
| self.assertEqual(list(result), symbols) |
| self.assertEqual( |
| self.sym_initial.status, upload_symbols.SymbolFile.UPLOADED |
| ) |
| self.assertEqual(self.request_mock.call_count, 3) |
| |
| def testPerformSymbolsFileUploadFailure(self) -> None: |
| """All network requests fail.""" |
| self.request_mock.side_effect = IOError("network failure") |
| symbols = [self.sym_initial] |
| |
| result = upload_symbols.PerformSymbolsFileUpload( |
| symbols, "fake_url", api_key="testkey" |
| ) |
| |
| self.assertEqual(list(result), symbols) |
| self.assertEqual( |
| self.sym_initial.status, upload_symbols.SymbolFile.ERROR |
| ) |
| self.assertEqual(self.request_mock.call_count, 6) |
| |
| def testPerformSymbolsFileUploadTransisentFailure(self) -> None: |
| """We fail once, then succeed.""" |
| self.urlopen_mock.side_effect = (IOError("network failure"), None) |
| symbols = [self.sym_initial] |
| |
| result = upload_symbols.PerformSymbolsFileUpload( |
| symbols, "fake_url", api_key="testkey" |
| ) |
| |
| self.assertEqual(list(result), symbols) |
| self.assertEqual( |
| self.sym_initial.status, upload_symbols.SymbolFile.UPLOADED |
| ) |
| self.assertEqual(self.request_mock.call_count, 3) |
| |
| def testPerformSymbolsFileUploadMixed(self) -> None: |
| """Upload symbols in mixed starting states. |
| |
| Demonstrate that INITIAL and ERROR are uploaded, but DUPLICATE/UPLOADED |
| are ignored. |
| """ |
| symbols = [ |
| self.sym_initial, |
| self.sym_error, |
| self.sym_duplicate, |
| self.sym_uploaded, |
| ] |
| |
| result = upload_symbols.PerformSymbolsFileUpload( |
| symbols, "fake_url", api_key="testkey" |
| ) |
| |
| # |
| self.assertEqual(list(result), symbols) |
| self.assertEqual( |
| self.sym_initial.status, upload_symbols.SymbolFile.UPLOADED |
| ) |
| self.assertEqual( |
| self.sym_error.status, upload_symbols.SymbolFile.UPLOADED |
| ) |
| self.assertEqual( |
| self.sym_duplicate.status, upload_symbols.SymbolFile.DUPLICATE |
| ) |
| self.assertEqual( |
| self.sym_uploaded.status, upload_symbols.SymbolFile.UPLOADED |
| ) |
| self.assertEqual(self.request_mock.call_count, 6) |
| |
| def testPerformSymbolsFileUploadErrorOut(self) -> None: |
| """Demonstate we exit only after X errors.""" |
| |
| symbol_count = upload_symbols.MAX_TOTAL_ERRORS_FOR_RETRY + 10 |
| symbols = [] |
| fail_file = None |
| |
| # potentially twice as many errors as we should attempt. |
| for _ in range(symbol_count): |
| # Each loop will get unique SymbolFile instances that use the same |
| # files. |
| fail = self.createSymbolFile("fail.sym") |
| fail_file = fail.file_name |
| symbols.append(self.createSymbolFile("pass.sym")) |
| symbols.append(fail) |
| |
| # Mock out UploadSymbolFile and fail for fail.sym files. |
| def failSome(_url, symbol, _api_key) -> None: |
| if symbol.file_name == fail_file: |
| raise IOError("network failure") |
| |
| upload_mock = self.PatchObject( |
| upload_symbols, "UploadSymbolFile", side_effect=failSome |
| ) |
| upload_mock.__name__ = "UploadSymbolFileMock2" |
| |
| result = upload_symbols.PerformSymbolsFileUpload( |
| symbols, "fake_url", api_key="testkey" |
| ) |
| |
| self.assertEqual(list(result), symbols) |
| |
| passed = sum( |
| s.status == upload_symbols.SymbolFile.UPLOADED for s in symbols |
| ) |
| failed = sum( |
| s.status == upload_symbols.SymbolFile.ERROR for s in symbols |
| ) |
| skipped = sum( |
| s.status == upload_symbols.SymbolFile.INITIAL for s in symbols |
| ) |
| |
| # Shows we all pass.sym files worked until limit hit. |
| self.assertEqual(passed, upload_symbols.MAX_TOTAL_ERRORS_FOR_RETRY) |
| |
| # Shows we all fail.sym files failed until limit hit. |
| self.assertEqual(failed, upload_symbols.MAX_TOTAL_ERRORS_FOR_RETRY) |
| |
| # Shows both pass/fail were skipped after limit hit. |
| self.assertEqual(skipped, 10 * 2) |
| |
| |
| @pytest.mark.usefixtures("singleton_manager") |
| class UploadSymbolsTest(SymbolsTestBase): |
| """Test UploadSymbols, along with most helper methods.""" |
| |
| def setUp(self) -> None: |
| # Results gathering. |
| self.failure_file = os.path.join(self.tempdir, "failures.txt") |
| |
| def testUploadSymbolsEmpty(self) -> None: |
| """Upload dir is empty.""" |
| result = upload_symbols.UploadSymbols([self.data], "fake_url") |
| |
| self.assertEqual(result, 0) |
| self.assertEqual(self.urlopen_mock.call_count, 0) |
| |
| def testUploadSymbols(self) -> None: |
| """Upload a few files.""" |
| self.createSymbolFile("slim.sym", self.SLIM_CONTENT) |
| self.createSymbolFile(os.path.join("nested", "inner.sym")) |
| self.createSymbolFile("fat.sym", self.FAT_CONTENT) |
| |
| result = upload_symbols.UploadSymbols( |
| [self.data], |
| "fake_url", |
| failed_list=self.failure_file, |
| strip_cfi=len(self.SLIM_CONTENT) + 1, |
| api_key="testkey", |
| ) |
| |
| self.assertEqual(result, 0) |
| self.assertEqual(self.request_mock.call_count, 10) |
| self.assertEqual(osutils.ReadFile(self.failure_file), "") |
| |
| def testUploadSymbolsLimited(self) -> None: |
| """Upload a few files.""" |
| self.createSymbolFile("slim.sym", self.SLIM_CONTENT) |
| self.createSymbolFile(os.path.join("nested", "inner.sym")) |
| self.createSymbolFile("fat.sym", self.FAT_CONTENT) |
| |
| result = upload_symbols.UploadSymbols( |
| [self.data], "fake_url", upload_limit=2, api_key="testkey" |
| ) |
| |
| self.assertEqual(result, 0) |
| self.assertEqual(self.request_mock.call_count, 7) |
| self.assertNotExists(self.failure_file) |
| |
| def testUploadSymbolsFailures(self) -> None: |
| """Upload a few files.""" |
| self.createSymbolFile("pass.sym") |
| fail = self.createSymbolFile("fail.sym") |
| |
| def failSome(_url, symbol, _api_key) -> None: |
| if symbol.file_name == fail.file_name: |
| raise IOError("network failure") |
| |
| # Mock out UploadSymbolFile so it's easy to see which file to fail for. |
| upload_mock = self.PatchObject( |
| upload_symbols, "UploadSymbolFile", side_effect=failSome |
| ) |
| # Mock __name__ for logging. |
| upload_mock.__name__ = "UploadSymbolFileMock" |
| |
| result = upload_symbols.UploadSymbols( |
| [self.data], |
| "fake_url", |
| failed_list=self.failure_file, |
| api_key="testkey", |
| ) |
| |
| self.assertEqual(result, 1) |
| self.assertEqual(upload_mock.call_count, 7) |
| self.assertEqual(osutils.ReadFile(self.failure_file), "fail.sym\n") |
