You cannot select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
260 lines
8.7 KiB
Python
260 lines
8.7 KiB
Python
#!/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"""
|
|
|
|
import logging
|
|
import os
|
|
import shutil
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
|
|
if sys.version_info.major == 2:
|
|
from StringIO import StringIO
|
|
import mock
|
|
else:
|
|
from io import StringIO
|
|
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)
|
|
|
|
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(['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(['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(['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(['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(['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(['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(['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))
|