depot_tools/tests/git_cache_test.py

#!/usr/bin/env vpython3
# 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.

"""Unit tests for git_cache.py"""

from io import StringIO
import logging
import os
import shutil
import subprocess
import sys
import tempfile
import unittest
from unittest import mock

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

from testing_support import coverage_utils
import git_cache

class GitCacheTest(unittest.TestCase):
  def setUp(self):
    self.cache_dir = tempfile.mkdtemp(prefix='git_cache_test_')
    self.addCleanup(shutil.rmtree, self.cache_dir, ignore_errors=True)
    self.origin_dir = tempfile.mkdtemp(suffix='origin.git')
    self.addCleanup(shutil.rmtree, self.origin_dir, ignore_errors=True)
    git_cache.Mirror.SetCachePath(self.cache_dir)

    # Ensure git_cache works with safe.bareRepository.
    mock.patch.dict(
        'os.environ', {
            'GIT_CONFIG_GLOBAL': os.path.join(self.cache_dir, '.gitconfig'),
        }).start()
    self.addCleanup(mock.patch.stopall)
    self.git([
        'config', '--file',
        os.path.join(self.cache_dir, '.gitconfig'), '--add',
        'safe.bareRepository', 'explicit'
    ])

  def git(self, cmd, cwd=None):
    cwd = cwd or self.origin_dir
    git = 'git.bat' if sys.platform == 'win32' else 'git'
    subprocess.check_call([git] + cmd, cwd=cwd)

  def testParseFetchSpec(self):
    testData = [
        ([], []),
        (['main'], [('+refs/heads/main:refs/heads/main',
                       r'\+refs/heads/main:.*')]),
        (['main/'], [('+refs/heads/main:refs/heads/main',
                       r'\+refs/heads/main:.*')]),
        (['+main'], [('+refs/heads/main:refs/heads/main',
                       r'\+refs/heads/main:.*')]),
        (['master'], [('+refs/heads/master:refs/heads/master',
                       r'\+refs/heads/master:.*')]),
        (['master/'], [('+refs/heads/master:refs/heads/master',
                       r'\+refs/heads/master:.*')]),
        (['+master'], [('+refs/heads/master:refs/heads/master',
                       r'\+refs/heads/master:.*')]),
        (['refs/heads/*'], [('+refs/heads/*:refs/heads/*',
                            r'\+refs/heads/\*:.*')]),
        (['foo/bar/*', 'baz'], [('+refs/heads/foo/bar/*:refs/heads/foo/bar/*',
                                r'\+refs/heads/foo/bar/\*:.*'),
                               ('+refs/heads/baz:refs/heads/baz',
                                r'\+refs/heads/baz:.*')]),
        (['refs/foo/*:refs/bar/*'], [('+refs/foo/*:refs/bar/*',
                                      r'\+refs/foo/\*:.*')])
        ]

    mirror = git_cache.Mirror('test://phony.example.biz')
    for fetch_specs, expected in testData:
      mirror = git_cache.Mirror('test://phony.example.biz', refs=fetch_specs)
      self.assertEqual(mirror.fetch_specs, set(expected))

  def testPopulate(self):
    self.git(['init', '-q'])
    with open(os.path.join(self.origin_dir, 'foo'), 'w') as f:
      f.write('touched\n')
    self.git(['add', 'foo'])
    self.git([
        '-c', 'user.name=Test user', '-c', 'user.email=joj@test.com', 'commit',
        '-m', 'foo'
    ])

    mirror = git_cache.Mirror(self.origin_dir)
    mirror.populate()

  def testPopulateResetFetchConfig(self):
    self.git(['init', '-q'])
    with open(os.path.join(self.origin_dir, 'foo'), 'w') as f:
      f.write('touched\n')
    self.git(['add', 'foo'])
    self.git([
        '-c', 'user.name=Test user', '-c', 'user.email=joj@test.com', 'commit',
        '-m', 'foo'
    ])

    mirror = git_cache.Mirror(self.origin_dir)
    mirror.populate()

    # Add a bad refspec to the cache's fetch config.
    cache_dir = os.path.join(
        self.cache_dir, mirror.UrlToCacheDir(self.origin_dir))
    self.git([
        '--git-dir', cache_dir, 'config', '--add', 'remote.origin.fetch',
        '+refs/heads/foo:refs/heads/foo'
    ],
             cwd=cache_dir)

    mirror.populate(reset_fetch_config=True)

  def testPopulateTwice(self):
    self.git(['init', '-q'])
    with open(os.path.join(self.origin_dir, 'foo'), 'w') as f:
      f.write('touched\n')
    self.git(['add', 'foo'])
    self.git([
        '-c', 'user.name=Test user', '-c', 'user.email=joj@test.com', 'commit',
        '-m', 'foo'
    ])

    mirror = git_cache.Mirror(self.origin_dir)
    mirror.populate()

    mirror.populate()

  @mock.patch('sys.stdout', StringIO())
  def testPruneRequired(self):
    self.git(['init', '-q'])
    with open(os.path.join(self.origin_dir, 'foo'), 'w') as f:
      f.write('touched\n')
    self.git(['checkout', '-b', 'foo'])
    self.git(['add', 'foo'])
    self.git([
        '-c', 'user.name=Test user', '-c', 'user.email=joj@test.com', 'commit',
        '-m', 'foo'
    ])
    mirror = git_cache.Mirror(self.origin_dir)
    mirror.populate()
    self.git(['checkout', '-b', 'foo_tmp', 'foo'])
    self.git(['branch', '-D', 'foo'])
    self.git(['checkout', '-b', 'foo/bar', 'foo_tmp'])
    mirror.populate()
    self.assertNotIn(git_cache.GIT_CACHE_CORRUPT_MESSAGE, sys.stdout.getvalue())

  def _makeGitRepoWithTag(self):
    self.git(['init', '-q'])
    with open(os.path.join(self.origin_dir, 'foo'), 'w') as f:
      f.write('touched\n')
    self.git(['add', 'foo'])
    self.git([
        '-c', 'user.name=Test user', '-c', 'user.email=joj@test.com', 'commit',
        '-m', 'foo'
    ])
    self.git(['tag', 'TAG'])
    self.git(['pack-refs'])

  def testPopulateFetchTagsByDefault(self):
    self._makeGitRepoWithTag()

    # Default behaviour includes tags.
    mirror = git_cache.Mirror(self.origin_dir)
    mirror.populate()

    cache_dir = os.path.join(self.cache_dir,
                             mirror.UrlToCacheDir(self.origin_dir))
    self.assertTrue(os.path.exists(cache_dir + '/refs/tags/TAG'))

  def testPopulateFetchWithoutTags(self):
    self._makeGitRepoWithTag()

    # Ask to not include tags.
    mirror = git_cache.Mirror(self.origin_dir)
    mirror.populate(no_fetch_tags=True)

    cache_dir = os.path.join(self.cache_dir,
                             mirror.UrlToCacheDir(self.origin_dir))
    self.assertFalse(os.path.exists(cache_dir + '/refs/tags/TAG'))

  def testPopulateResetFetchConfigEmptyFetchConfig(self):
    self.git(['init', '-q'])
    with open(os.path.join(self.origin_dir, 'foo'), 'w') as f:
      f.write('touched\n')
    self.git(['add', 'foo'])
    self.git([
        '-c', 'user.name=Test user', '-c', 'user.email=joj@test.com', 'commit',
        '-m', 'foo'
    ])

    mirror = git_cache.Mirror(self.origin_dir)
    mirror.populate(reset_fetch_config=True)


