[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', encoding='utf-8') 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'
        os.environ['FORCE_WARNING_TIME'] = '0'  # ONLY FOR TEST USE!

        self.tempdir = tempfile.TemporaryDirectory(prefix='git_cache_test-')
        self.upstream = os.path.join(self.tempdir.name, 'upstream')
        subprocess.run(['git', '-c', 'init.defaultBranch=main',
                       '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, 'main')
        self.assertEqual(_git(d, 'show', f'{rev}:file'), b'Contents')

    def test_fetch_twice(self) -> None:
        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', f'{rev2}:file'), b'Contents')

    def test_fetch_then_ensure(self) -> None:
        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', 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, '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', f'{rev2}:file'), b'Contents')

    def test_fetch_new_file(self) -> None:
        d1, rev1 = git_cache.fetch(self.upstream, 'main')
        _commit_file(self.upstream, 'foofile', 'foo', 'Foo')
        d2, rev2 = git_cache.fetch(self.upstream, 'main')
        self.assertEqual(d1, d2)
        self.assertNotEqual(rev1, rev2)
        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, '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, 'main', rev1)
        self.assertEqual(d1, d2)
        p = subprocess.run(
            ['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, 'main')
        self.tempdir.cleanup()
        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, '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, 'main', rev2)
        self.assertEqual(d1, d2)
        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, '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, 'main', 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, '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', 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, 'main')
        shutil.rmtree(d1)
        d2, rev2 = git_cache.fetch(self.upstream, 'main')
        self.assertEqual(d1, d2)
        self.assertEqual(rev1, rev2)
        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, 'main')
        shutil.rmtree(d1)
        d2 = git_cache.ensure_rev_available(self.upstream, 'main', rev)
        self.assertEqual(d1, d2)
        self.assertEqual(_git(d2, 'show', f'{rev}:file'), b'Contents')

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

    def test_ensure_raises_on_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()
        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, '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__':
    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:
a28f4bb7	30	with open(os.path.join(directory, filename), 'w', encoding='utf-8') as f:
bef7ce53 SW	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	54	os.environ['BACKOFF_MAX_TIME'] = '0'
083b90e7	55	os.environ['FORCE_WARNING_TIME'] = '0' # ONLY FOR TEST USE!
f36d5c6f	56
bef7ce53 SW	57	self.tempdir = tempfile.TemporaryDirectory(prefix='git_cache_test-')
bef7ce53 SW	58	self.upstream = os.path.join(self.tempdir.name, 'upstream')
fc9c09b3	59	subprocess.run(['git', '-c', 'init.defaultBranch=main',
21971f7f	60	'init', self.upstream], check=True)
bef7ce53 SW	61	_commit_file(self.upstream, 'file', 'Contents', 'First commit')
	62
	63	def tearDown(self) -> None:
8d251eca	64	_setenv('XDG_CACHE_HOME', self.old_XDG_CACHE_HOME)
35000f72	65	_setenv('XDG_DATA_HOME', self.old_XDG_DATA_HOME)
bef7ce53 SW	66
	67	self.tempdir.cleanup()
	68	self.xdgcache.cleanup()
	69
	70	def test_fetch(self) -> None:
fc9c09b3	71	d, rev = git_cache.fetch(self.upstream, 'main')
f580771a	72	self.assertEqual(_git(d, 'show', f'{rev}:file'), b'Contents')
bef7ce53 SW	73
bef7ce53 SW	74	def test_fetch_twice(self) -> None:
fc9c09b3	75	d1, rev1 = git_cache.fetch(self.upstream, 'main')
f580771a	76	self.assertEqual(_git(d1, 'show', f'{rev1}:file'), b'Contents')
fc9c09b3	77	d2, rev2 = git_cache.fetch(self.upstream, 'main')
bef7ce53 SW	78	self.assertEqual(d1, d2)
bef7ce53 SW	79	self.assertEqual(rev1, rev2)
f580771a	80	self.assertEqual(_git(d2, 'show', f'{rev2}:file'), b'Contents')
bef7ce53 SW	81
bef7ce53 SW	82	def test_fetch_then_ensure(self) -> None:
fc9c09b3	83	d1, rev = git_cache.fetch(self.upstream, 'main')
f580771a	84	self.assertEqual(_git(d1, 'show', f'{rev}:file'), b'Contents')
fc9c09b3	85	d2 = git_cache.ensure_rev_available(self.upstream, 'main', rev)
bef7ce53	86	self.assertEqual(d1, d2)
f580771a	87	self.assertEqual(_git(d2, 'show', f'{rev}:file'), b'Contents')
bef7ce53 SW	88
	89	def test_ensure_then_fetch(self) -> None:
	90	rev1 = _git(
	91	self.upstream, 'log', '--format=%H', '-n1').strip().decode()
fc9c09b3	92	d1 = git_cache.ensure_rev_available(self.upstream, 'main', rev1)
f580771a	93	self.assertEqual(_git(d1, 'show', f'{rev1}:file'), b'Contents')
fc9c09b3	94	d2, rev2 = git_cache.fetch(self.upstream, 'main')
bef7ce53 SW	95	self.assertEqual(d1, d2)
bef7ce53 SW	96	self.assertEqual(rev1, rev2)
f580771a	97	self.assertEqual(_git(d2, 'show', f'{rev2}:file'), b'Contents')
bef7ce53 SW	98
bef7ce53 SW	99	def test_fetch_new_file(self) -> None:
fc9c09b3	100	d1, rev1 = git_cache.fetch(self.upstream, 'main')
bef7ce53	101	_commit_file(self.upstream, 'foofile', 'foo', 'Foo')
fc9c09b3	102	d2, rev2 = git_cache.fetch(self.upstream, 'main')
bef7ce53 SW	103	self.assertEqual(d1, d2)
bef7ce53 SW	104	self.assertNotEqual(rev1, rev2)
f580771a	105	self.assertEqual(_git(d2, 'show', f'{rev2}:foofile'), b'foo')
bef7ce53 SW	106
bef7ce53 SW	107	def test_ensure_doesnt_fetch_new_file(self) -> None:
fc9c09b3	108	d1, rev1 = git_cache.fetch(self.upstream, 'main')
bef7ce53 SW	109	_commit_file(self.upstream, 'foofile', 'foo', 'Foo')
	110	rev2 = _git(
	111	self.upstream, 'log', '--format=%H', '-n1').strip().decode()
	112	self.assertNotEqual(rev1, rev2)
fc9c09b3	113	d2 = git_cache.ensure_rev_available(self.upstream, 'main', rev1)
bef7ce53 SW	114	self.assertEqual(d1, d2)
bef7ce53 SW	115	p = subprocess.run(
f580771a	116	['git', '-C', d2, 'show', f'{rev2}:foofile'], check=False)
bef7ce53 SW	117	self.assertNotEqual(p.returncode, 0)
	118
	119	def test_ensure_doesnt_fetch_from_deleted_upstream(self) -> None:
fc9c09b3	120	d1, rev = git_cache.fetch(self.upstream, 'main')
bef7ce53	121	self.tempdir.cleanup()
fc9c09b3	122	d2 = git_cache.ensure_rev_available(self.upstream, 'main', rev)
bef7ce53 SW	123	self.assertEqual(d1, d2)
	124
	125	def test_ensure_fetches_new_file(self) -> None:
fc9c09b3	126	d1, rev1 = git_cache.fetch(self.upstream, 'main')
bef7ce53 SW	127	_commit_file(self.upstream, 'foofile', 'foo', 'Foo')
	128	rev2 = _git(
	129	self.upstream, 'log', '--format=%H', '-n1').strip().decode()
	130	self.assertNotEqual(rev1, rev2)
fc9c09b3	131	d2 = git_cache.ensure_rev_available(self.upstream, 'main', rev2)
bef7ce53	132	self.assertEqual(d1, d2)
f580771a	133	self.assertEqual(_git(d2, 'show', f'{rev2}:foofile'), b'foo')
bef7ce53 SW	134
	135	def test_fetch_raises_on_invalid_repo(self) -> None:
	136	self.tempdir.cleanup()
	137	with self.assertRaises(Exception):
fc9c09b3	138	git_cache.fetch(self.upstream, 'main')
bef7ce53 SW	139
	140	def test_ensure_raises_on_invalid_repo(self) -> None:
	141	rev = _git(self.upstream, 'log', '--format=%H', '-n1').strip().decode()
	142	self.tempdir.cleanup()
	143	with self.assertRaises(Exception):
fc9c09b3	144	git_cache.ensure_rev_available(self.upstream, 'main', rev)
bef7ce53 SW	145
	146	def test_fetch_raises_on_invalid_ref(self) -> None:
	147	with self.assertRaises(Exception):
	148	git_cache.fetch(self.upstream, 'nobranch')
	149
	150	def test_ensure_raises_on_invalid_ref(self) -> None:
	151	rev = _git(self.upstream, 'log', '--format=%H', '-n1').strip().decode()
	152	with self.assertRaises(Exception):
	153	git_cache.ensure_rev_available(self.upstream, 'nobranch', rev)
	154
	155	def test_ensure_raises_on_invalid_rev(self) -> None:
	156	with self.assertRaises(Exception):
	157	git_cache.ensure_rev_available(
	158	self.upstream,
	159	'nobranch',
	160	'1234567890abcdef01234567890abcdef1234567')
	161
	162	def test_ensure_raises_on_rev_from_other_branch(self) -> None:
	163	_git(self.upstream, 'checkout', '-b', 'otherbranch')
	164	_commit_file(self.upstream, 'foofile', 'foo', 'Foo')
	165	rev = _git(self.upstream, 'log', '--format=%H', '-n1').strip().decode()
	166	with self.assertRaises(Exception):
fc9c09b3	167	git_cache.ensure_rev_available(self.upstream, 'main', rev)
bef7ce53 SW	168
	169	def test_ensure_other_branch(self) -> None:
	170	_git(self.upstream, 'checkout', '-b', 'otherbranch')
	171	_commit_file(self.upstream, 'foofile', 'foo', 'Foo')
	172	rev = _git(self.upstream, 'log', '--format=%H', '-n1').strip().decode()
	173	d = git_cache.ensure_rev_available(self.upstream, 'otherbranch', rev)
f580771a	174	self.assertEqual(_git(d, 'show', f'{rev}:foofile'), b'foo')
bef7ce53	175
eb638847 SW	176	def test_catch_up(self) -> None:
	177	_git(self.upstream, 'checkout', '-b', 'otherbranch')
	178	_commit_file(self.upstream, 'foofile', 'foo', 'Foo')
	179	rev = _git(self.upstream, 'log', '--format=%H', '-n1').strip().decode()
	180	d = git_cache.ensure_rev_available(self.upstream, 'otherbranch', rev)
f580771a	181	self.assertEqual(_git(d, 'show', f'{rev}:foofile'), b'foo')
fc9c09b3	182	_git(self.upstream, 'checkout', 'main')
eb638847	183	_git(self.upstream, 'merge', '--ff-only', 'otherbranch')
fc9c09b3	184	d = git_cache.ensure_rev_available(self.upstream, 'main', rev)
f580771a	185	self.assertEqual(_git(d, 'show', f'{rev}:foofile'), b'foo')
eb638847	186
bef7ce53	187	def test_fetch_after_cache_deleted(self) -> None:
fc9c09b3	188	d1, rev1 = git_cache.fetch(self.upstream, 'main')
bef7ce53	189	shutil.rmtree(d1)
fc9c09b3	190	d2, rev2 = git_cache.fetch(self.upstream, 'main')
bef7ce53 SW	191	self.assertEqual(d1, d2)
bef7ce53 SW	192	self.assertEqual(rev1, rev2)
f580771a	193	self.assertEqual(_git(d2, 'show', f'{rev2}:file'), b'Contents')
bef7ce53 SW	194
bef7ce53 SW	195	def test_ensure_after_cache_deleted(self) -> None:
fc9c09b3	196	d1, rev = git_cache.fetch(self.upstream, 'main')
bef7ce53	197	shutil.rmtree(d1)
fc9c09b3	198	d2 = git_cache.ensure_rev_available(self.upstream, 'main', rev)
bef7ce53	199	self.assertEqual(d1, d2)
f580771a	200	self.assertEqual(_git(d2, 'show', f'{rev}:file'), b'Contents')
bef7ce53 SW	201
bef7ce53 SW	202	def test_fetch_raises_on_amend(self) -> None:
fc9c09b3	203	git_cache.fetch(self.upstream, 'main')
bef7ce53 SW	204	_git(self.upstream, 'commit', '--amend', '-m', 'Amended')
bef7ce53 SW	205	with self.assertRaises(Exception):
fc9c09b3	206	git_cache.fetch(self.upstream, 'main')
bef7ce53 SW	207
bef7ce53 SW	208	def test_ensure_raises_on_amend(self) -> None:
fc9c09b3	209	git_cache.fetch(self.upstream, 'main')
bef7ce53 SW	210	_git(self.upstream, 'commit', '--amend', '-m', 'Amended')
	211	rev = _git(self.upstream, 'log', '--format=%H', '-n1').strip().decode()
	212	with self.assertRaises(Exception):
fc9c09b3	213	git_cache.ensure_rev_available(self.upstream, 'main', rev)
bef7ce53	214
35000f72	215	def test_fetch_raises_on_amend_after_cache_deleted(self) -> None:
fc9c09b3	216	d, _ = git_cache.fetch(self.upstream, 'main')
35000f72 SW	217	shutil.rmtree(d)
	218	_git(self.upstream, 'commit', '--amend', '-m', 'Amended')
	219	with self.assertRaises(Exception):
fc9c09b3	220	git_cache.fetch(self.upstream, 'main')
35000f72 SW	221
35000f72 SW	222	def test_ensure_raises_on_amend_after_cache_deleted(self) -> None:
fc9c09b3	223	d, _ = git_cache.fetch(self.upstream, 'main')
35000f72 SW	224	shutil.rmtree(d)
	225	_git(self.upstream, 'commit', '--amend', '-m', 'Amended')
	226	rev = _git(self.upstream, 'log', '--format=%H', '-n1').strip().decode()
	227	with self.assertRaises(Exception):
fc9c09b3	228	git_cache.ensure_rev_available(self.upstream, 'main', rev)
35000f72	229
083b90e7	230	def test_force_fetch_after_amend(self) -> None:
fc9c09b3	231	git_cache.fetch(self.upstream, 'main')
083b90e7	232	_git(self.upstream, 'commit', '--amend', '-m', 'Amended')
fc9c09b3	233	git_cache.fetch(self.upstream, 'main', force=True)
083b90e7 SW	234
083b90e7 SW	235	def test_force_ensure_after_amend(self) -> None:
fc9c09b3	236	git_cache.fetch(self.upstream, 'main')
083b90e7 SW	237	_git(self.upstream, 'commit', '--amend', '-m', 'Amended')
	238	rev = _git(self.upstream, 'log', '--format=%H', '-n1').strip().decode()
	239	git_cache.ensure_rev_available(
fc9c09b3	240	self.upstream, 'main', rev, force=True)
083b90e7 SW	241
083b90e7 SW	242	def test_force_fetch_after_amend_and_cache_delete(self) -> None:
fc9c09b3	243	d, _ = git_cache.fetch(self.upstream, 'main')
083b90e7 SW	244	shutil.rmtree(d)
083b90e7 SW	245	_git(self.upstream, 'commit', '--amend', '-m', 'Amended')
fc9c09b3	246	git_cache.fetch(self.upstream, 'main', force=True)
083b90e7 SW	247
083b90e7 SW	248	def test_force_ensure_after_amend_and_cache_delete(self) -> None:
fc9c09b3	249	d, _ = git_cache.fetch(self.upstream, 'main')
083b90e7 SW	250	shutil.rmtree(d)
	251	_git(self.upstream, 'commit', '--amend', '-m', 'Amended')
	252	rev = _git(self.upstream, 'log', '--format=%H', '-n1').strip().decode()
	253	git_cache.ensure_rev_available(
fc9c09b3	254	self.upstream, 'main', rev, force=True)
083b90e7	255
bef7ce53 SW	256
	257	if __name__ == '__main__':
	258	unittest.main()