# -*- coding: utf-8 -*-
# Copyright 2021 Google Inc. All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
"""Tests for stet_util.py."""

from __future__ import absolute_import
from __future__ import print_function
from __future__ import division
from __future__ import unicode_literals

import os
import shutil

from gslib import storage_url
from gslib.tests import testcase
from gslib.tests import util
from gslib.tests.util import unittest
from gslib.utils import execution_util
from gslib.utils import stet_util

import mock


class TestStetUtil(testcase.GsUtilUnitTestCase):
  """Test STET utils."""

  @mock.patch.object(execution_util, 'ExecuteExternalCommand')
  def test_stet_upload_uses_binary_and_config_from_boto(
      self, mock_execute_external_command):
    fake_config_path = self.CreateTempFile()
    mock_execute_external_command.return_value = ('stdout', 'stderr')
    mock_logger = mock.Mock()

    source_url = storage_url.StorageUrlFromString('in')
    destination_url = storage_url.StorageUrlFromString('gs://bucket/obj')
    with util.SetBotoConfigForTest([
        ('GSUtil', 'stet_binary_path', 'fake_binary_path'),
        ('GSUtil', 'stet_config_path', fake_config_path),
    ]):
      out_file_url = stet_util.encrypt_upload(source_url, destination_url,
                                              mock_logger)

    self.assertEqual(out_file_url,
                     storage_url.StorageUrlFromString('in_.stet_tmp'))
    mock_execute_external_command.assert_called_once_with([
        'fake_binary_path',
        'encrypt',
        '--config-file={}'.format(fake_config_path),
        '--blob-id=gs://bucket/obj',
        'in',
        'in_.stet_tmp',
    ])
    mock_logger.debug.assert_called_once_with('stderr')

  @mock.patch.object(execution_util, 'ExecuteExternalCommand')
  def test_stet_upload_runs_with_binary_from_path_with_correct_settings(
      self, mock_execute_external_command):
    fake_config_path = self.CreateTempFile()
    temporary_path_directory = self.CreateTempDir()
    fake_stet_binary_path = self.CreateTempFile(tmpdir=temporary_path_directory,
                                                file_name='stet')
    previous_path = os.getenv('PATH')
    os.environ['PATH'] += os.path.pathsep + temporary_path_directory

    mock_execute_external_command.return_value = ('stdout', 'stderr')
    mock_logger = mock.Mock()

    source_url = storage_url.StorageUrlFromString('in')
    destination_url = storage_url.StorageUrlFromString('gs://bucket/obj')
    with util.SetBotoConfigForTest([
        ('GSUtil', 'stet_binary_path', None),
        ('GSUtil', 'stet_config_path', fake_config_path),
    ]):
      out_file_url = stet_util.encrypt_upload(source_url, destination_url,
                                              mock_logger)

    self.assertEqual(out_file_url,
                     storage_url.StorageUrlFromString('in_.stet_tmp'))
    mock_execute_external_command.assert_called_once_with([
        fake_stet_binary_path,
        'encrypt',
        '--config-file={}'.format(fake_config_path),
        '--blob-id=gs://bucket/obj',
        'in',
        'in_.stet_tmp',
    ])
    mock_logger.debug.assert_called_once_with('stderr')

    os.environ['PATH'] = previous_path

  @mock.patch.object(execution_util, 'ExecuteExternalCommand')
  def test_stet_upload_uses_config_from_default_path_with_correct_settings(
      self, mock_execute_external_command):
    mock_execute_external_command.return_value = ('stdout', 'stderr')
    mock_logger = mock.Mock()

    source_url = storage_url.StorageUrlFromString('in')
    destination_url = storage_url.StorageUrlFromString('gs://bucket/obj')
    with util.SetBotoConfigForTest([
        ('GSUtil', 'stet_binary_path', 'fake_binary_path'),
        ('GSUtil', 'stet_config_path', None),
    ]):
      with mock.patch.object(os.path,
                             'exists',
                             new=mock.Mock(return_value=True)):
        out_file_url = stet_util.encrypt_upload(source_url, destination_url,
                                                mock_logger)

    self.assertEqual(out_file_url,
                     storage_url.StorageUrlFromString('in_.stet_tmp'))
    mock_execute_external_command.assert_called_once_with([
        'fake_binary_path',
        'encrypt',
        '--config-file={}'.format(
            os.path.expanduser(stet_util.DEFAULT_STET_CONFIG_PATH)),
        '--blob-id=gs://bucket/obj',
        'in',
        'in_.stet_tmp',
    ])
    mock_logger.debug.assert_called_once_with('stderr')

  @mock.patch.object(shutil, 'move')
  @mock.patch.object(execution_util, 'ExecuteExternalCommand')
  def test_stet_download_runs_binary_and_replaces_temp_file(
      self, mock_execute_external_command, mock_move):
    fake_config_path = self.CreateTempFile()
    mock_execute_external_command.return_value = ('stdout', 'stderr')
    mock_logger = mock.Mock()

    source_url = storage_url.StorageUrlFromString('gs://bucket/obj')
    destination_url = storage_url.StorageUrlFromString('out')
    with util.SetBotoConfigForTest([
        ('GSUtil', 'stet_binary_path', 'fake_binary_path'),
        ('GSUtil', 'stet_config_path', fake_config_path),
    ]):
      stet_util.decrypt_download(source_url, destination_url, mock_logger)

    mock_execute_external_command.assert_called_once_with([
        'fake_binary_path', 'decrypt',
        '--config-file={}'.format(fake_config_path),
        '--blob-id=gs://bucket/obj', 'out', 'out_.stet_tmp'
    ])
    mock_logger.debug.assert_called_once_with('stderr')
    mock_move.assert_called_once_with('out_.stet_tmp', 'out')

  @mock.patch.object(stet_util,
                     '_get_stet_binary_from_path',
                     new=mock.Mock(return_value=None))
  def test_stet_util_errors_if_no_binary(self):
    fake_config_path = self.CreateTempFile()
    source_url = storage_url.StorageUrlFromString('in')
    destination_url = storage_url.StorageUrlFromString('gs://bucket/obj')
    with util.SetBotoConfigForTest([
        ('GSUtil', 'stet_binary_path', None),
        ('GSUtil', 'stet_config_path', fake_config_path),
    ]):
      with self.assertRaises(KeyError):
        stet_util.encrypt_upload(source_url, destination_url, None)

  def test_stet_util_errors_if_no_config(self):
    source_url = storage_url.StorageUrlFromString('in')
    destination_url = storage_url.StorageUrlFromString('gs://bucket/obj')
    with util.SetBotoConfigForTest([
        ('GSUtil', 'stet_binary_path', 'fake_binary_path'),
        ('GSUtil', 'stet_config_path', None),
    ]):
      with mock.patch.object(os.path,
                             'exists',
                             new=mock.Mock(return_value=False)):
        with self.assertRaises(KeyError):
          stet_util.encrypt_upload(source_url, destination_url, None)

  @mock.patch.object(os.path, 'expanduser', autospec=True)
  @mock.patch.object(execution_util,
                     'ExecuteExternalCommand',
                     new=mock.Mock(return_value=('stdout', 'stderr')))
  def test_stet_util_expands_home_directory_symbol(self, mock_expanduser):
    fake_config_path = self.CreateTempFile()
    source_url = storage_url.StorageUrlFromString('in')
    destination_url = storage_url.StorageUrlFromString('gs://bucket/obj')
    with util.SetBotoConfigForTest([
        ('GSUtil', 'stet_binary_path', 'fake_binary_path'),
        ('GSUtil', 'stet_config_path', fake_config_path),
    ]):
      stet_util.encrypt_upload(source_url, destination_url, mock.Mock())
    mock_expanduser.assert_has_calls(
        [mock.call('fake_binary_path'),
         mock.call(fake_config_path)])

