# Copyright 2015 The Chromium Authors. All rights reserved.
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.

import os
import shutil
import stat
import sys
import tempfile
import unittest
import uuid
import zipfile

import mock

from dependency_manager import dependency_manager_util
from dependency_manager import exceptions


class DependencyManagerUtilTest(unittest.TestCase):
  # This class intentionally uses actual file I/O to test real system behavior.

  def setUp(self):
    self.tmp_dir = os.path.abspath(tempfile.mkdtemp(prefix='telemetry'))
    self.sub_dir = os.path.join(self.tmp_dir, 'sub_dir')
    os.mkdir(self.sub_dir)

    self.read_only_path = (os.path.join(self.tmp_dir, 'read_only'))
    with open(self.read_only_path, 'w+') as read_file:
      read_file.write('Read-only file')
    os.chmod(self.read_only_path, stat.S_IRUSR)

    self.writable_path = (os.path.join(self.tmp_dir, 'writable'))
    with open(self.writable_path, 'w+') as writable_file:
      writable_file.write('Writable file')
    os.chmod(self.writable_path, stat.S_IRUSR | stat.S_IWUSR)

    self.executable_path = (os.path.join(self.tmp_dir, 'executable'))
    with open(self.executable_path, 'w+') as executable_file:
      executable_file.write('Executable file')
    os.chmod(self.executable_path, stat.S_IRWXU)

    self.sub_read_only_path = (os.path.join(self.sub_dir, 'read_only'))
    with open(self.sub_read_only_path, 'w+') as read_file:
      read_file.write('Read-only sub file')
    os.chmod(self.sub_read_only_path, stat.S_IRUSR)

    self.sub_writable_path = (os.path.join(self.sub_dir, 'writable'))
    with open(self.sub_writable_path, 'w+') as writable_file:
      writable_file.write('Writable sub file')
    os.chmod(self.sub_writable_path, stat.S_IRUSR | stat.S_IWUSR)

    self.sub_executable_path = (os.path.join(self.sub_dir, 'executable'))
    with open(self.sub_executable_path, 'w+') as executable_file:
      executable_file.write('Executable sub file')
    os.chmod(self.sub_executable_path, stat.S_IRWXU)

    self.AssertExpectedDirFiles(self.tmp_dir)
    self.archive_path = self.CreateZipArchiveFromDir(self.tmp_dir)

  def tearDown(self):
    if os.path.isdir(self.tmp_dir):
      dependency_manager_util.RemoveDir(self.tmp_dir)
    if os.path.isfile(self.archive_path):
      os.remove(self.archive_path)

  def AssertExpectedDirFiles(self, top_dir):
    sub_dir = os.path.join(top_dir, 'sub_dir')
    read_only_path = (os.path.join(top_dir, 'read_only'))
    writable_path = (os.path.join(top_dir, 'writable'))
    executable_path = (os.path.join(top_dir, 'executable'))
    sub_read_only_path = (os.path.join(sub_dir, 'read_only'))
    sub_writable_path = (os.path.join(sub_dir, 'writable'))
    sub_executable_path = (os.path.join(sub_dir, 'executable'))
    # assert contents as expected
    self.assertTrue(os.path.isdir(top_dir))
    self.assertTrue(os.path.isdir(sub_dir))
    self.assertTrue(os.path.isfile(read_only_path))
    self.assertTrue(os.path.isfile(writable_path))
    self.assertTrue(os.path.isfile(executable_path))
    self.assertTrue(os.path.isfile(sub_read_only_path))
    self.assertTrue(os.path.isfile(sub_writable_path))
    self.assertTrue(os.path.isfile(sub_executable_path))

    # assert permissions as expected
    self.assertTrue(
        stat.S_IRUSR & stat.S_IMODE(os.stat(read_only_path).st_mode))
    self.assertTrue(
        stat.S_IRUSR & stat.S_IMODE(os.stat(sub_read_only_path).st_mode))
    self.assertTrue(
        stat.S_IRUSR & stat.S_IMODE(os.stat(writable_path).st_mode))
    self.assertTrue(
        stat.S_IWUSR & stat.S_IMODE(os.stat(writable_path).st_mode))
    self.assertTrue(
        stat.S_IRUSR & stat.S_IMODE(os.stat(sub_writable_path).st_mode))
    self.assertTrue(
        stat.S_IWUSR & stat.S_IMODE(os.stat(sub_writable_path).st_mode))
    if not sys.platform.startswith('win'):
      self.assertEqual(
          stat.S_IRWXU,
          stat.S_IRWXU & stat.S_IMODE(os.stat(executable_path).st_mode))
      self.assertEqual(
          stat.S_IRWXU,
          stat.S_IRWXU & stat.S_IMODE(os.stat(sub_executable_path).st_mode))

  def CreateZipArchiveFromDir(self, dir_path):
    try:
      base_path = os.path.join(tempfile.gettempdir(), str(uuid.uuid4()))
      archive_path = shutil.make_archive(base_path, 'zip', dir_path)
      self.assertTrue(os.path.exists(archive_path))
      self.assertTrue(zipfile.is_zipfile(archive_path))
    except:
      if os.path.isfile(archive_path):
        os.remove(archive_path)
      raise
    return archive_path

  def testRemoveDirWithSubDir(self):
    dependency_manager_util.RemoveDir(self.tmp_dir)

    self.assertFalse(os.path.exists(self.tmp_dir))
    self.assertFalse(os.path.exists(self.sub_dir))
    self.assertFalse(os.path.exists(self.read_only_path))
    self.assertFalse(os.path.exists(self.writable_path))
    self.assertFalse(os.path.isfile(self.executable_path))
    self.assertFalse(os.path.exists(self.sub_read_only_path))
    self.assertFalse(os.path.exists(self.sub_writable_path))
    self.assertFalse(os.path.isfile(self.sub_executable_path))

  def testUnzipFile(self):
    self.AssertExpectedDirFiles(self.tmp_dir)
    unzip_path = os.path.join(tempfile.gettempdir(), str(uuid.uuid4()))
    dependency_manager_util.UnzipArchive(self.archive_path, unzip_path)
    self.AssertExpectedDirFiles(unzip_path)
    self.AssertExpectedDirFiles(self.tmp_dir)
    dependency_manager_util.RemoveDir(unzip_path)

  def testUnzipFileContainingLongPath(self):
    try:
      dir_path = self.tmp_dir
      if sys.platform.startswith('win'):
        dir_path = u'\\\\?\\' + dir_path

      archive_suffix = ''
      # 260 is the Windows API path length limit.
      while len(archive_suffix) < 260:
        archive_suffix = os.path.join(archive_suffix, 'really')
      contents_dir_path = os.path.join(dir_path, archive_suffix)
      os.makedirs(contents_dir_path)
      filename = os.path.join(contents_dir_path, 'longpath.txt')
      open(filename, 'a').close()

      base_path = os.path.join(tempfile.gettempdir(), str(uuid.uuid4()))
      archive_path = shutil.make_archive(base_path, 'zip', dir_path)
      self.assertTrue(os.path.exists(archive_path))
      self.assertTrue(zipfile.is_zipfile(archive_path))
    except:
      if os.path.isfile(archive_path):
        os.remove(archive_path)
      raise

    unzip_path = os.path.join(tempfile.gettempdir(), str(uuid.uuid4()))
    dependency_manager_util.UnzipArchive(archive_path, unzip_path)
    dependency_manager_util.RemoveDir(unzip_path)

  def testUnzipFileFailure(self):
    # zipfile is not used on MacOS. See crbug.com/700097.
    if sys.platform.startswith('darwin'):
      return
    unzip_path = os.path.join(tempfile.gettempdir(), str(uuid.uuid4()))
    self.assertFalse(os.path.exists(unzip_path))
    with mock.patch(
        'dependency_manager.dependency_manager_util.zipfile.ZipFile.extractall'  # pylint: disable=line-too-long
        ) as zipfile_mock:
      zipfile_mock.side_effect = IOError
      self.assertRaises(
          IOError, dependency_manager_util.UnzipArchive, self.archive_path,
          unzip_path)
    self.AssertExpectedDirFiles(self.tmp_dir)
    self.assertFalse(os.path.exists(unzip_path))

  def testVerifySafeArchivePasses(self):
    with zipfile.ZipFile(self.archive_path) as archive:
      dependency_manager_util.VerifySafeArchive(archive)

  def testVerifySafeArchiveFailsOnRelativePathWithPardir(self):
    tmp_file = tempfile.NamedTemporaryFile(delete=False)
    tmp_file_name = tmp_file.name
    tmp_file.write('Bad file!')
    tmp_file.close()
    with zipfile.ZipFile(self.archive_path, 'w') as archive:
      archive.write(tmp_file_name, '../../foo')
      self.assertRaises(
          exceptions.ArchiveError, dependency_manager_util.VerifySafeArchive,
          archive)