tests/gclient_scm_test.py

#!/usr/bin/python
# Copyright (c) 2010 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.

"""Unit tests for gclient_scm.py."""

# Import before super_mox to keep valid references.
from os import rename
from shutil import rmtree
from subprocess import Popen, PIPE, STDOUT
import tempfile
import __builtin__

# Fixes include path.
from super_mox import mox, SuperMoxBaseTestBase, SuperMoxTestBase

import gclient_scm


class GCBaseTestCase(SuperMoxTestBase):
  # Like unittest's assertRaises, but checks for Gclient.Error.
  def assertRaisesError(self, msg, fn, *args, **kwargs):
    try:
      fn(*args, **kwargs)
    except gclient_scm.gclient_utils.Error, e:
      self.assertEquals(e.args[0], msg)
    else:
      self.fail('%s not raised' % msg)


class BaseTestCase(GCBaseTestCase):
  def setUp(self):
    GCBaseTestCase.setUp(self)
    self.mox.StubOutWithMock(gclient_scm.gclient_utils, 'CheckCallAndFilter')
    self.mox.StubOutWithMock(gclient_scm.gclient_utils,
        'CheckCallAndFilterAndHeader')
    self.mox.StubOutWithMock(gclient_scm.gclient_utils, 'FileRead')
    self.mox.StubOutWithMock(gclient_scm.gclient_utils, 'FileWrite')
    self.mox.StubOutWithMock(gclient_scm.gclient_utils, 'RemoveDirectory')
    self._CaptureSVNInfo = gclient_scm.scm.SVN.CaptureInfo
    self.mox.StubOutWithMock(gclient_scm.scm.SVN, 'Capture')
    self.mox.StubOutWithMock(gclient_scm.scm.SVN, 'CaptureInfo')
    self.mox.StubOutWithMock(gclient_scm.scm.SVN, 'CaptureStatus')
    self.mox.StubOutWithMock(gclient_scm.scm.SVN, 'RunAndGetFileList')
    self._scm_wrapper = gclient_scm.CreateSCM
    gclient_scm.sys.stdout.flush = lambda: None
    gclient_scm.scm.SVN.current_version = None


