[git-cache] / test_git_cache.py

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 bytes(p.stdout)


def _commit_file(
        directory: str,
        filename: str,
        contents: str,
        commit_message: str) -> None:
    with open(os.path.join(directory, filename), 'w') as f:
        f.write(contents)
    _git(directory, 'add', filename)
    _git(directory, 'commit', '-m', commit_message)


# pylint: disable=too-many-public-methods
class TestGitCache(unittest.TestCase):

    def setUp(self) -> None:
        self.xdgcache = tempfile.TemporaryDirectory(
            prefix='git_cache_test-cache-')
        self.xdgdata = tempfile.TemporaryDirectory(
            prefix='git_cache_test-data-')
        self.old_XDG_CACHE_HOME = os.environ.get('XDG_CACHE_HOME')
        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'

        os.environ['BACKOFF_MAX_TIME'] = '0'

        self.tempdir = tempfile.TemporaryDirectory(prefix='git_cache_test-')
        self.upstream = os.path.join(self.tempdir.name, 'upstream')
        subprocess.run(['git', 'init', self.upstream], check=True)
        _commit_file(self.upstream, 'file', 'Contents', 'First commit')

    def tearDown(self) -> None:
        _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')

    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')
        self.assertEqual(d1, d2)
        self.assertEqual(rev1, rev2)
        self.assertEqual(_git(d2, 'show', '%s:file' % rev2), 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)
        self.assertEqual(d1, d2)
        self.assertEqual(_git(d2, 'show', '%s:file' % rev), 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')
        self.assertEqual(d1, d2)
        self.assertEqual(rev1, rev2)
        self.assertEqual(_git(d2, 'show', '%s:file' % rev2), b'Contents')

    def test_fetch_new_file(self) -> None:
        d1, rev1 = git_cache.fetch(self.upstream, 'master')
        _commit_file(self.upstream, 'foofile', 'foo', 'Foo')
        d2, rev2 = git_cache.fetch(self.upstream, 'master')
        self.assertEqual(d1, d2)
        self.assertNotEqual(rev1, rev2)
        self.assertEqual(_git(d2, 'show', '%s:foofile' % rev2), b'foo')

    def test_ensure_doesnt_fetch_new_file(self) -> None:
        d1, rev1 = git_cache.fetch(self.upstream, 'master')
        _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)
        self.assertEqual(d1, d2)
        p = subprocess.run(
            ['git', '-C', d2, 'show', '%s:foofile' % rev2], 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')
        self.tempdir.cleanup()
        d2 = git_cache.ensure_rev_available(self.upstream, 'master', rev)
        self.assertEqual(d1, d2)

    def test_ensure_fetches_new_file(self) -> None:
        d1, rev1 = git_cache.fetch(self.upstream, 'master')
        _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)
        self.assertEqual(d1, d2)
        self.assertEqual(_git(d2, 'show', '%s:foofile' % rev2), b'foo')

    def test_fetch_raises_on_invalid_repo(self) -> None:
        self.tempdir.cleanup()
        with self.assertRaises(Exception):
            git_cache.fetch(self.upstream, 'master')

    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)

    def test_fetch_raises_on_invalid_ref(self) -> None:
        with self.assertRaises(Exception):
            git_cache.fetch(self.upstream, 'nobranch')

    def test_ensure_raises_on_invalid_ref(self) -> None:
        rev = _git(self.upstream, 'log', '--format=%H', '-n1').strip().decode()
        with self.assertRaises(Exception):
            git_cache.ensure_rev_available(self.upstream, 'nobranch', rev)

    def test_ensure_raises_on_invalid_rev(self) -> None:
        with self.assertRaises(Exception):
            git_cache.ensure_rev_available(
                self.upstream,
                'nobranch',
                '1234567890abcdef01234567890abcdef1234567')

    def test_ensure_raises_on_rev_from_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()
        with self.assertRaises(Exception):
            git_cache.ensure_rev_available(self.upstream, 'master', 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')

    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', '%s:foofile' % rev), b'foo')
        _git(self.upstream, 'checkout', 'master')
        _git(self.upstream, 'merge', '--ff-only', 'otherbranch')
        d = git_cache.ensure_rev_available(self.upstream, 'master', rev)
        self.assertEqual(_git(d, 'show', '%s:foofile' % rev), b'foo')

    def test_fetch_after_cache_deleted(self) -> None:
        d1, rev1 = git_cache.fetch(self.upstream, 'master')
        shutil.rmtree(d1)
        d2, rev2 = git_cache.fetch(self.upstream, 'master')
        self.assertEqual(d1, d2)
        self.assertEqual(rev1, rev2)
        self.assertEqual(_git(d2, 'show', '%s:file' % rev2), b'Contents')

    def test_ensure_after_cache_deleted(self) -> None:
        d1, rev = git_cache.fetch(self.upstream, 'master')
        shutil.rmtree(d1)
        d2 = git_cache.ensure_rev_available(self.upstream, 'master', rev)
        self.assertEqual(d1, d2)
        self.assertEqual(_git(d2, 'show', '%s:file' % rev), b'Contents')

    def test_fetch_raises_on_amend(self) -> None:
        git_cache.fetch(self.upstream, 'master')
        _git(self.upstream, 'commit', '--amend', '-m', 'Amended')
        with self.assertRaises(Exception):
            git_cache.fetch(self.upstream, 'master')

    def test_ensure_raises_on_amend(self) -> None:
        git_cache.fetch(self.upstream, 'master')
        _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)

    def test_fetch_raises_on_amend_after_cache_deleted(self) -> None:
        d, _ = git_cache.fetch(self.upstream, 'master')
        shutil.rmtree(d)
        _git(self.upstream, 'commit', '--amend', '-m', 'Amended')
        with self.assertRaises(Exception):
            git_cache.fetch(self.upstream, 'master')

    def test_ensure_raises_on_amend_after_cache_deleted(self) -> None:
        d, _ = git_cache.fetch(self.upstream, 'master')
        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)


