[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'

        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
	40	self.tempdir = tempfile.TemporaryDirectory(prefix='git_cache_test-')
	41	self.upstream = os.path.join(self.tempdir.name, 'upstream')
	42	subprocess.run(['git', 'init', self.upstream], check=True)
	43	_commit_file(self.upstream, 'file', 'Contents', 'First commit')
	44
	45	def tearDown(self) -> None:
	46	if self.old_XDG_CACHE_HOME is None:
	47	del os.environ['XDG_CACHE_HOME']
	48	else:
	49	os.environ['XDG_CACHE_HOME'] = self.old_XDG_CACHE_HOME
	50
	51	self.tempdir.cleanup()
	52	self.xdgcache.cleanup()
	53
	54	def test_fetch(self) -> None:
	55	d, rev = git_cache.fetch(self.upstream, 'master')
	56	self.assertEqual(_git(d, 'show', '%s:file' % rev), b'Contents')
	57
	58	def test_fetch_twice(self) -> None:
	59	d1, rev1 = git_cache.fetch(self.upstream, 'master')
	60	self.assertEqual(_git(d1, 'show', '%s:file' % rev1), b'Contents')
	61	d2, rev2 = git_cache.fetch(self.upstream, 'master')
	62	self.assertEqual(d1, d2)
	63	self.assertEqual(rev1, rev2)
	64	self.assertEqual(_git(d2, 'show', '%s:file' % rev2), b'Contents')
65
66	def test_fetch_then_ensure(self) -> None:
67	d1, rev = git_cache.fetch(self.upstream, 'master')
68	self.assertEqual(_git(d1, 'show', '%s:file' % rev), b'Contents')
69	d2 = git_cache.ensure_rev_available(self.upstream, 'master', rev)
70	self.assertEqual(d1, d2)
71	self.assertEqual(_git(d2, 'show', '%s:file' % rev), b'Contents')
72
73	def test_ensure_then_fetch(self) -> None:
74	rev1 = _git(
75	self.upstream, 'log', '--format=%H', '-n1').strip().decode()
76	d1 = git_cache.ensure_rev_available(self.upstream, 'master', rev1)
77	self.assertEqual(_git(d1, 'show', '%s:file' % rev1), b'Contents')
78	d2, rev2 = git_cache.fetch(self.upstream, 'master')
79	self.assertEqual(d1, d2)
80	self.assertEqual(rev1, rev2)
81	self.assertEqual(_git(d2, 'show', '%s:file' % rev2), b'Contents')
82
83	def test_fetch_new_file(self) -> None:
84	d1, rev1 = git_cache.fetch(self.upstream, 'master')
85	_commit_file(self.upstream, 'foofile', 'foo', 'Foo')
86	d2, rev2 = git_cache.fetch(self.upstream, 'master')
87	self.assertEqual(d1, d2)
88	self.assertNotEqual(rev1, rev2)
89	self.assertEqual(_git(d2, 'show', '%s:foofile' % rev2), b'foo')
90
91	def test_ensure_doesnt_fetch_new_file(self) -> None:
92	d1, rev1 = git_cache.fetch(self.upstream, 'master')
93	_commit_file(self.upstream, 'foofile', 'foo', 'Foo')
94	rev2 = _git(
95	self.upstream, 'log', '--format=%H', '-n1').strip().decode()
96	self.assertNotEqual(rev1, rev2)
97	d2 = git_cache.ensure_rev_available(self.upstream, 'master', rev1)
98	self.assertEqual(d1, d2)
99	p = subprocess.run(
100	['git', '-C', d2, 'show', '%s:foofile' % rev2], check=False)
101	self.assertNotEqual(p.returncode, 0)
102
103	def test_ensure_doesnt_fetch_from_deleted_upstream(self) -> None:
104	d1, rev = git_cache.fetch(self.upstream, 'master')
105	self.tempdir.cleanup()
106	d2 = git_cache.ensure_rev_available(self.upstream, 'master', rev)
107	self.assertEqual(d1, d2)
108
109	def test_ensure_fetches_new_file(self) -> None:
110	d1, rev1 = git_cache.fetch(self.upstream, 'master')
111	_commit_file(self.upstream, 'foofile', 'foo', 'Foo')
112	rev2 = _git(
113	self.upstream, 'log', '--format=%H', '-n1').strip().decode()
114	self.assertNotEqual(rev1, rev2)
115	d2 = git_cache.ensure_rev_available(self.upstream, 'master', rev2)
116	self.assertEqual(d1, d2)
117	self.assertEqual(_git(d2, 'show', '%s:foofile' % rev2), b'foo')
118
119	def test_fetch_raises_on_invalid_repo(self) -> None:
120	self.tempdir.cleanup()
121	with self.assertRaises(Exception):
122	git_cache.fetch(self.upstream, 'master')
123
124	def test_ensure_raises_on_invalid_repo(self) -> None:
125	rev = _git(self.upstream, 'log', '--format=%H', '-n1').strip().decode()
126	self.tempdir.cleanup()
127	with self.assertRaises(Exception):
128	git_cache.ensure_rev_available(self.upstream, 'master', rev)
129
130	def test_fetch_raises_on_invalid_ref(self) -> None:
131	with self.assertRaises(Exception):
132	git_cache.fetch(self.upstream, 'nobranch')
133
134	def test_ensure_raises_on_invalid_ref(self) -> None:
135	rev = _git(self.upstream, 'log', '--format=%H', '-n1').strip().decode()
136	with self.assertRaises(Exception):
137	git_cache.ensure_rev_available(self.upstream, 'nobranch', rev)
138
139	def test_ensure_raises_on_invalid_rev(self) -> None:
140	with self.assertRaises(Exception):
141	git_cache.ensure_rev_available(
142	self.upstream,
143	'nobranch',
144	'1234567890abcdef01234567890abcdef1234567')
145
146	def test_ensure_raises_on_rev_from_other_branch(self) -> None:
147	_git(self.upstream, 'checkout', '-b', 'otherbranch')
148	_commit_file(self.upstream, 'foofile', 'foo', 'Foo')
149	rev = _git(self.upstream, 'log', '--format=%H', '-n1').strip().decode()
150	with self.assertRaises(Exception):
151	git_cache.ensure_rev_available(self.upstream, 'master', rev)
152
153	def test_ensure_other_branch(self) -> None:
154	_git(self.upstream, 'checkout', '-b', 'otherbranch')
155	_commit_file(self.upstream, 'foofile', 'foo', 'Foo')
156	rev = _git(self.upstream, 'log', '--format=%H', '-n1').strip().decode()
157	d = git_cache.ensure_rev_available(self.upstream, 'otherbranch', rev)
158	self.assertEqual(_git(d, 'show', '%s:foofile' % rev), b'foo')
159
160	def test_fetch_after_cache_deleted(self) -> None:
161	d1, rev1 = git_cache.fetch(self.upstream, 'master')
162	shutil.rmtree(d1)
163	d2, rev2 = git_cache.fetch(self.upstream, 'master')
164	self.assertEqual(d1, d2)
165	self.assertEqual(rev1, rev2)
166	self.assertEqual(_git(d2, 'show', '%s:file' % rev2), b'Contents')
167
168	def test_ensure_after_cache_deleted(self) -> None:
169	d1, rev = git_cache.fetch(self.upstream, 'master')
170	shutil.rmtree(d1)
171	d2 = git_cache.ensure_rev_available(self.upstream, 'master', rev)
172	self.assertEqual(d1, d2)
173	self.assertEqual(_git(d2, 'show', '%s:file' % rev), b'Contents')
174
175	def test_fetch_raises_on_amend(self) -> None:
176	git_cache.fetch(self.upstream, 'master')
177	_git(self.upstream, 'commit', '--amend', '-m', 'Amended')
178	with self.assertRaises(Exception):
179	git_cache.fetch(self.upstream, 'master')
180
181	def test_ensure_raises_on_amend(self) -> None:
182	git_cache.fetch(self.upstream, 'master')
183	_git(self.upstream, 'commit', '--amend', '-m', 'Amended')
184	rev = _git(self.upstream, 'log', '--format=%H', '-n1').strip().decode()
185	with self.assertRaises(Exception):
186	git_cache.ensure_rev_available(self.upstream, 'master', rev)
187
188
189	if __name__ == '__main__':
190	unittest.main()