class GitCacheDirTest(unittest.TestCase):
  def setUp(self):
    try:
      delattr(git_cache.Mirror, 'cachepath')
    except AttributeError:
      pass
    super(GitCacheDirTest, self).setUp()

  def tearDown(self):
    try:
      delattr(git_cache.Mirror, 'cachepath')
    except AttributeError:
      pass
    super(GitCacheDirTest, self).tearDown()

  def test_git_config_read(self):
    (fd, tmpFile) = tempfile.mkstemp()
    old = git_cache.Mirror._GIT_CONFIG_LOCATION
    try:
      try:
        os.write(fd, b'[cache]\n  cachepath="hello world"\n')
      finally:
        os.close(fd)

      git_cache.Mirror._GIT_CONFIG_LOCATION = ['-f', tmpFile]

      self.assertEqual(git_cache.Mirror.GetCachePath(), 'hello world')
    finally:
      git_cache.Mirror._GIT_CONFIG_LOCATION = old
      os.remove(tmpFile)

  def test_environ_read(self):
    path = os.environ.get('GIT_CACHE_PATH')
    config = os.environ.get('GIT_CONFIG')
    try:
      os.environ['GIT_CACHE_PATH'] = 'hello world'
      os.environ['GIT_CONFIG'] = 'disabled'

      self.assertEqual(git_cache.Mirror.GetCachePath(), 'hello world')
    finally:
      for name, val in zip(('GIT_CACHE_PATH', 'GIT_CONFIG'), (path, config)):
        if val is None:
          os.environ.pop(name, None)
        else:
          os.environ[name] = val

  def test_manual_set(self):
    git_cache.Mirror.SetCachePath('hello world')
    self.assertEqual(git_cache.Mirror.GetCachePath(), 'hello world')

  def test_unconfigured(self):
    path = os.environ.get('GIT_CACHE_PATH')
    config = os.environ.get('GIT_CONFIG')
    try:
      os.environ.pop('GIT_CACHE_PATH', None)
      os.environ['GIT_CONFIG'] = 'disabled'

      with self.assertRaisesRegexp(RuntimeError, 'cache\.cachepath'):
        git_cache.Mirror.GetCachePath()

      # negatively cached value still raises
      with self.assertRaisesRegexp(RuntimeError, 'cache\.cachepath'):
        git_cache.Mirror.GetCachePath()
    finally:
      for name, val in zip(('GIT_CACHE_PATH', 'GIT_CONFIG'), (path, config)):
        if val is None:
          os.environ.pop(name, None)
        else:
          os.environ[name] = val


class MirrorTest(unittest.TestCase):
  def test_same_cache_for_authenticated_and_unauthenticated_urls(self):
    # GoB can fetch a repo via two different URLs; if the url contains '/a/'
    # it forces authenticated access instead of allowing anonymous access,
    # even in the case where a repo is public. We want this in order to make
    # sure bots are authenticated and get the right quotas. However, we
    # only want to maintain a single cache for the repo.
    self.assertEqual(git_cache.Mirror.UrlToCacheDir(
        'https://chromium.googlesource.com/a/chromium/src.git'),
        'chromium.googlesource.com-chromium-src')


if __name__ == '__main__':
  logging.basicConfig(
      level=logging.DEBUG if '-v' in sys.argv else logging.ERROR)
  sys.exit(coverage_utils.covered_main((
    os.path.join(DEPOT_TOOLS_ROOT, 'git_cache.py')
  ), required_percentage=0))