if __name__ == '__main__':
    unittest.main()
Commit	Line	Data
bef7ce53 SW	1	import os.path
	2	import tempfile
	3	import shutil
	4	import subprocess
	5	import unittest
	6
8d251eca SW	7	from typing import Optional
8d251eca SW	8
bef7ce53 SW	9	import git_cache
	10
	11
8d251eca SW	12	def _setenv(var: str, value: Optional[str]) -> None:
	13	if value is None:
	14	del os.environ[var]
	15	else:
	16	os.environ[var] = value
	17
	18
bef7ce53 SW	19	def _git(directory: str, *args: str) -> bytes:
	20	p = subprocess.run(['git', '-C', directory] + list(args),
	21	stdout=subprocess.PIPE, check=True)
5dd71f12	22	return bytes(p.stdout)
bef7ce53 SW	23
	24
	25	def _commit_file(
	26	directory: str,
	27	filename: str,
	28	contents: str,
	29	commit_message: str) -> None:
	30	with open(os.path.join(directory, filename), 'w') as f:
	31	f.write(contents)
	32	_git(directory, 'add', filename)
	33	_git(directory, 'commit', '-m', commit_message)
	34
	35
	36	# pylint: disable=too-many-public-methods
	37	class TestGitCache(unittest.TestCase):
	38
	39	def setUp(self) -> None:
35000f72 SW	40	self.xdgcache = tempfile.TemporaryDirectory(
	41	prefix='git_cache_test-cache-')
	42	self.xdgdata = tempfile.TemporaryDirectory(
	43	prefix='git_cache_test-data-')
bef7ce53	44	self.old_XDG_CACHE_HOME = os.environ.get('XDG_CACHE_HOME')
35000f72	45	self.old_XDG_DATA_HOME = os.environ.get('XDG_DATA_HOME')
8d251eca	46	_setenv('XDG_CACHE_HOME', self.xdgcache.name)
35000f72	47	_setenv('XDG_DATA_HOME', self.xdgdata.name)
bef7ce53 SW	48
	49	os.environ['GIT_AUTHOR_NAME'] = 'test_git_cache'
	50	os.environ['GIT_COMMITTER_NAME'] = 'test_git_cache'
	51	os.environ['GIT_AUTHOR_EMAIL'] = 'test_git_cache@example.com'
	52	os.environ['GIT_COMMITTER_EMAIL'] = 'test_git_cache@example.com'
	53
f36d5c6f SW	54	os.environ['BACKOFF_MAX_TIME'] = '0'
f36d5c6f SW	55
bef7ce53 SW	56	self.tempdir = tempfile.TemporaryDirectory(prefix='git_cache_test-')
	57	self.upstream = os.path.join(self.tempdir.name, 'upstream')
	58	subprocess.run(['git', 'init', self.upstream], check=True)
	59	_commit_file(self.upstream, 'file', 'Contents', 'First commit')
	60
	61	def tearDown(self) -> None:
