#!/usr/bin/python
#
# Copyright (c) 2012 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 the gs.py module."""

import os
import sys
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.dirname(
    os.path.abspath(__file__)))))

import mox
import unittest

from chromite.lib import cros_build_lib
from chromite.lib import cros_test_lib
from chromite.lib import gs
from chromite.lib import osutils


#pylint: disable=E1101,W0212
class GSContextTest(cros_test_lib.TempDirMixin, mox.MoxTestBase):
  """Tests for GSContext()"""

  _GSResponsePreconditionFailed = """
[Setting Content-Type=text/x-python]
        GSResponseError:: status=412, code=PreconditionFailed,
        reason=Precondition Failed."""

  def setUp(self):
    cros_test_lib.TempDirMixin.setUp(self)
    for attr in ('gsutil_bin', 'boto_file', 'acl_file'):
      path = os.path.join(self.tempdir, attr)
      osutils.WriteFile(path, '')
      setattr(self, attr, path)
    self.defaults = {'DEFAULT_BOTO_FILE': self.boto_file,
                     'DEFAULT_GSUTIL_BIN': self.gsutil_bin}
    self.bad_path = os.path.join(self.tempdir, 'nonexistent')

    self._boto_config_env = os.environ.pop("BOTO_CONFIG", None)
    mox.MoxTestBase.setUp(self)

    # No command should be ran w/out us explicitly asseting it.
    self.mox.StubOutWithMock(cros_build_lib, 'RunCommand')

  def tearDown(self):
    mox.MoxTestBase.tearDown(self)
    cros_test_lib.TempDirMixin.tearDown(self)
    if self._boto_config_env is not None:
      os.environ['BOTO_CONFIG'] = self._boto_config_env

  def MkContext(self, *args, **kwds):
    # Note we derive on the fly here so we can ensure that
    # there always is a valid fallback for boto_file; we
    # do so in a way that allows us to also check that
    # behaviour when the default is missing.

    class Context(gs.GSContext):
      # This bit of voodoo is basically a way to scrape
      # default overrides out of keywords into the class
      # scope; used for testing the defaults.
      # First inject in our defaults, then the overrides.
      locals().update(self.defaults)
      locals().update((key, kwds.pop(key)) for key in list(kwds)
                      if key.startswith("DEFAULT_"))

    suppress_DoCommand = kwds.pop('suppress_DoCommand', True)
    context = Context(*args, **kwds)
    if suppress_DoCommand:
      self.mox.StubOutWithMock(context, '_DoCommand')
    return context

  def testInitGsutilBin(self):
    """Test we use the given gsutil binary, erroring where appropriate."""

    self.assertEqual(self.MkContext().gsutil_bin, self.gsutil_bin)
    self.assertRaises(gs.GSContextException, self.MkContext,
                      DEFAULT_BOTO_FILE=self.bad_path)
    self.assertEqual(
        self.MkContext(gsutil_bin=self.boto_file).gsutil_bin,
        self.boto_file)
    self.assertRaises(gs.GSContextException, self.MkContext,
                      gsutil_bin=self.bad_path)

  def testInitBotoFile(self):
    """Test we use the given boto_file, erroring where appropriate."""

    self.assertEqual(self.MkContext().boto_file, self.boto_file)
    self.assertRaises(gs.GSContextException, self.MkContext,
                      DEFAULT_BOTO_FILE=self.bad_path)
    # Check env usage next; no need to cleanup, teardown handles it,
    # and we want the env var to persist for the next part of this test.
    os.environ['BOTO_CONFIG'] = self.bad_path
    self.assertRaises(gs.GSContextException, self.MkContext)
    os.environ['BOTO_CONFIG'] = self.gsutil_bin
    self.assertEqual(self.MkContext().boto_file, self.gsutil_bin)

    self.assertEqual(self.MkContext(boto_file=self.acl_file).boto_file,
                     self.acl_file)
    self.assertRaises(gs.GSContextException, self.MkContext,
                      boto_file=self.bad_path)

  def testInitAclFile(self):
    self.assertEqual(self.MkContext().acl_file, None)
    self.assertEqual(self.MkContext(acl_file=self.acl_file).acl_file,
                     self.acl_file)
    self.assertRaises(gs.GSContextException, self.MkContext,
                      acl_file=self.bad_path)

  def assertGSCommand(self, context, cmd, **kwds):
    return context._DoCommand(cmd, **kwds)

  def testCopy(self, functor=None, filename=None):
    """Test CopyTo functionality."""

    local_path = lambda  x: '/tmp/file%s' % x
    if filename is not None:
      given_remote = lambda x: 'gs://test/path/%s' % x
      expected_remote = lambda x: 'gs://test/path/%s/%s' % (x, filename)
    else:
      given_remote = expected_remote = lambda x: 'gs://test/path/file%s' % x

    ctx1 = self.MkContext()
    self.assertGSCommand(ctx1, ['cp', '--', local_path(1), expected_remote(1)])
    ctx2 = self.MkContext(acl_file=self.acl_file)
    self.assertGSCommand(ctx2, ['cp', '-a', self.acl_file, '--',
                                local_path(2), expected_remote(2)])
    ctx3 = self.MkContext()
    self.assertGSCommand(ctx3, ['cp', '-a', self.gsutil_bin, '--',
                                local_path('brass'),
                                expected_remote('brass')])
    ctx4 = self.MkContext(acl_file=self.gsutil_bin)
    self.assertGSCommand(ctx4, ['cp', '-a', self.acl_file, '--',
                                local_path('monkey'),
                                expected_remote('monkey')])
    ctx5 = self.MkContext()
    self.assertGSCommand(ctx5, ['cp', '--', local_path(5),
                                expected_remote(5)],
                         headers=['x-goog-if-sequence-number-match:0'])
    ctx6 = self.MkContext()
    self.assertGSCommand(ctx6, ['cp', '--', self.gsutil_bin,
                                expected_remote(6)],
                         headers=['x-goog-if-sequence-number-match:6'])
    ctx7 = self.MkContext()
    ret = self.assertGSCommand(ctx7, ['cp', '--', local_path(7),
                                      expected_remote(7)])
    ret.AndRaise(cros_build_lib.RunCommandError(
        'blah', cros_build_lib.CommandResult(returncode=1, output='')))
    ctx8 = self.MkContext()
    ret = self.assertGSCommand(ctx8, ['cp', '--', local_path(8),
                                      expected_remote(8)])
    ret.AndRaise(cros_build_lib.RunCommandError(
        'blah', cros_build_lib.CommandResult(returncode=1,
                                             output='code=PreconditionFailed')))

    self.mox.ReplayAll()
    if functor is None:
      functor = lambda ctx:ctx.Copy
    functor(ctx1)(local_path(1), given_remote(1))
    functor(ctx2)(local_path(2), given_remote(2))
    functor(ctx3)(local_path('brass'), given_remote('brass'),
                  acl=self.gsutil_bin)
    functor(ctx4)(local_path('monkey'), given_remote('monkey'),
                  acl=self.acl_file)
    functor(ctx5)(local_path(5), given_remote(5), version=0)
    functor(ctx6)(self.gsutil_bin, given_remote(6), version=6)
    self.assertRaises(cros_build_lib.RunCommandError,
                      functor(ctx7), local_path(7), given_remote(7))
    self.assertRaises(gs.GSContextException,
                      functor(ctx8), local_path(8), given_remote(8))
    self.mox.VerifyAll()

  def testCopyInto(self):
    def functor(ctx):
      return lambda *args, **kwds:ctx.CopyInto(*args, filename='blah', **kwds)
    self.testCopy(functor=functor, filename='blah')

  def testDoCommand(self):
    """Verify the internal DoCommand function works correctly."""
    self.mox.StubOutWithMock(cros_build_lib, 'RetryCommand')
    ctx1 = self.MkContext(suppress_DoCommand=False)
    cros_build_lib.RetryCommand(
       cros_build_lib.RunCommandCaptureOutput, ctx1._retries,
       [self.gsutil_bin, 'cp', '--', '/blah', 'gs://foon'],
       sleep=ctx1._sleep_time, extra_env={'BOTO_CONFIG':self.boto_file})
    ctx2 = self.MkContext(suppress_DoCommand=False, retries=2, sleep=1)
    cros_build_lib.RetryCommand(
       cros_build_lib.RunCommandCaptureOutput, ctx2._retries,
       [self.gsutil_bin, '-h',  'x-goog-if-sequence-number-match:1',
        'cp', '--', '/blah', 'gs://foon'],
       sleep=ctx2._sleep_time, extra_env={'BOTO_CONFIG':self.boto_file})

    self.mox.ReplayAll()
    ctx1.Copy('/blah', 'gs://foon')
    ctx2.Copy('/blah', 'gs://foon', version=1)
    self.mox.VerifyAll()

  def testSetAcl(self):
    ctx1 = self.MkContext()
    self.assertGSCommand(ctx1, ['setacl', 'monkeys', 'gs://abc/1'])
    ctx2 = self.MkContext(acl_file=self.acl_file)
    self.assertGSCommand(ctx2, ['setacl', self.acl_file, 'gs://abc/2'])
    # Ensure it blows up if there isn't an acl specified in any fashion.
    ctx3 = self.MkContext()

    self.mox.ReplayAll()
    ctx1.SetACL('gs://abc/1', 'monkeys')
    ctx2.SetACL('gs://abc/2')
    self.assertRaises(gs.GSContextException, ctx3.SetACL, 'gs://abc/3')
    self.mox.VerifyAll()

if __name__ == '__main__':
  cros_build_lib.SetupBasicLogging()
  unittest.main()