class SVNWrapperTestCase(BaseTestCase):
  class OptionsObject(object):
     def __init__(self, test_case, verbose=False, revision=None):
      self.verbose = verbose
      self.revision = revision
      self.manually_grab_svn_rev = True
      self.deps_os = None
      self.force = False
      self.reset = False
      self.nohooks = False
      self.stdout = gclient_scm.sys.stdout

  def Options(self, *args, **kwargs):
    return self.OptionsObject(self, *args, **kwargs)

  def setUp(self):
    BaseTestCase.setUp(self)
    self.root_dir = self.Dir()
    self.args = self.Args()
    self.url = self.Url()
    self.relpath = 'asf'

  def testDir(self):
    members = [
        'FullUrlForRelativeUrl', 'RunCommand',
        'cleanup', 'diff', 'export', 'pack', 'relpath', 'revert',
        'revinfo', 'runhooks', 'status', 'update',
        'updatesingle', 'url',
    ]

    # If you add a member, be sure to add the relevant test!
    self.compareMembers(self._scm_wrapper('svn://a'), members)

  def testUnsupportedSCM(self):
    args = ['gopher://foo', self.root_dir, self.relpath]
    exception_msg = 'No SCM found for url gopher://foo'
    self.assertRaisesError(exception_msg, self._scm_wrapper, *args)

  def testSVNFullUrlForRelativeUrl(self):
    self.url = 'svn://a/b/c/d'

    self.mox.ReplayAll()
    scm = self._scm_wrapper(url=self.url, root_dir=self.root_dir,
                            relpath=self.relpath)
    self.assertEqual(scm.FullUrlForRelativeUrl('/crap'), 'svn://a/b/crap')

  def testGITFullUrlForRelativeUrl(self):
    self.url = 'git://a/b/c/d'

    self.mox.ReplayAll()
    scm = self._scm_wrapper(url=self.url, root_dir=self.root_dir,
                            relpath=self.relpath)
    self.assertEqual(scm.FullUrlForRelativeUrl('/crap'), 'git://a/b/c/crap')

  def testRunCommandException(self):
    options = self.Options(verbose=False)
    file_path = gclient_scm.os.path.join(self.root_dir, self.relpath, '.git')
    gclient_scm.os.path.exists(file_path).AndReturn(False)

    self.mox.ReplayAll()
    scm = self._scm_wrapper(url=self.url, root_dir=self.root_dir,
                            relpath=self.relpath)
    exception = "Unsupported argument(s): %s" % ','.join(self.args)
    self.assertRaisesError(exception, scm.RunCommand,
                           'update', options, self.args)

  def testRunCommandUnknown(self):
    # TODO(maruel): if ever used.
    pass

  def testRevertMissing(self):
    options = self.Options(verbose=True)
    base_path = gclient_scm.os.path.join(self.root_dir, self.relpath)
    gclient_scm.os.path.isdir(base_path).AndReturn(False)
    gclient_scm.scm.SVN.Capture(['--version']
        ).AndReturn('svn, version 1.5.1 (r32289)')
    # It'll to a checkout instead.
    gclient_scm.os.path.exists(gclient_scm.os.path.join(base_path, '.git')
                               ).AndReturn(False)
    print("\n_____ %s is missing, synching instead" % self.relpath)
    # Checkout.
    gclient_scm.os.path.exists(base_path).AndReturn(False)
    files_list = self.mox.CreateMockAnything()
    gclient_scm.scm.SVN.RunAndGetFileList(options.verbose,
                                          ['checkout', self.url, base_path,
                                           '--force'],
                                          cwd=self.root_dir,
                                          file_list=files_list,
                                          stdout=options.stdout)

    self.mox.ReplayAll()
    scm = self._scm_wrapper(url=self.url, root_dir=self.root_dir,
                            relpath=self.relpath)
    scm.revert(options, self.args, files_list)

  def testRevertNone(self):
    options = self.Options(verbose=True)
    base_path = gclient_scm.os.path.join(self.root_dir, self.relpath)
    gclient_scm.os.path.isdir(base_path).AndReturn(True)
    gclient_scm.scm.SVN.CaptureStatus(base_path).AndReturn([])
    gclient_scm.scm.SVN.RunAndGetFileList(options.verbose,
                                          ['update', '--revision', 'BASE'],
                                          cwd=base_path,
                                          file_list=mox.IgnoreArg(),
                                          stdout=options.stdout)

    self.mox.ReplayAll()
    scm = self._scm_wrapper(url=self.url, root_dir=self.root_dir,
                            relpath=self.relpath)
    file_list = []
    scm.revert(options, self.args, file_list)

  def testRevert2Files(self):
    options = self.Options(verbose=True)
    base_path = gclient_scm.os.path.join(self.root_dir, self.relpath)
    gclient_scm.os.path.isdir(base_path).AndReturn(True)
    items = [
      ('M      ', 'a'),
      ('A      ', 'b'),
    ]
    file_path1 = gclient_scm.os.path.join(base_path, 'a')
    file_path2 = gclient_scm.os.path.join(base_path, 'b')
    gclient_scm.scm.SVN.CaptureStatus(base_path).AndReturn(items)
    gclient_scm.os.path.exists(file_path1).AndReturn(True)
    gclient_scm.os.path.isfile(file_path1).AndReturn(True)
    gclient_scm.os.remove(file_path1)
    gclient_scm.os.path.exists(file_path2).AndReturn(True)
    gclient_scm.os.path.isfile(file_path2).AndReturn(True)
    gclient_scm.os.remove(file_path2)
    gclient_scm.scm.SVN.RunAndGetFileList(options.verbose,
                                          ['update', '--revision', 'BASE'],
                                          cwd=base_path,
                                          file_list=mox.IgnoreArg(),
                                          stdout=options.stdout)
    print(gclient_scm.os.path.join(base_path, 'a'))
    print(gclient_scm.os.path.join(base_path, 'b'))

    self.mox.ReplayAll()
    scm = self._scm_wrapper(url=self.url, root_dir=self.root_dir,
                            relpath=self.relpath)
    file_list = []
    scm.revert(options, self.args, file_list)

  def testRevertDirectory(self):
    options = self.Options(verbose=True)
    base_path = gclient_scm.os.path.join(self.root_dir, self.relpath)
    gclient_scm.os.path.isdir(base_path).AndReturn(True)
    items = [
      ('~      ', 'a'),
    ]
    gclient_scm.scm.SVN.CaptureStatus(base_path).AndReturn(items)
    file_path = gclient_scm.os.path.join(base_path, 'a')
    print(file_path)
    gclient_scm.os.path.exists(file_path).AndReturn(True)
    gclient_scm.os.path.isfile(file_path).AndReturn(False)
    gclient_scm.os.path.islink(file_path).AndReturn(False)
    gclient_scm.os.path.isdir(file_path).AndReturn(True)
    gclient_scm.gclient_utils.RemoveDirectory(file_path)
    file_list1 = []
    gclient_scm.scm.SVN.RunAndGetFileList(options.verbose,
                                          ['update', '--revision', 'BASE'],
                                          cwd=base_path,
                                          file_list=mox.IgnoreArg(),
                                          stdout=options.stdout)

    self.mox.ReplayAll()
    scm = self._scm_wrapper(url=self.url, root_dir=self.root_dir,
                            relpath=self.relpath)
    file_list2 = []
    scm.revert(options, self.args, file_list2)

  def testStatus(self):
    options = self.Options(verbose=True)
    base_path = gclient_scm.os.path.join(self.root_dir, self.relpath)
    gclient_scm.os.path.isdir(base_path).AndReturn(True)
    gclient_scm.scm.SVN.RunAndGetFileList(options.verbose,
                                          ['status'] + self.args,
                                          cwd=base_path, file_list=[],
                                          stdout=options.stdout).AndReturn(None)

    self.mox.ReplayAll()
    scm = self._scm_wrapper(url=self.url, root_dir=self.root_dir,
                            relpath=self.relpath)
    file_list = []
    self.assertEqual(scm.status(options, self.args, file_list), None)


  # TODO(maruel):  TEST REVISIONS!!!
  # TODO(maruel):  TEST RELOCATE!!!
  def testUpdateCheckout(self):
    options = self.Options(verbose=True)
    base_path = gclient_scm.os.path.join(self.root_dir, self.relpath)
    file_info = gclient_scm.gclient_utils.PrintableObject()
    file_info.root = 'blah'
    file_info.url = self.url
    file_info.uuid = 'ABC'
    file_info.revision = 42
    gclient_scm.os.path.exists(gclient_scm.os.path.join(base_path, '.git')
                               ).AndReturn(False)
    # Checkout.
    gclient_scm.os.path.exists(base_path).AndReturn(False)
    files_list = self.mox.CreateMockAnything()
    gclient_scm.scm.SVN.Capture(['--version']
        ).AndReturn('svn, version 1.5.1 (r32289)')
    gclient_scm.scm.SVN.RunAndGetFileList(options.verbose,
                                          ['checkout', self.url, base_path,
                                           '--force'],
                                          cwd=self.root_dir,
                                          file_list=files_list,
                                          stdout=options.stdout)
    self.mox.ReplayAll()
    scm = self._scm_wrapper(url=self.url, root_dir=self.root_dir,
                            relpath=self.relpath)
    scm.update(options, (), files_list)

  def testUpdateUpdate(self):
    options = self.Options(verbose=True)
    base_path = gclient_scm.os.path.join(self.root_dir, self.relpath)
    options.force = True
    options.nohooks = False
    file_info = {
      'Repository Root': 'blah',
      'URL': self.url,
      'UUID': 'ABC',
      'Revision': 42,
    }
    gclient_scm.os.path.exists(gclient_scm.os.path.join(base_path, '.git')
                               ).AndReturn(False)

    # Verify no locked files.
    gclient_scm.scm.SVN.CaptureStatus(gclient_scm.os.path.join(base_path, '.')
        ).AndReturn([])

    # Checkout or update.
    gclient_scm.os.path.exists(base_path).AndReturn(True)
    gclient_scm.scm.SVN.CaptureInfo(
        gclient_scm.os.path.join(base_path, "."), '.'
        ).AndReturn(file_info)
    # Cheat a bit here.
    gclient_scm.scm.SVN.CaptureInfo(file_info['URL'], '.').AndReturn(file_info)
    additional_args = []
    if options.manually_grab_svn_rev:
      additional_args = ['--revision', str(file_info['Revision'])]
    gclient_scm.scm.SVN.Capture(['--version']
        ).AndReturn('svn, version 1.5.1 (r32289)')
    additional_args.append('--force')
    files_list = []
    gclient_scm.scm.SVN.RunAndGetFileList(
        options.verbose,
        ['update', base_path] + additional_args,
        cwd=self.root_dir, file_list=files_list, stdout=options.stdout)

    self.mox.ReplayAll()
    scm = self._scm_wrapper(url=self.url, root_dir=self.root_dir,
                            relpath=self.relpath)
    scm.update(options, (), files_list)

  def testUpdateSingleCheckout(self):
    options = self.Options(verbose=True)
    base_path = gclient_scm.os.path.join(self.root_dir, self.relpath)
    file_info = {
      'URL': self.url,
      'Revision': 42,
    }

    # Checks to make sure that we support svn co --depth.
    gclient_scm.scm.SVN.current_version = None
    gclient_scm.scm.SVN.Capture(['--version']
        ).AndReturn('svn, version 1.5.1 (r32289)')
    gclient_scm.os.path.exists(gclient_scm.os.path.join(base_path, '.svn')
        ).AndReturn(False)
    gclient_scm.os.path.exists(gclient_scm.os.path.join(base_path, 'DEPS')
        ).AndReturn(False)

    # Verify no locked files.
    gclient_scm.scm.SVN.CaptureStatus(gclient_scm.os.path.join(base_path, '.')
        ).AndReturn([])

    # When checking out a single file, we issue an svn checkout and svn update.
    files_list = self.mox.CreateMockAnything()
    gclient_scm.gclient_utils.CheckCallAndFilterAndHeader(
        ['svn', 'checkout', '--depth', 'empty', self.url, base_path],
        always=True, cwd=self.root_dir, stdout=options.stdout)
    gclient_scm.scm.SVN.RunAndGetFileList(options.verbose, ['update', 'DEPS'],
        cwd=base_path, file_list=files_list, stdout=options.stdout)

    # Now we fall back on scm.update().
    gclient_scm.os.path.exists(gclient_scm.os.path.join(base_path, '.git')
        ).AndReturn(False)
    gclient_scm.os.path.exists(base_path).AndReturn(True)
    gclient_scm.scm.SVN.CaptureInfo(
        gclient_scm.os.path.join(base_path, "."), '.'
        ).AndReturn(file_info)
    gclient_scm.scm.SVN.CaptureInfo(file_info['URL'], '.').AndReturn(file_info)
    options.stdout.write("\n_____ %s at 42" % self.relpath)
    options.stdout.write('\n')

    self.mox.ReplayAll()
    scm = self._scm_wrapper(url=self.url, root_dir=self.root_dir,
                            relpath=self.relpath)
    scm.updatesingle(options, ['DEPS'], files_list)

  def testUpdateSingleCheckoutSVN14(self):
    options = self.Options(verbose=True)
    base_path = gclient_scm.os.path.join(self.root_dir, self.relpath)
    file_info = {
      'URL': self.url,
      'Revision': 42,
    }

    # Checks to make sure that we support svn co --depth.
    gclient_scm.scm.SVN.current_version = None
    gclient_scm.scm.SVN.Capture(['--version']
        ).AndReturn('svn, version 1.4.4 (r25188)')
    gclient_scm.os.path.exists(gclient_scm.os.path.join(base_path)
        ).AndReturn(True)

    # When checking out a single file with svn 1.4, we use svn export
    files_list = self.mox.CreateMockAnything()
    gclient_scm.gclient_utils.CheckCallAndFilterAndHeader(
        ['svn', 'export', gclient_scm.os.path.join(self.url, 'DEPS'),
            gclient_scm.os.path.join(base_path, 'DEPS')],
        always=True, cwd=self.root_dir, stdout=options.stdout)

    self.mox.ReplayAll()
    scm = self._scm_wrapper(url=self.url, root_dir=self.root_dir,
                            relpath=self.relpath)
    scm.updatesingle(options, ['DEPS'], files_list)

  def testUpdateSingleCheckoutSVNUpgrade(self):
    options = self.Options(verbose=True)
    base_path = gclient_scm.os.path.join(self.root_dir, self.relpath)
    file_info = {
      'URL': self.url,
      'Revision': 42,
    }

    # Checks to make sure that we support svn co --depth.
    gclient_scm.scm.SVN.current_version = None
    gclient_scm.scm.SVN.Capture(['--version']
        ).AndReturn('svn, version 1.5.1 (r32289)')
    gclient_scm.os.path.exists(gclient_scm.os.path.join(base_path, '.svn')
        ).AndReturn(False)
    # If DEPS already exists, assume we're upgrading from svn1.4, so delete
    # the old DEPS file.
    gclient_scm.os.path.exists(gclient_scm.os.path.join(base_path, 'DEPS')
        ).AndReturn(True)
    gclient_scm.os.remove(gclient_scm.os.path.join(base_path, 'DEPS'))

    # Verify no locked files.
    gclient_scm.scm.SVN.CaptureStatus(gclient_scm.os.path.join(base_path, '.')
        ).AndReturn([])

    # When checking out a single file, we issue an svn checkout and svn update.
    files_list = self.mox.CreateMockAnything()
    gclient_scm.gclient_utils.CheckCallAndFilterAndHeader(
        ['svn', 'checkout', '--depth', 'empty', self.url, base_path],
        always=True, cwd=self.root_dir, stdout=options.stdout)
    gclient_scm.scm.SVN.RunAndGetFileList(options.verbose, ['update', 'DEPS'],
        cwd=base_path, file_list=files_list, stdout=options.stdout)

    # Now we fall back on scm.update().
    gclient_scm.os.path.exists(gclient_scm.os.path.join(base_path, '.git')
        ).AndReturn(False)
    gclient_scm.os.path.exists(base_path).AndReturn(True)
    gclient_scm.scm.SVN.CaptureInfo(
        gclient_scm.os.path.join(base_path, "."), '.'
        ).AndReturn(file_info)
    gclient_scm.scm.SVN.CaptureInfo(file_info['URL'], '.').AndReturn(file_info)
    print("\n_____ %s at 42" % self.relpath)

    self.mox.ReplayAll()
    scm = self._scm_wrapper(url=self.url, root_dir=self.root_dir,
                            relpath=self.relpath)
    scm.updatesingle(options, ['DEPS'], files_list)

  def testUpdateSingleUpdate(self):
    options = self.Options(verbose=True)
    base_path = gclient_scm.os.path.join(self.root_dir, self.relpath)
    file_info = {
      'URL': self.url,
      'Revision': 42,
    }
    # Checks to make sure that we support svn co --depth.
    gclient_scm.scm.SVN.current_version = None
    gclient_scm.scm.SVN.Capture(['--version']
        ).AndReturn('svn, version 1.5.1 (r32289)')
    gclient_scm.os.path.exists(gclient_scm.os.path.join(base_path, '.svn')
        ).AndReturn(True)

    # Verify no locked files.
    gclient_scm.scm.SVN.CaptureStatus(gclient_scm.os.path.join(base_path, '.')
        ).AndReturn([])

    # Now we fall back on scm.update().
    files_list = self.mox.CreateMockAnything()
    gclient_scm.os.path.exists(gclient_scm.os.path.join(base_path, '.git')
                               ).AndReturn(False)
    gclient_scm.os.path.exists(base_path).AndReturn(True)
    gclient_scm.scm.SVN.CaptureInfo(
        gclient_scm.os.path.join(base_path, "."), '.'
        ).AndReturn(file_info)
    gclient_scm.scm.SVN.CaptureInfo(file_info['URL'], '.').AndReturn(file_info)
    print("\n_____ %s at 42" % self.relpath)

    self.mox.ReplayAll()
    scm = self._scm_wrapper(url=self.url, root_dir=self.root_dir,
                            relpath=self.relpath)
    scm.updatesingle(options, ['DEPS'], files_list)

  def testUpdateGit(self):
    options = self.Options(verbose=True)
    file_path = gclient_scm.os.path.join(self.root_dir, self.relpath, '.git')
    gclient_scm.os.path.exists(file_path).AndReturn(True)
    print("________ found .git directory; skipping %s" % self.relpath)

    self.mox.ReplayAll()
    scm = self._scm_wrapper(url=self.url, root_dir=self.root_dir,
                            relpath=self.relpath)
    file_list = []
    scm.update(options, self.args, file_list)