8d251eca	62	_setenv('XDG_CACHE_HOME', self.old_XDG_CACHE_HOME)
35000f72	63	_setenv('XDG_DATA_HOME', self.old_XDG_DATA_HOME)
bef7ce53 SW	64
	65	self.tempdir.cleanup()
	66	self.xdgcache.cleanup()
	67
	68	def test_fetch(self) -> None:
	69	d, rev = git_cache.fetch(self.upstream, 'master')
	70	self.assertEqual(_git(d, 'show', '%s:file' % rev), b'Contents')
	71
	72	def test_fetch_twice(self) -> None:
	73	d1, rev1 = git_cache.fetch(self.upstream, 'master')
	74	self.assertEqual(_git(d1, 'show', '%s:file' % rev1), b'Contents')
	75	d2, rev2 = git_cache.fetch(self.upstream, 'master')
	76	self.assertEqual(d1, d2)
	77	self.assertEqual(rev1, rev2)
	78	self.assertEqual(_git(d2, 'show', '%s:file' % rev2), b'Contents')
	79
	80	def test_fetch_then_ensure(self) -> None:
	81	d1, rev = git_cache.fetch(self.upstream, 'master')
	82	self.assertEqual(_git(d1, 'show', '%s:file' % rev), b'Contents')
	83	d2 = git_cache.ensure_rev_available(self.upstream, 'master', rev)
	84	self.assertEqual(d1, d2)
	85	self.assertEqual(_git(d2, 'show', '%s:file' % rev), b'Contents')
	86
	87	def test_ensure_then_fetch(self) -> None:
	88	rev1 = _git(
	89	self.upstream, 'log', '--format=%H', '-n1').strip().decode()
	90	d1 = git_cache.ensure_rev_available(self.upstream, 'master', rev1)
	91	self.assertEqual(_git(d1, 'show', '%s:file' % rev1), b'Contents')
	92	d2, rev2 = git_cache.fetch(self.upstream, 'master')
	93	self.assertEqual(d1, d2)
	94	self.assertEqual(rev1, rev2)
	95	self.assertEqual(_git(d2, 'show', '%s:file' % rev2), b'Contents')
	96
	97	def test_fetch_new_file(self) -> None:
	98	d1, rev1 = git_cache.fetch(self.upstream, 'master')
	99	_commit_file(self.upstream, 'foofile', 'foo', 'Foo')
	100	d2, rev2 = git_cache.fetch(self.upstream, 'master')
	101	self.assertEqual(d1, d2)
	102	self.assertNotEqual(rev1, rev2)
	103	self.assertEqual(_git(d2, 'show', '%s:foofile' % rev2), b'foo')
	104
	105	def test_ensure_doesnt_fetch_new_file(self) -> None:
	106	d1, rev1 = git_cache.fetch(self.upstream, 'master')
	107	_commit_file(self.upstream, 'foofile', 'foo', 'Foo')
	108	rev2 = _git(
	109	self.upstream, 'log', '--format=%H', '-n1').strip().decode()
	110	self.assertNotEqual(rev1, rev2)
	111	d2 = git_cache.ensure_rev_available(self.upstream, 'master', rev1)
	112	self.assertEqual(d1, d2)
	113	p = subprocess.run(
	114	['git', '-C', d2, 'show', '%s:foofile' % rev2], check=False)
	115	self.assertNotEqual(p.returncode, 0)
	116
	117	def test_ensure_doesnt_fetch_from_deleted_upstream(self) -> None:
	118	d1, rev = git_cache.fetch(self.upstream, 'master')
	119	self.tempdir.cleanup()
	120	d2 = git_cache.ensure_rev_available(self.upstream, 'master', rev)
	121	self.assertEqual(d1, d2)
	122
	123	def test_ensure_fetches_new_file(self) -> None:
	124	d1, rev1 = git_cache.fetch(self.upstream, 'master')
	125	_commit_file(self.upstream, 'foofile', 'foo', 'Foo')
	126	rev2 = _git(
	127	self.upstream, 'log', '--format=%H', '-n1').strip().decode()
