X-Git-Url: http://git.scottworley.com/git-cache/blobdiff_plain/bef7ce53ce70499b4f13985cc9d4b64e2d8aace4..ce9660cb142544ff3612bb2e39e2d2860a72dc5c:/test_git_cache.py diff --git a/test_git_cache.py b/test_git_cache.py index 7408cad..1845c01 100644 --- a/test_git_cache.py +++ b/test_git_cache.py @@ -1,16 +1,32 @@ +# git-cache: Cache git content locally +# +# This program is free software: you can redistribute it and/or modify it +# under the terms of the GNU General Public License as published by the +# Free Software Foundation, version 3. + + import os.path import tempfile import shutil import subprocess import unittest +from typing import Optional + import git_cache +def _setenv(var: str, value: Optional[str]) -> None: + if value is None: + del os.environ[var] + else: + os.environ[var] = value + + def _git(directory: str, *args: str) -> bytes: p = subprocess.run(['git', '-C', directory] + list(args), stdout=subprocess.PIPE, check=True) - return p.stdout + return bytes(p.stdout) def _commit_file( @@ -18,7 +34,7 @@ def _commit_file( filename: str, contents: str, commit_message: str) -> None: - with open(os.path.join(directory, filename), 'w') as f: + with open(os.path.join(directory, filename), 'w', encoding='utf-8') as f: f.write(contents) _git(directory, 'add', filename) _git(directory, 'commit', '-m', commit_message) @@ -28,104 +44,112 @@ def _commit_file( class TestGitCache(unittest.TestCase): def setUp(self) -> None: - self.xdgcache = tempfile.TemporaryDirectory(prefix='git_cache_test-') + self.xdgcache = tempfile.TemporaryDirectory( # pylint: disable=consider-using-with + prefix='git_cache_test-cache-') + self.xdgdata = tempfile.TemporaryDirectory( # pylint: disable=consider-using-with + prefix='git_cache_test-data-') self.old_XDG_CACHE_HOME = os.environ.get('XDG_CACHE_HOME') - os.environ['XDG_CACHE_HOME'] = self.xdgcache.name + self.old_XDG_DATA_HOME = os.environ.get('XDG_DATA_HOME') + _setenv('XDG_CACHE_HOME', self.xdgcache.name) + _setenv('XDG_DATA_HOME', self.xdgdata.name) os.environ['GIT_AUTHOR_NAME'] = 'test_git_cache' os.environ['GIT_COMMITTER_NAME'] = 'test_git_cache' os.environ['GIT_AUTHOR_EMAIL'] = 'test_git_cache@example.com' os.environ['GIT_COMMITTER_EMAIL'] = 'test_git_cache@example.com' - self.tempdir = tempfile.TemporaryDirectory(prefix='git_cache_test-') + os.environ['BACKOFF_MAX_TIME'] = '0' + os.environ['FORCE_WARNING_TIME'] = '0' # ONLY FOR TEST USE! + + self.tempdir = tempfile.TemporaryDirectory( # pylint: disable=consider-using-with + prefix='git_cache_test-') self.upstream = os.path.join(self.tempdir.name, 'upstream') - subprocess.run(['git', 'init', self.upstream], check=True) + subprocess.run(['git', '-c', 'init.defaultBranch=main', + 'init', self.upstream], check=True) _commit_file(self.upstream, 'file', 'Contents', 'First commit') def tearDown(self) -> None: - if self.old_XDG_CACHE_HOME is None: - del os.environ['XDG_CACHE_HOME'] - else: - os.environ['XDG_CACHE_HOME'] = self.old_XDG_CACHE_HOME + _setenv('XDG_CACHE_HOME', self.old_XDG_CACHE_HOME) + _setenv('XDG_DATA_HOME', self.old_XDG_DATA_HOME) self.tempdir.cleanup() self.xdgcache.cleanup() def test_fetch(self) -> None: - d, rev = git_cache.fetch(self.upstream, 'master') - self.assertEqual(_git(d, 'show', '%s:file' % rev), b'Contents') + d, rev = git_cache.fetch(self.upstream, 'main') + self.assertEqual(_git(d, 'show', f'{rev}:file'), b'Contents') def test_fetch_twice(self) -> None: - d1, rev1 = git_cache.fetch(self.upstream, 'master') - self.assertEqual(_git(d1, 'show', '%s:file' % rev1), b'Contents') - d2, rev2 = git_cache.fetch(self.upstream, 'master') + d1, rev1 = git_cache.fetch(self.upstream, 'main') + self.assertEqual(_git(d1, 'show', f'{rev1}:file'), b'Contents') + d2, rev2 = git_cache.fetch(self.upstream, 'main') self.assertEqual(d1, d2) self.assertEqual(rev1, rev2) - self.assertEqual(_git(d2, 'show', '%s:file' % rev2), b'Contents') + self.assertEqual(_git(d2, 'show', f'{rev2}:file'), b'Contents') def test_fetch_then_ensure(self) -> None: - d1, rev = git_cache.fetch(self.upstream, 'master') - self.assertEqual(_git(d1, 'show', '%s:file' % rev), b'Contents') - d2 = git_cache.ensure_rev_available(self.upstream, 'master', rev) + d1, rev = git_cache.fetch(self.upstream, 'main') + self.assertEqual(_git(d1, 'show', f'{rev}:file'), b'Contents') + d2 = git_cache.ensure_rev_available(self.upstream, 'main', rev) self.assertEqual(d1, d2) - self.assertEqual(_git(d2, 'show', '%s:file' % rev), b'Contents') + self.assertEqual(_git(d2, 'show', f'{rev}:file'), b'Contents') def test_ensure_then_fetch(self) -> None: rev1 = _git( self.upstream, 'log', '--format=%H', '-n1').strip().decode() - d1 = git_cache.ensure_rev_available(self.upstream, 'master', rev1) - self.assertEqual(_git(d1, 'show', '%s:file' % rev1), b'Contents') - d2, rev2 = git_cache.fetch(self.upstream, 'master') + d1 = git_cache.ensure_rev_available(self.upstream, 'main', rev1) + self.assertEqual(_git(d1, 'show', f'{rev1}:file'), b'Contents') + d2, rev2 = git_cache.fetch(self.upstream, 'main') self.assertEqual(d1, d2) self.assertEqual(rev1, rev2) - self.assertEqual(_git(d2, 'show', '%s:file' % rev2), b'Contents') + self.assertEqual(_git(d2, 'show', f'{rev2}:file'), b'Contents') def test_fetch_new_file(self) -> None: - d1, rev1 = git_cache.fetch(self.upstream, 'master') + d1, rev1 = git_cache.fetch(self.upstream, 'main') _commit_file(self.upstream, 'foofile', 'foo', 'Foo') - d2, rev2 = git_cache.fetch(self.upstream, 'master') + d2, rev2 = git_cache.fetch(self.upstream, 'main') self.assertEqual(d1, d2) self.assertNotEqual(rev1, rev2) - self.assertEqual(_git(d2, 'show', '%s:foofile' % rev2), b'foo') + self.assertEqual(_git(d2, 'show', f'{rev2}:foofile'), b'foo') def test_ensure_doesnt_fetch_new_file(self) -> None: - d1, rev1 = git_cache.fetch(self.upstream, 'master') + d1, rev1 = git_cache.fetch(self.upstream, 'main') _commit_file(self.upstream, 'foofile', 'foo', 'Foo') rev2 = _git( self.upstream, 'log', '--format=%H', '-n1').strip().decode() self.assertNotEqual(rev1, rev2) - d2 = git_cache.ensure_rev_available(self.upstream, 'master', rev1) + d2 = git_cache.ensure_rev_available(self.upstream, 'main', rev1) self.assertEqual(d1, d2) p = subprocess.run( - ['git', '-C', d2, 'show', '%s:foofile' % rev2], check=False) + ['git', '-C', d2, 'show', f'{rev2}:foofile'], check=False) self.assertNotEqual(p.returncode, 0) def test_ensure_doesnt_fetch_from_deleted_upstream(self) -> None: - d1, rev = git_cache.fetch(self.upstream, 'master') + d1, rev = git_cache.fetch(self.upstream, 'main') self.tempdir.cleanup() - d2 = git_cache.ensure_rev_available(self.upstream, 'master', rev) + d2 = git_cache.ensure_rev_available(self.upstream, 'main', rev) self.assertEqual(d1, d2) def test_ensure_fetches_new_file(self) -> None: - d1, rev1 = git_cache.fetch(self.upstream, 'master') + d1, rev1 = git_cache.fetch(self.upstream, 'main') _commit_file(self.upstream, 'foofile', 'foo', 'Foo') rev2 = _git( self.upstream, 'log', '--format=%H', '-n1').strip().decode() self.assertNotEqual(rev1, rev2) - d2 = git_cache.ensure_rev_available(self.upstream, 'master', rev2) + d2 = git_cache.ensure_rev_available(self.upstream, 'main', rev2) self.assertEqual(d1, d2) - self.assertEqual(_git(d2, 'show', '%s:foofile' % rev2), b'foo') + self.assertEqual(_git(d2, 'show', f'{rev2}:foofile'), b'foo') def test_fetch_raises_on_invalid_repo(self) -> None: self.tempdir.cleanup() with self.assertRaises(Exception): - git_cache.fetch(self.upstream, 'master') + git_cache.fetch(self.upstream, 'main') def test_ensure_raises_on_invalid_repo(self) -> None: rev = _git(self.upstream, 'log', '--format=%H', '-n1').strip().decode() self.tempdir.cleanup() with self.assertRaises(Exception): - git_cache.ensure_rev_available(self.upstream, 'master', rev) + git_cache.ensure_rev_available(self.upstream, 'main', rev) def test_fetch_raises_on_invalid_ref(self) -> None: with self.assertRaises(Exception): @@ -148,42 +172,94 @@ class TestGitCache(unittest.TestCase): _commit_file(self.upstream, 'foofile', 'foo', 'Foo') rev = _git(self.upstream, 'log', '--format=%H', '-n1').strip().decode() with self.assertRaises(Exception): - git_cache.ensure_rev_available(self.upstream, 'master', rev) + git_cache.ensure_rev_available(self.upstream, 'main', rev) def test_ensure_other_branch(self) -> None: _git(self.upstream, 'checkout', '-b', 'otherbranch') _commit_file(self.upstream, 'foofile', 'foo', 'Foo') rev = _git(self.upstream, 'log', '--format=%H', '-n1').strip().decode() d = git_cache.ensure_rev_available(self.upstream, 'otherbranch', rev) - self.assertEqual(_git(d, 'show', '%s:foofile' % rev), b'foo') + self.assertEqual(_git(d, 'show', f'{rev}:foofile'), b'foo') + + def test_catch_up(self) -> None: + _git(self.upstream, 'checkout', '-b', 'otherbranch') + _commit_file(self.upstream, 'foofile', 'foo', 'Foo') + rev = _git(self.upstream, 'log', '--format=%H', '-n1').strip().decode() + d = git_cache.ensure_rev_available(self.upstream, 'otherbranch', rev) + self.assertEqual(_git(d, 'show', f'{rev}:foofile'), b'foo') + _git(self.upstream, 'checkout', 'main') + _git(self.upstream, 'merge', '--ff-only', 'otherbranch') + d = git_cache.ensure_rev_available(self.upstream, 'main', rev) + self.assertEqual(_git(d, 'show', f'{rev}:foofile'), b'foo') def test_fetch_after_cache_deleted(self) -> None: - d1, rev1 = git_cache.fetch(self.upstream, 'master') + d1, rev1 = git_cache.fetch(self.upstream, 'main') shutil.rmtree(d1) - d2, rev2 = git_cache.fetch(self.upstream, 'master') + d2, rev2 = git_cache.fetch(self.upstream, 'main') self.assertEqual(d1, d2) self.assertEqual(rev1, rev2) - self.assertEqual(_git(d2, 'show', '%s:file' % rev2), b'Contents') + self.assertEqual(_git(d2, 'show', f'{rev2}:file'), b'Contents') def test_ensure_after_cache_deleted(self) -> None: - d1, rev = git_cache.fetch(self.upstream, 'master') + d1, rev = git_cache.fetch(self.upstream, 'main') shutil.rmtree(d1) - d2 = git_cache.ensure_rev_available(self.upstream, 'master', rev) + d2 = git_cache.ensure_rev_available(self.upstream, 'main', rev) self.assertEqual(d1, d2) - self.assertEqual(_git(d2, 'show', '%s:file' % rev), b'Contents') + self.assertEqual(_git(d2, 'show', f'{rev}:file'), b'Contents') def test_fetch_raises_on_amend(self) -> None: - git_cache.fetch(self.upstream, 'master') + git_cache.fetch(self.upstream, 'main') _git(self.upstream, 'commit', '--amend', '-m', 'Amended') with self.assertRaises(Exception): - git_cache.fetch(self.upstream, 'master') + git_cache.fetch(self.upstream, 'main') def test_ensure_raises_on_amend(self) -> None: - git_cache.fetch(self.upstream, 'master') + git_cache.fetch(self.upstream, 'main') + _git(self.upstream, 'commit', '--amend', '-m', 'Amended') + rev = _git(self.upstream, 'log', '--format=%H', '-n1').strip().decode() + with self.assertRaises(Exception): + git_cache.ensure_rev_available(self.upstream, 'main', rev) + + def test_fetch_raises_on_amend_after_cache_deleted(self) -> None: + d, _ = git_cache.fetch(self.upstream, 'main') + shutil.rmtree(d) + _git(self.upstream, 'commit', '--amend', '-m', 'Amended') + with self.assertRaises(Exception): + git_cache.fetch(self.upstream, 'main') + + def test_ensure_raises_on_amend_after_cache_deleted(self) -> None: + d, _ = git_cache.fetch(self.upstream, 'main') + shutil.rmtree(d) _git(self.upstream, 'commit', '--amend', '-m', 'Amended') rev = _git(self.upstream, 'log', '--format=%H', '-n1').strip().decode() with self.assertRaises(Exception): - git_cache.ensure_rev_available(self.upstream, 'master', rev) + git_cache.ensure_rev_available(self.upstream, 'main', rev) + + def test_force_fetch_after_amend(self) -> None: + git_cache.fetch(self.upstream, 'main') + _git(self.upstream, 'commit', '--amend', '-m', 'Amended') + git_cache.fetch(self.upstream, 'main', force=True) + + def test_force_ensure_after_amend(self) -> None: + git_cache.fetch(self.upstream, 'main') + _git(self.upstream, 'commit', '--amend', '-m', 'Amended') + rev = _git(self.upstream, 'log', '--format=%H', '-n1').strip().decode() + git_cache.ensure_rev_available( + self.upstream, 'main', rev, force=True) + + def test_force_fetch_after_amend_and_cache_delete(self) -> None: + d, _ = git_cache.fetch(self.upstream, 'main') + shutil.rmtree(d) + _git(self.upstream, 'commit', '--amend', '-m', 'Amended') + git_cache.fetch(self.upstream, 'main', force=True) + + def test_force_ensure_after_amend_and_cache_delete(self) -> None: + d, _ = git_cache.fetch(self.upstream, 'main') + shutil.rmtree(d) + _git(self.upstream, 'commit', '--amend', '-m', 'Amended') + rev = _git(self.upstream, 'log', '--format=%H', '-n1').strip().decode() + git_cache.ensure_rev_available( + self.upstream, 'main', rev, force=True) if __name__ == '__main__':