Release 1.5.0

[git-cache] / test_git_cache.py

# 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 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(  # 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')
        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(  # pylint: disable=consider-using-with
            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)

    def test_fetch_tag(self) -> None:
        _git(self.upstream, 'tag', 'v1.2.3')
        d, rev = git_cache.fetch(self.upstream, 'tag v1.2.3')
        self.assertEqual(_git(d, 'show', f'{rev}:file'), b'Contents')

    def test_ensure_tag(self) -> None:
        rev = _git(self.upstream, 'log', '--format=%H', '-n1').strip().decode()
        _git(self.upstream, 'tag', 'v1.2.3')
        d = git_cache.ensure_rev_available(self.upstream, 'tag v1.2.3', rev)
        self.assertEqual(_git(d, 'show', f'{rev}:file'), b'Contents')

    def test_fetch_annotated_tag(self) -> None:
        _git(self.upstream, 'tag', '--annotate', '-m', 'Tag', 'v1.2.3')
        d, rev = git_cache.fetch(self.upstream, 'tag v1.2.3')
        self.assertEqual(_git(d, 'show', f'{rev}:file'), b'Contents')

    def test_ensure_annotated_tag(self) -> None:
        rev = _git(self.upstream, 'log', '--format=%H', '-n1').strip().decode()
        _git(self.upstream, 'tag', '--annotate', '-m', 'Tag', 'v1.2.3')
        d = git_cache.ensure_rev_available(self.upstream, 'tag v1.2.3', rev)
        self.assertEqual(_git(d, 'show', f'{rev}:file'), b'Contents')


if __name__ == '__main__':
    unittest.main()