[git-cache] / test_git_cache.py

import os.path
import tempfile
import shutil
import subprocess
import unittest

import git_cache


def _git(directory: str, *args: str) -> bytes:
    p = subprocess.run(['git', '-C', directory] + list(args),
                       stdout=subprocess.PIPE, check=True)
    return 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-')
        self.old_XDG_CACHE_HOME = os.environ.get('XDG_CACHE_HOME')
        os.environ['XDG_CACHE_HOME'] = self.xdgcache.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:
        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

        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_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)


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