class GitWrapperTestCase(BaseTestCase):
  """This class doesn't use pymox."""
  class OptionsObject(object):
     def __init__(self, test_case, verbose=False, revision=None):
      self.verbose = verbose
      self.revision = revision
      self.manually_grab_svn_rev = True
      self.deps_os = None
      self.force = False
      self.reset = False
      self.nohooks = False

  sample_git_import = """blob
mark :1
data 6
Hello

blob
mark :2
data 4
Bye

reset refs/heads/master
commit refs/heads/master
mark :3
author Bob <bob@example.com> 1253744361 -0700
committer Bob <bob@example.com> 1253744361 -0700
data 8
A and B
M 100644 :1 a
M 100644 :2 b

blob
mark :4
data 10
Hello
You

blob
mark :5
data 8
Bye
You

commit refs/heads/origin
mark :6
author Alice <alice@example.com> 1253744424 -0700
committer Alice <alice@example.com> 1253744424 -0700
data 13
Personalized
from :3
M 100644 :4 a
M 100644 :5 b

reset refs/heads/master
from :3
"""
  def Options(self, *args, **kwargs):
    return self.OptionsObject(self, *args, **kwargs)

  def CreateGitRepo(self, git_import, path):
    """Do it for real."""
    try:
      Popen(['git', 'init'], stdout=PIPE, stderr=STDOUT,
            cwd=path).communicate()
    except OSError:
      # git is not available, skip this test.
      return False
    Popen(['git', 'fast-import'], stdin=PIPE, stdout=PIPE, stderr=STDOUT,
          cwd=path).communicate(input=git_import)
    Popen(['git', 'checkout'], stdout=PIPE, stderr=STDOUT,
          cwd=path).communicate()
    Popen(['git', 'remote', 'add', '-f', 'origin', '.'], stdout=PIPE,
          stderr=STDOUT, cwd=path).communicate()
    Popen(['git', 'checkout', '-b', 'new', 'origin/master'], stdout=PIPE,
          stderr=STDOUT, cwd=path).communicate()
    Popen(['git', 'push', 'origin', 'origin/origin:origin/master'], stdout=PIPE,
          stderr=STDOUT, cwd=path).communicate()
    Popen(['git', 'config', '--unset', 'remote.origin.fetch'], stdout=PIPE,
          stderr=STDOUT, cwd=path).communicate()
    return True

  def setUp(self):
    self.args = self.Args()
    self.url = 'git://foo'
    self.root_dir = tempfile.mkdtemp()
    self.relpath = '.'
    self.base_path = gclient_scm.os.path.join(self.root_dir, self.relpath)
    self.enabled = self.CreateGitRepo(self.sample_git_import, self.base_path)
    SuperMoxBaseTestBase.setUp(self)

  def tearDown(self):
    SuperMoxBaseTestBase.tearDown(self)
    rmtree(self.root_dir)

  def testDir(self):
    members = [
        'FullUrlForRelativeUrl', 'RunCommand',
        'cleanup', 'diff', 'export', 'pack', 'relpath', 'revert',
        'revinfo', 'runhooks', 'status', 'update', 'url',
    ]

    # If you add a member, be sure to add the relevant test!
    self.compareMembers(gclient_scm.CreateSCM(url=self.url), members)

  def testRevertMissing(self):
    if not self.enabled:
      return
    options = self.Options()
    file_path = gclient_scm.os.path.join(self.base_path, 'a')
    scm = gclient_scm.CreateSCM(url=self.url, root_dir=self.root_dir,
                                relpath=self.relpath)
    file_list = []
    scm.update(options, None, file_list)
    gclient_scm.os.remove(file_path)
    file_list = []
    scm.revert(options, self.args, file_list)
    self.assertEquals(file_list, [file_path])
    file_list = []
    scm.diff(options, self.args, file_list)
    self.assertEquals(file_list, [])

  def testRevertNone(self):
    if not self.enabled:
      return
    options = self.Options()
    scm = gclient_scm.CreateSCM(url=self.url, root_dir=self.root_dir,
                                relpath=self.relpath)
    file_list = []
    scm.update(options, None, file_list)
    file_list = []
    scm.revert(options, self.args, file_list)
    self.assertEquals(file_list, [])
    self.assertEquals(scm.revinfo(options, self.args, None),
                      'a7142dc9f0009350b96a11f372b6ea658592aa95')


  def testRevertModified(self):
    if not self.enabled:
      return
    options = self.Options()
    scm = gclient_scm.CreateSCM(url=self.url, root_dir=self.root_dir,
                                relpath=self.relpath)
    file_list = []
    scm.update(options, None, file_list)
    file_path = gclient_scm.os.path.join(self.base_path, 'a')
    open(file_path, 'a').writelines('touched\n')
    file_list = []
    scm.revert(options, self.args, file_list)
    self.assertEquals(file_list, [file_path])
    file_list = []
    scm.diff(options, self.args, file_list)
    self.assertEquals(file_list, [])
    self.assertEquals(scm.revinfo(options, self.args, None),
                      'a7142dc9f0009350b96a11f372b6ea658592aa95')

  def testRevertNew(self):
    if not self.enabled:
      return
    options = self.Options()
    scm = gclient_scm.CreateSCM(url=self.url, root_dir=self.root_dir,
                                relpath=self.relpath)
    file_list = []
    scm.update(options, None, file_list)
    file_path = gclient_scm.os.path.join(self.base_path, 'c')
    f = open(file_path, 'w')
    f.writelines('new\n')
    f.close()
    Popen(['git', 'add', 'c'], stdout=PIPE,
          stderr=STDOUT, cwd=self.base_path).communicate()
    file_list = []
    scm.revert(options, self.args, file_list)
    self.assertEquals(file_list, [file_path])
    file_list = []
    scm.diff(options, self.args, file_list)
    self.assertEquals(file_list, [])
    self.assertEquals(scm.revinfo(options, self.args, None),
                      'a7142dc9f0009350b96a11f372b6ea658592aa95')

  def testStatusNew(self):
    if not self.enabled:
      return
    options = self.Options()
    file_path = gclient_scm.os.path.join(self.base_path, 'a')
    open(file_path, 'a').writelines('touched\n')
    scm = gclient_scm.CreateSCM(url=self.url, root_dir=self.root_dir,
                                relpath=self.relpath)
    file_list = []
    scm.status(options, self.args, file_list)
    self.assertEquals(file_list, [file_path])

  def testStatus2New(self):
    if not self.enabled:
      return
    options = self.Options()
    expected_file_list = []
    for f in ['a', 'b']:
        file_path = gclient_scm.os.path.join(self.base_path, f)
        open(file_path, 'a').writelines('touched\n')
        expected_file_list.extend([file_path])
    scm = gclient_scm.CreateSCM(url=self.url, root_dir=self.root_dir,
                                relpath=self.relpath)
    file_list = []
    scm.status(options, self.args, file_list)
    expected_file_list = [gclient_scm.os.path.join(self.base_path, x)
                          for x in ['a', 'b']]
    self.assertEquals(sorted(file_list), expected_file_list)

  def testUpdateCheckout(self):
    if not self.enabled:
      return
    options = self.Options(verbose=True)
    root_dir = tempfile.mkdtemp()
    relpath = 'foo'
    base_path = gclient_scm.os.path.join(root_dir, relpath)
    url = gclient_scm.os.path.join(self.root_dir, self.relpath, '.git')
    try:
      scm = gclient_scm.CreateSCM(url=url, root_dir=root_dir,
                                  relpath=relpath)
      file_list = []
      scm.update(options, (), file_list)
      self.assertEquals(len(file_list), 2)
      self.assert_(gclient_scm.os.path.isfile(
          gclient_scm.os.path.join(base_path, 'a')))
      self.assertEquals(scm.revinfo(options, (), None),
                        '069c602044c5388d2d15c3f875b057c852003458')
    finally:
      rmtree(root_dir)

  def testUpdateUpdate(self):
    if not self.enabled:
      return
    options = self.Options()
    expected_file_list = [gclient_scm.os.path.join(self.base_path, x)
                          for x in ['a', 'b']]
    scm = gclient_scm.CreateSCM(url=self.url, root_dir=self.root_dir,
                                relpath=self.relpath)
    file_list = []
    scm.update(options, (), file_list)
    self.assertEquals(file_list, expected_file_list)
    self.assertEquals(scm.revinfo(options, (), None),
                      'a7142dc9f0009350b96a11f372b6ea658592aa95')

  def testUpdateUnstagedConflict(self):
    if not self.enabled:
      return
    options = self.Options()
    scm = gclient_scm.CreateSCM(url=self.url, root_dir=self.root_dir,
                                relpath=self.relpath)
    file_path = gclient_scm.os.path.join(self.base_path, 'b')
    f = open(file_path, 'w').writelines('conflict\n')
    exception = (
        "error: Your local changes to 'b' would be overwritten by merge.  "
        "Aborting.\n"
        "Please, commit your changes or stash them before you can merge.\n")
    self.assertRaisesError(exception, scm.update, options, (), [])

  def testUpdateConflict(self):
    if not self.enabled:
      return
    options = self.Options()
    scm = gclient_scm.CreateSCM(url=self.url, root_dir=self.root_dir,
                                relpath=self.relpath)
    file_path = gclient_scm.os.path.join(self.base_path, 'b')
    f = open(file_path, 'w').writelines('conflict\n')
    scm._Run(['commit', '-am', 'test'])
    self.mox.StubOutWithMock(__builtin__, 'raw_input')
    __builtin__.raw_input.__call__(mox.StrContains('Cannot fast-forward merge, '
                                                   'attempt to rebase? (y)es / '
                                                   '(q)uit / (s)kip : ')
                                   ).AndReturn('y')
    self.mox.ReplayAll()
    exception = \
        'Conflict while rebasing this branch.\n' \
        'Fix the conflict and run gclient again.\n' \
        "See 'man git-rebase' for details.\n"
    self.assertRaisesError(exception, scm.update, options, (), [])
    exception = \
        '\n____ . at refs/heads/master\n' \
        '\tYou have unstaged changes.\n' \
        '\tPlease commit, stash, or reset.\n'
    self.assertRaisesError(exception, scm.update, options, (), [])

  def testUpdateNotGit(self):
    if not self.enabled:
      return
    options = self.Options()
    scm = gclient_scm.CreateSCM(url=self.url, root_dir=self.root_dir,
                                relpath=self.relpath)
    git_path = gclient_scm.os.path.join(self.base_path, '.git')
    rename(git_path, git_path + 'foo')
    exception = \
        '\n____ . at refs/heads/master\n' \
        '\tPath is not a git repo. No .git dir.\n' \
        '\tTo resolve:\n' \
        '\t\trm -rf .\n' \
        '\tAnd run gclient sync again\n'
    self.assertRaisesError(exception, scm.update, options, (), [])

  def testRevinfo(self):
    if not self.enabled:
      return
    options = self.Options()
    scm = gclient_scm.CreateSCM(url=self.url, root_dir=self.root_dir,
                                relpath=self.relpath)
    rev_info = scm.revinfo(options, (), None)
    self.assertEquals(rev_info, '069c602044c5388d2d15c3f875b057c852003458')


if __name__ == '__main__':
  import unittest
  unittest.main()

# vim: ts=2:sw=2:tw=80:et: