| #!/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)) |