128	self.assertNotEqual(rev1, rev2)
129	d2 = git_cache.ensure_rev_available(self.upstream, 'master', rev2)
130	self.assertEqual(d1, d2)
131	self.assertEqual(_git(d2, 'show', '%s:foofile' % rev2), b'foo')
132
133	def test_fetch_raises_on_invalid_repo(self) -> None:
134	self.tempdir.cleanup()
135	with self.assertRaises(Exception):
136	git_cache.fetch(self.upstream, 'master')
137
138	def test_ensure_raises_on_invalid_repo(self) -> None:
139	rev = _git(self.upstream, 'log', '--format=%H', '-n1').strip().decode()
140	self.tempdir.cleanup()
141	with self.assertRaises(Exception):
142	git_cache.ensure_rev_available(self.upstream, 'master', rev)
143
144	def test_fetch_raises_on_invalid_ref(self) -> None:
145	with self.assertRaises(Exception):
146	git_cache.fetch(self.upstream, 'nobranch')
147
148	def test_ensure_raises_on_invalid_ref(self) -> None:
149	rev = _git(self.upstream, 'log', '--format=%H', '-n1').strip().decode()
150	with self.assertRaises(Exception):
151	git_cache.ensure_rev_available(self.upstream, 'nobranch', rev)
152
153	def test_ensure_raises_on_invalid_rev(self) -> None:
154	with self.assertRaises(Exception):
155	git_cache.ensure_rev_available(
156	self.upstream,
157	'nobranch',
158	'1234567890abcdef01234567890abcdef1234567')
159
160	def test_ensure_raises_on_rev_from_other_branch(self) -> None:
161	_git(self.upstream, 'checkout', '-b', 'otherbranch')
162	_commit_file(self.upstream, 'foofile', 'foo', 'Foo')
163	rev = _git(self.upstream, 'log', '--format=%H', '-n1').strip().decode()
164	with self.assertRaises(Exception):
165	git_cache.ensure_rev_available(self.upstream, 'master', rev)
166
167	def test_ensure_other_branch(self) -> None:
168	_git(self.upstream, 'checkout', '-b', 'otherbranch')
169	_commit_file(self.upstream, 'foofile', 'foo', 'Foo')
170	rev = _git(self.upstream, 'log', '--format=%H', '-n1').strip().decode()
171	d = git_cache.ensure_rev_available(self.upstream, 'otherbranch', rev)
172	self.assertEqual(_git(d, 'show', '%s:foofile' % rev), b'foo')
173
eb638847 SW	174	def test_catch_up(self) -> None:
	175	_git(self.upstream, 'checkout', '-b', 'otherbranch')
	176	_commit_file(self.upstream, 'foofile', 'foo', 'Foo')
	177	rev = _git(self.upstream, 'log', '--format=%H', '-n1').strip().decode()
	178	d = git_cache.ensure_rev_available(self.upstream, 'otherbranch', rev)
	179	self.assertEqual(_git(d, 'show', '%s:foofile' % rev), b'foo')
	180	_git(self.upstream, 'checkout', 'master')
	181	_git(self.upstream, 'merge', '--ff-only', 'otherbranch')
	182	d = git_cache.ensure_rev_available(self.upstream, 'master', rev)
	183	self.assertEqual(_git(d, 'show', '%s:foofile' % rev), b'foo')
	184
bef7ce53 SW	185	def test_fetch_after_cache_deleted(self) -> None:
	186	d1, rev1 = git_cache.fetch(self.upstream, 'master')
	187	shutil.rmtree(d1)
	188	d2, rev2 = git_cache.fetch(self.upstream, 'master')
	189	self.assertEqual(d1, d2)
	190	self.assertEqual(rev1, rev2)
	191	self.assertEqual(_git(d2, 'show', '%s:file' % rev2), b'Contents')
	192
	193	def test_ensure_after_cache_deleted(self) -> None:
	194	d1, rev = git_cache.fetch(self.upstream, 'master')
	195	shutil.rmtree(d1)
	196	d2 = git_cache.ensure_rev_available(self.upstream, 'master', rev)
	197	self.assertEqual(d1, d2)
	198	self.assertEqual(_git(d2, 'show', '%s:file' % rev), b'Contents')
	199
	200	def test_fetch_raises_on_amend(self) -> None:
	201	git_cache.fetch(self.upstream, 'master')
	202	_git(self.upstream, 'commit', '--amend', '-m', 'Amended')
	203	with self.assertRaises(Exception):
	204	git_cache.fetch(self.upstream, 'master')
	205
	206	def test_ensure_raises_on_amend(self) -> None:
	207	git_cache.fetch(self.upstream, 'master')
	208	_git(self.upstream, 'commit', '--amend', '-m', 'Amended')
	209	rev = _git(self.upstream, 'log', '--format=%H', '-n1').strip().decode()
	210	with self.assertRaises(Exception):
	211	git_cache.ensure_rev_available(self.upstream, 'master', rev)
	212
35000f72 SW	213	def test_fetch_raises_on_amend_after_cache_deleted(self) -> None:
	214	d, _ = git_cache.fetch(self.upstream, 'master')
	215	shutil.rmtree(d)
	216	_git(self.upstream, 'commit', '--amend', '-m', 'Amended')
	217	with self.assertRaises(Exception):
	218	git_cache.fetch(self.upstream, 'master')
	219
	220	def test_ensure_raises_on_amend_after_cache_deleted(self) -> None:
	221	d, _ = git_cache.fetch(self.upstream, 'master')
	222	shutil.rmtree(d)
	223	_git(self.upstream, 'commit', '--amend', '-m', 'Amended')
	224	rev = _git(self.upstream, 'log', '--format=%H', '-n1').strip().decode()
	225	with self.assertRaises(Exception):
	226	git_cache.ensure_rev_available(self.upstream, 'master', rev)
	227
bef7ce53 SW	228
	229	if __name__ == '__main__':
	230	unittest.main()