# -*- coding: utf-8 -*-
# Copyright (c) 2013 The Chromium OS Authors. All rights reserved.
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.

"""Unittests for upload_symbols.py"""

from __future__ import print_function

import errno
import itertools
import os
import signal
import socket
import sys
import time

import mock
from six.moves import BaseHTTPServer
from six.moves import socketserver
from six.moves import urllib
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 = os.path.join(constants.CHROMITE_DIR, 'third_party')
while True:
  try:
    sys.path.remove(third_party)
  except ValueError:
    break
del third_party

from chromite.lib import cros_logging as logging
from chromite.lib import cros_test_lib
from chromite.lib import osutils
from chromite.lib import parallel
from chromite.lib import remote_access
from chromite.scripts import cros_generate_breakpad_symbols
from chromite.scripts import upload_symbols


assert sys.version_info >= (3, 6), 'This module requires Python 3.6+'


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):
    # 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(BaseHTTPServer.BaseHTTPRequestHandler):
  """HTTP handler for symbol POSTs"""

  RESP_CODE = None
  RESP_MSG = None

  def do_POST(self):
    """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):
    """Stub the logger as it writes to stderr"""


class SymbolServer(socketserver.ThreadingTCPServer, BaseHTTPServer.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):
    """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):
    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):
    # Only kill the server if we forked one.
    if self.httpd_pid:
      os.kill(self.httpd_pid, signal.SIGUSR1)

  def testSuccess(self):
    """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):
    """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):
    """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):
        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):
    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):
    result = upload_symbols.BatchGenerator([], 2)
    self.assertEqual(list(result), [])

    # BatchGenerator accepts iterators, so passing it here is safe.
    # pylint: disable=range-builtin-not-iterating
    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):
    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):
    symbols = list(upload_symbols.FindSymbolFiles(
        self.working, []))
    self.assertEqual(symbols, [])

  def testFile(self):
    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):
    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):
    self.slim = self.createSymbolFile('slim.sym', self.SLIM_CONTENT)
    self.fat = self.createSymbolFile('fat.sym', self.FAT_CONTENT)

    self.warn_mock = self.PatchObject(logging, 'PrintBuildbotStepWarnings')

  def _testNotStripped(self, symbol, size=None, content=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):
    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):
    """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):
    """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):
    """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):
    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):
    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):
    # 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):
    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):
    """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):
    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):
    """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):
    """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):
    """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):
    """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):
    """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):
      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):
    # Results gathering.
    self.failure_file = os.path.join(self.tempdir, 'failures.txt')

  def testUploadSymbolsEmpty(self):
    """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):
    """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):
    """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):
    """Upload a few files."""
    self.createSymbolFile('pass.sym')
    fail = self.createSymbolFile('fail.sym')

    def failSome(_url, symbol, _api_key):
      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')

# TODO: We removed --network integration tests.


def main(_argv):
  # pylint: disable=protected-access
  # Set timeouts small so that if the unit test hangs, it won't hang for long.
  parallel._BackgroundTask.STARTUP_TIMEOUT = 5
  parallel._BackgroundTask.EXIT_TIMEOUT = 5

  # Run the tests.
  cros_test_lib.main(level='info', module=__name__)