[((1122, 1181), 'mock.patch.object', 'mock.patch.object', (['execution_util', '"""ExecuteExternalCommand"""'], {}), "(execution_util, 'ExecuteExternalCommand')\n", (1139, 1181), False, 'import mock\n'), ((2265, 2324), 'mock.patch.object', 'mock.patch.object', (['execution_util', '"""ExecuteExternalCommand"""'], {}), "(execution_util, 'ExecuteExternalCommand')\n", (2282, 2324), False, 'import mock\n'), ((3760, 3819), 'mock.patch.object', 'mock.patch.object', (['execution_util', '"""ExecuteExternalCommand"""'], {}), "(execution_util, 'ExecuteExternalCommand')\n", (3777, 3819), False, 'import mock\n'), ((5061, 5094), 'mock.patch.object', 'mock.patch.object', (['shutil', '"""move"""'], {}), "(shutil, 'move')\n", (5078, 5094), False, 'import mock\n'), ((5098, 5157), 'mock.patch.object', 'mock.patch.object', (['execution_util', '"""ExecuteExternalCommand"""'], {}), "(execution_util, 'ExecuteExternalCommand')\n", (5115, 5157), False, 'import mock\n'), ((7334, 7389), 'mock.patch.object', 'mock.patch.object', (['os.path', '"""expanduser"""'], {'autospec': '(True)'}), "(os.path, 'expanduser', autospec=True)\n", (7351, 7389), False, 'import mock\n'), ((1416, 1427), 'mock.Mock', 'mock.Mock', ([], {}), '()\n', (1425, 1427), False, 'import mock\n'), ((1446, 1484), 'gslib.storage_url.StorageUrlFromString', 'storage_url.StorageUrlFromString', (['"""in"""'], {}), "('in')\n", (1478, 1484), False, 'from gslib import storage_url\n'), ((1507, 1558), 'gslib.storage_url.StorageUrlFromString', 'storage_url.StorageUrlFromString', (['"""gs://bucket/obj"""'], {}), "('gs://bucket/obj')\n", (1539, 1558), False, 'from gslib import storage_url\n'), ((2706, 2723), 'os.getenv', 'os.getenv', (['"""PATH"""'], {}), "('PATH')\n", (2715, 2723), False, 'import os\n'), ((2882, 2893), 'mock.Mock', 'mock.Mock', ([], {}), '()\n', (2891, 2893), False, 'import mock\n'), ((2912, 2950), 'gslib.storage_url.StorageUrlFromString', 'storage_url.StorageUrlFromString', (['"""in"""'], {}), "('in')\n", (2944, 2950), False, 'from gslib import storage_url\n'), ((2973, 3024), 'gslib.storage_url.StorageUrlFromString', 'storage_url.StorageUrlFromString', (['"""gs://bucket/obj"""'], {}), "('gs://bucket/obj')\n", (3005, 3024), False, 'from gslib import storage_url\n'), ((4028, 4039), 'mock.Mock', 'mock.Mock', ([], {}), '()\n', (4037, 4039), False, 'import mock\n'), ((4058, 4096), 'gslib.storage_url.StorageUrlFromString', 'storage_url.StorageUrlFromString', (['"""in"""'], {}), "('in')\n", (4090, 4096), False, 'from gslib import storage_url\n'), ((4119, 4170), 'gslib.storage_url.StorageUrlFromString', 'storage_url.StorageUrlFromString', (['"""gs://bucket/obj"""'], {}), "('gs://bucket/obj')\n", (4151, 4170), False, 'from gslib import storage_url\n'), ((5407, 5418), 'mock.Mock', 'mock.Mock', ([], {}), '()\n', (5416, 5418), False, 'import mock\n'), ((5437, 5488), 'gslib.storage_url.StorageUrlFromString', 'storage_url.StorageUrlFromString', (['"""gs://bucket/obj"""'], {}), "('gs://bucket/obj')\n", (5469, 5488), False, 'from gslib import storage_url\n'), ((5511, 5550), 'gslib.storage_url.StorageUrlFromString', 'storage_url.StorageUrlFromString', (['"""out"""'], {}), "('out')\n", (5543, 5550), False, 'from gslib import storage_url\n'), ((6376, 6414), 'gslib.storage_url.StorageUrlFromString', 'storage_url.StorageUrlFromString', (['"""in"""'], {}), "('in')\n", (6408, 6414), False, 'from gslib import storage_url\n'), ((6437, 6488), 'gslib.storage_url.StorageUrlFromString', 'storage_url.StorageUrlFromString', (['"""gs://bucket/obj"""'], {}), "('gs://bucket/obj')\n", (6469, 6488), False, 'from gslib import storage_url\n'), ((6812, 6850), 'gslib.storage_url.StorageUrlFromString', 'storage_url.StorageUrlFromString', (['"""in"""'], {}), "('in')\n", (6844, 6850), False, 'from gslib import storage_url\n'), ((6873, 6924), 'gslib.storage_url.StorageUrlFromString', 'storage_url.StorageUrlFromString', (['"""gs://bucket/obj"""'], {}), "('gs://bucket/obj')\n", (6905, 6924), False, 'from gslib import storage_url\n'), ((7682, 7720), 'gslib.storage_url.StorageUrlFromString', 'storage_url.StorageUrlFromString', (['"""in"""'], {}), "('in')\n", (7714, 7720), False, 'from gslib import storage_url\n'), ((7743, 7794), 'gslib.storage_url.StorageUrlFromString', 'storage_url.StorageUrlFromString', (['"""gs://bucket/obj"""'], {}), "('gs://bucket/obj')\n", (7775, 7794), False, 'from gslib import storage_url\n'), ((1568, 1701), 'gslib.tests.util.SetBotoConfigForTest', 'util.SetBotoConfigForTest', (["[('GSUtil', 'stet_binary_path', 'fake_binary_path'), ('GSUtil',\n    'stet_config_path', fake_config_path)]"], {}), "([('GSUtil', 'stet_binary_path',\n    'fake_binary_path'), ('GSUtil', 'stet_config_path', fake_config_path)])\n", (1593, 1701), False, 'from gslib.tests import util\n'), ((1743, 1809), 'gslib.utils.stet_util.encrypt_upload', 'stet_util.encrypt_upload', (['source_url', 'destination_url', 'mock_logger'], {}), '(source_url, destination_url, mock_logger)\n', (1767, 1809), False, 'from gslib.utils import stet_util\n'), ((1913, 1961), 'gslib.storage_url.StorageUrlFromString', 'storage_url.StorageUrlFromString', (['"""in_.stet_tmp"""'], {}), "('in_.stet_tmp')\n", (1945, 1961), False, 'from gslib import storage_url\n'), ((3034, 3153), 'gslib.tests.util.SetBotoConfigForTest', 'util.SetBotoConfigForTest', (["[('GSUtil', 'stet_binary_path', None), ('GSUtil', 'stet_config_path',\n    fake_config_path)]"], {}), "([('GSUtil', 'stet_binary_path', None), ('GSUtil',\n    'stet_config_path', fake_config_path)])\n", (3059, 3153), False, 'from gslib.tests import util\n'), ((3195, 3261), 'gslib.utils.stet_util.encrypt_upload', 'stet_util.encrypt_upload', (['source_url', 'destination_url', 'mock_logger'], {}), '(source_url, destination_url, mock_logger)\n', (3219, 3261), False, 'from gslib.utils import stet_util\n'), ((3365, 3413), 'gslib.storage_url.StorageUrlFromString', 'storage_url.StorageUrlFromString', (['"""in_.stet_tmp"""'], {}), "('in_.stet_tmp')\n", (3397, 3413), False, 'from gslib import storage_url\n'), ((4180, 4301), 'gslib.tests.util.SetBotoConfigForTest', 'util.SetBotoConfigForTest', (["[('GSUtil', 'stet_binary_path', 'fake_binary_path'), ('GSUtil',\n    'stet_config_path', None)]"], {}), "([('GSUtil', 'stet_binary_path',\n    'fake_binary_path'), ('GSUtil', 'stet_config_path', None)])\n", (4205, 4301), False, 'from gslib.tests import util\n'), ((4658, 4706), 'gslib.storage_url.StorageUrlFromString', 'storage_url.StorageUrlFromString', (['"""in_.stet_tmp"""'], {}), "('in_.stet_tmp')\n", (4690, 4706), False, 'from gslib import storage_url\n'), ((5560, 5693), 'gslib.tests.util.SetBotoConfigForTest', 'util.SetBotoConfigForTest', (["[('GSUtil', 'stet_binary_path', 'fake_binary_path'), ('GSUtil',\n    'stet_config_path', fake_config_path)]"], {}), "([('GSUtil', 'stet_binary_path',\n    'fake_binary_path'), ('GSUtil', 'stet_config_path', fake_config_path)])\n", (5585, 5693), False, 'from gslib.tests import util\n'), ((5720, 5788), 'gslib.utils.stet_util.decrypt_download', 'stet_util.decrypt_download', (['source_url', 'destination_url', 'mock_logger'], {}), '(source_url, destination_url, mock_logger)\n', (5746, 5788), False, 'from gslib.utils import stet_util\n'), ((6498, 6617), 'gslib.tests.util.SetBotoConfigForTest', 'util.SetBotoConfigForTest', (["[('GSUtil', 'stet_binary_path', None), ('GSUtil', 'stet_config_path',\n    fake_config_path)]"], {}), "([('GSUtil', 'stet_binary_path', None), ('GSUtil',\n    'stet_config_path', fake_config_path)])\n", (6523, 6617), False, 'from gslib.tests import util\n'), ((6236, 6264), 'mock.Mock', 'mock.Mock', ([], {'return_value': 'None'}), '(return_value=None)\n', (6245, 6264), False, 'import mock\n'), ((6934, 7055), 'gslib.tests.util.SetBotoConfigForTest', 'util.SetBotoConfigForTest', (["[('GSUtil', 'stet_binary_path', 'fake_binary_path'), ('GSUtil',\n    'stet_config_path', None)]"], {}), "([('GSUtil', 'stet_binary_path',\n    'fake_binary_path'), ('GSUtil', 'stet_config_path', None)])\n", (6959, 7055), False, 'from gslib.tests import util\n'), ((7804, 7937), 'gslib.tests.util.SetBotoConfigForTest', 'util.SetBotoConfigForTest', (["[('GSUtil', 'stet_binary_path', 'fake_binary_path'), ('GSUtil',\n    'stet_config_path', fake_config_path)]"], {}), "([('GSUtil', 'stet_binary_path',\n    'fake_binary_path'), ('GSUtil', 'stet_config_path', fake_config_path)])\n", (7829, 7937), False, 'from gslib.tests import util\n'), ((7499, 7543), 'mock.Mock', 'mock.Mock', ([], {'return_value': "('stdout', 'stderr')"}), "(return_value=('stdout', 'stderr'))\n", (7508, 7543), False, 'import mock\n'), ((4486, 4552), 'gslib.utils.stet_util.encrypt_upload', 'stet_util.encrypt_upload', (['source_url', 'destination_url', 'mock_logger'], {}), '(source_url, destination_url, mock_logger)\n', (4510, 4552), False, 'from gslib.utils import stet_util\n'), ((6686, 6745), 'gslib.utils.stet_util.encrypt_upload', 'stet_util.encrypt_upload', (['source_url', 'destination_url', 'None'], {}), '(source_url, destination_url, None)\n', (6710, 6745), False, 'from gslib.utils import stet_util\n'), ((8018, 8029), 'mock.Mock', 'mock.Mock', ([], {}), '()\n', (8027, 8029), False, 'import mock\n'), ((8078, 8107), 'mock.call', 'mock.call', (['"""fake_binary_path"""'], {}), "('fake_binary_path')\n", (8087, 8107), False, 'import mock\n'), ((8118, 8145), 'mock.call', 'mock.call', (['fake_config_path'], {}), '(fake_config_path)\n', (8127, 8145), False, 'import mock\n'), ((4862, 4916), 'os.path.expanduser', 'os.path.expanduser', (['stet_util.DEFAULT_STET_CONFIG_PATH'], {}), '(stet_util.DEFAULT_STET_CONFIG_PATH)\n', (4880, 4916), False, 'import os\n'), ((7270, 7329), 'gslib.utils.stet_util.encrypt_upload', 'stet_util.encrypt_upload', (['source_url', 'destination_url', 'None'], {}), '(source_url, destination_url, None)\n', (7294, 7329), False, 'from gslib.utils import stet_util\n'), ((4432, 4460), 'mock.Mock', 'mock.Mock', ([], {'return_value': '(True)'}), '(return_value=True)\n', (4441, 4460), False, 'import mock\n'), ((7186, 7215), 'mock.Mock', 'mock.Mock', ([], {'return_value': '(False)'}), '(return_value=False)\n', (7195, 7215), False, 'import mock\n')]