# -*- coding: utf-8 -*- # # Copyright (C) 2014-2023 Edgewall Software # All rights reserved. # # This software is licensed as described in the file COPYING, which # you should have received as part of this distribution. The terms # are also available at https://trac.edgewall.org/wiki/TracLicense. # # This software consists of voluntary contributions made by many # individuals. For the exact contribution history, see the revision # history and logs, available at https://trac.edgewall.org/log/. import io import os import sys import unittest from datetime import datetime, timedelta from subprocess import DEVNULL, PIPE, Popen from trac.core import TracError from trac.test import EnvironmentStub, MockRequest, locate, makeSuite, \ mkdtemp, rmtree from trac.util import create_file from trac.util.compat import close_fds from trac.util.datefmt import to_timestamp, utc from trac.util.text import to_utf8 from trac.versioncontrol.api import Changeset, DbRepositoryProvider, \ InvalidRepository, Node, \ NoSuchChangeset, NoSuchNode, \ RepositoryManager from trac.versioncontrol.web_ui.browser import BrowserModule from trac.versioncontrol.web_ui.log import LogModule from tracopt.versioncontrol.git.PyGIT import StorageFactory from tracopt.versioncontrol.git.git_fs import GitCachedRepository, \ GitRepository, \ GitwebProjectsRepositoryProvider class GitCommandMixin(object): git_bin = locate('git') def _git_commit(self, *args, **kwargs): env = kwargs.get('env') or os.environ.copy() if 'date' in kwargs: self._set_committer_date(env, kwargs.pop('date')) args = ('commit',) + args kwargs['env'] = env return self._git(*args, **kwargs) def _spawn_git(self, *args, **kwargs): args = (self.git_bin,) + args kwargs.setdefault('stdin', DEVNULL) kwargs.setdefault('stdout', PIPE) kwargs.setdefault('stderr', PIPE) kwargs.setdefault('cwd', self.repos_path) return Popen(args, close_fds=close_fds, **kwargs) def _git(self, *args, **kwargs): args = ('-c', 'init.defaultBranch=master', '-c', 'protocol.file.allow=always') + args with self._spawn_git(*args, **kwargs) as proc: stdout, stderr = proc.communicate() self.assertEqual(0, proc.returncode, 'git exits with %r, args %r, kwargs %r, stdout %r, ' 'stderr %r' % (proc.returncode, args, kwargs, stdout, stderr)) return proc def _git_fast_import(self, data, **kwargs): if isinstance(data, str): data = data.encode('utf-8') with self._spawn_git('fast-import', stdin=PIPE, **kwargs) as proc: stdout, stderr = proc.communicate(input=data) self.assertEqual(0, proc.returncode, 'git exits with %r, stdout %r, stderr %r' % (proc.returncode, stdout, stderr)) def _git_date_format(self, dt): if dt.tzinfo is None: dt = dt.replace(tzinfo=utc) offset = dt.utcoffset() secs = offset.days * 3600 * 24 + offset.seconds hours, rem = divmod(abs(secs), 3600) return '%d %c%02d:%02d' % (to_timestamp(dt), '-' if secs < 0 else '+', hours, rem // 60) def _set_committer_date(self, env, dt): if not isinstance(dt, str): if dt.tzinfo is None: dt = dt.replace(tzinfo=utc) dt = self._git_date_format(dt) env['GIT_COMMITTER_DATE'] = dt env['GIT_AUTHOR_DATE'] = dt class BaseTestCase(unittest.TestCase, GitCommandMixin): gitignore_content = b'*.py[co]\n' \ b'\312\326\267\347\307\331.txt\n' def setUp(self): self.env = EnvironmentStub() self.tmpdir = mkdtemp() self.repos_path = os.path.join(self.tmpdir, 'gitrepos') os.mkdir(self.repos_path) if self.git_bin: self.env.config.set('git', 'git_bin', self.git_bin) def tearDown(self): for repos in self._repomgr.get_real_repositories(): repos.close() self._repomgr.reload_repositories() StorageFactory._clean() self.env.reset_db() if os.path.isdir(self.tmpdir): rmtree(self.tmpdir) @property def _repomgr(self): return RepositoryManager(self.env) @property def _dbrepoprov(self): return DbRepositoryProvider(self.env) def _add_repository(self, reponame='gitrepos', bare=False, path=None): if path is None: path = self.repos_path if not bare: path = os.path.join(path, '.git') self._dbrepoprov.add_repository(reponame, path, 'git') def _git_init(self, data=True, bare=False, **kwargs): if bare: self._git('init', '--bare', **kwargs) else: self._git('init', **kwargs) if not bare and data: self._git('config', 'user.name', 'Joe', **kwargs) self._git('config', 'user.email', 'joe@example.com', **kwargs) create_file(os.path.join(self.repos_path, '.gitignore'), self.gitignore_content, 'wb') self._git('add', '.gitignore', **kwargs) self._git_commit('-a', '-m', 'test', date=datetime(2001, 1, 29, 16, 39, 56), **kwargs) class SanityCheckingTestCase(BaseTestCase): def test_bare(self): self._git_init(bare=True) self._dbrepoprov.add_repository('gitrepos', self.repos_path, 'git') self._repomgr.get_repository('gitrepos') def test_non_bare(self): self._git_init(bare=False) self._dbrepoprov.add_repository('gitrepos.1', os.path.join(self.repos_path, '.git'), 'git') self._repomgr.get_repository('gitrepos.1') self._dbrepoprov.add_repository('gitrepos.2', self.repos_path, 'git') self._repomgr.get_repository('gitrepos.2') def test_no_head_file(self): self._git_init(bare=True) os.unlink(os.path.join(self.repos_path, 'HEAD')) self._dbrepoprov.add_repository('gitrepos', self.repos_path, 'git') self.assertRaises(TracError, self._repomgr.get_repository, 'gitrepos') def test_no_objects_dir(self): self._git_init(bare=True) rmtree(os.path.join(self.repos_path, 'objects')) self._dbrepoprov.add_repository('gitrepos', self.repos_path, 'git') self.assertRaises(TracError, self._repomgr.get_repository, 'gitrepos') def test_no_refs_dir(self): self._git_init(bare=True) rmtree(os.path.join(self.repos_path, 'refs')) self._dbrepoprov.add_repository('gitrepos', self.repos_path, 'git') self.assertRaises(TracError, self._repomgr.get_repository, 'gitrepos') class PersistentCacheTestCase(BaseTestCase): def test_persistent(self): self.env.config.set('git', 'persistent_cache', 'enabled') self._git_init() self._add_repository() youngest = self._repository.youngest_rev self._repomgr.reload_repositories() # clear repository cache self._commit(datetime(2014, 1, 29, 16, 44, 54, 0, utc)) self.assertEqual(youngest, self._repository.youngest_rev) self._repository.sync() self.assertNotEqual(youngest, self._repository.youngest_rev) def test_non_persistent(self): self.env.config.set('git', 'persistent_cache', 'disabled') self._git_init() self._add_repository() youngest = self._repository.youngest_rev self._repomgr.reload_repositories() # clear repository cache self._commit(datetime(2014, 1, 29, 16, 44, 54, 0, utc)) youngest_2 = self._repository.youngest_rev self.assertNotEqual(youngest, youngest_2) self._repository.sync() self.assertNotEqual(youngest, self._repository.youngest_rev) self.assertEqual(youngest_2, self._repository.youngest_rev) def _commit(self, date): gitignore = os.path.join(self.repos_path, '.gitignore') create_file(gitignore, date.isoformat()) self._git_commit('-a', '-m', date.isoformat(), date=date) @property def _repository(self): return self._repomgr.get_repository('gitrepos') class HistoryTimeRangeTestCase(BaseTestCase): def test_without_cache(self): self._test_timerange('disabled') def test_with_cache(self): self._test_timerange('enabled') def _test_timerange(self, cached_repository): self.env.config.set('git', 'cached_repository', cached_repository) self._git_init() filename = os.path.join(self.repos_path, '.gitignore') start = datetime(2000, 1, 1, 0, 0, 0, 0, utc) ts = datetime(2014, 2, 5, 15, 24, 6, 0, utc) for idx in range(3): create_file(filename, 'commit-%d.txt' % idx) self._git_commit('-a', '-m', 'commit %d' % idx, date=ts) self._add_repository() repos = self._repomgr.get_repository('gitrepos') repos.sync() revs = [repos.youngest_rev] while True: parents = repos.parent_revs(revs[-1]) if not parents: break revs.extend(parents) self.assertEqual(4, len(revs)) csets = list(repos.get_changesets(start, ts)) self.assertEqual(1, len(csets)) self.assertEqual(revs[-1], csets[0].rev) # is oldest rev csets = list(repos.get_changesets(start, ts + timedelta(seconds=1))) self.assertEqual(revs, [cset.rev for cset in csets]) class GitNormalTestCase(BaseTestCase): def test_get_node(self): self.env.config.set('git', 'persistent_cache', 'false') self.env.config.set('git', 'cached_repository', 'false') self._git_init() self._add_repository() repos = self._repomgr.get_repository('gitrepos') rev = repos.youngest_rev self.assertIsInstance(rev, str) self.assertEqual(40, len(rev)) self.assertEqual(rev, repos.get_node('/').rev) self.assertEqual(rev, repos.get_node('/', rev[:7]).rev) self.assertEqual(rev, repos.get_node('/.gitignore').rev) self.assertEqual(rev, repos.get_node('/.gitignore', rev[:7]).rev) self.assertEqual(self.gitignore_content, repos.get_node('/.gitignore').get_content().read()) self.assertEqual(len(self.gitignore_content), repos.get_node('/.gitignore').content_length) self.assertRaises(NoSuchNode, repos.get_node, '/non-existent') self.assertRaises(NoSuchNode, repos.get_node, '/non-existent', rev[:7]) self.assertRaises(NoSuchNode, repos.get_node, '/non-existent', rev) self.assertRaises(NoSuchChangeset, repos.get_node, '/', 'invalid-revision') self.assertRaises(NoSuchChangeset, repos.get_node, '/.gitignore', 'invalid-revision') self.assertRaises(NoSuchChangeset, repos.get_node, '/non-existent', 'invalid-revision') self._git('branch', 'tïckét10605', 'master') repos.sync() self.assertEqual(rev, repos.get_node('/', 'tïckét10605').rev) self.assertEqual(rev, repos.get_node('/.gitignore', 'tïckét10605').rev) def _test_on_empty_repos(self, cached_repository): self.env.config.set('git', 'persistent_cache', 'false') self.env.config.set('git', 'cached_repository', 'true' if cached_repository else 'false') self._git_init(data=False, bare=True) self._add_repository(bare=True) repos = self._repomgr.get_repository('gitrepos') if cached_repository: # call sync() thrice with empty repository (#11851) for i in range(3): repos.sync() rows = self.env.db_query("SELECT value FROM repository " "WHERE id=%s AND name=%s", (repos.id, 'youngest_rev')) self.assertEqual('', rows[0][0]) else: repos.sync() youngest_rev = repos.youngest_rev self.assertIsNone(youngest_rev) self.assertIsNone(repos.oldest_rev) self.assertIsNone(repos.normalize_rev('')) self.assertIsNone(repos.normalize_rev(None)) self.assertIsNone(repos.display_rev('')) self.assertIsNone(repos.display_rev(None)) self.assertIsNone(repos.short_rev('')) self.assertIsNone(repos.short_rev(None)) node = repos.get_node('/', youngest_rev) self.assertEqual([], list(node.get_entries())) self.assertEqual([], list(node.get_history())) self.assertRaises(NoSuchNode, repos.get_node, '/path', youngest_rev) req = MockRequest(self.env, path_info='/browser/gitrepos') browser_mod = BrowserModule(self.env) self.assertTrue(browser_mod.match_request(req)) rv = browser_mod.process_request(req) self.assertEqual('browser.html', rv[0]) self.assertIsNone(rv[1]['rev']) req = MockRequest(self.env, path_info='/log/gitrepos') log_mod = LogModule(self.env) self.assertTrue(log_mod.match_request(req)) rv = log_mod.process_request(req) self.assertEqual('revisionlog.html', rv[0]) self.assertEqual([], rv[1]['items']) def test_on_empty_and_cached_repos(self): self._test_on_empty_repos(True) def test_on_empty_and_non_cached_repos(self): self._test_on_empty_repos(False) class GitRepositoryTestCase(BaseTestCase): cached_repository = 'disabled' def setUp(self): BaseTestCase.setUp(self) self.env.config.set('git', 'cached_repository', self.cached_repository) def _create_merge_commit(self): for idx, branch in enumerate(('alpha', 'beta')): self._git('checkout', '-b', branch, 'master') for n in range(2): filename = 'file-%s-%d.txt' % (branch, n) create_file(os.path.join(self.repos_path, filename)) self._git('add', filename) self._git_commit('-a', '-m', filename, date=datetime(2014, 2, 2, 17, 12, n * 2 + idx)) self._git('checkout', 'alpha') self._git('merge', '-m', 'Merge branch "beta" to "alpha"', 'beta') def test_invalid_path_raises(self): def try_init(reponame): params = {'name': reponame} with self.assertRaises(InvalidRepository) as cm: GitRepository(self.env, '/the/invalid/path', params, self.env.log) return cm.exception e = try_init('') self.assertEqual('"(default)" is not readable or not a Git ' 'repository.', str(e)) e = try_init('therepos') self.assertEqual('"therepos" is not readable or not a Git repository.', str(e)) def test_repository_instance(self): self._git_init() self._add_repository('gitrepos') self.assertEqual(GitRepository, type(self._repomgr.get_repository('gitrepos'))) def test_reset_head(self): self._git_init() create_file(os.path.join(self.repos_path, 'file.txt'), 'text') self._git('add', 'file.txt') self._git_commit('-a', '-m', 'test', date=datetime(2014, 2, 2, 17, 12, 18)) self._add_repository('gitrepos') repos = self._repomgr.get_repository('gitrepos') repos.sync() youngest_rev = repos.youngest_rev entries = list(repos.get_node('').get_history()) self.assertEqual(2, len(entries)) self.assertEqual('', entries[0][0]) self.assertEqual(youngest_rev, entries[0][1]) self.assertEqual(Changeset.EDIT, entries[0][2]) self.assertEqual('', entries[1][0]) self.assertIsInstance(entries[1][1], str) self.assertEqual(Changeset.ADD, entries[1][2]) self.assertEqual(entries[1][1], repos.previous_rev(youngest_rev, '')) self.assertEqual(None, repos.next_rev(youngest_rev, '')) self.assertEqual(None, repos.previous_rev(entries[1][1], '')) self.assertEqual(youngest_rev, repos.next_rev(entries[1][1], '')) self._git('reset', '--hard', 'HEAD~') repos.sync() new_entries = list(repos.get_node('').get_history()) self.assertEqual(1, len(new_entries)) self.assertEqual(new_entries[0], entries[1]) self.assertNotEqual(youngest_rev, repos.youngest_rev) self.assertEqual(new_entries[0][1], repos.youngest_rev) self.assertEqual(None, repos.next_rev(repos.youngest_rev, '')) self.assertEqual(None, repos.previous_rev(repos.youngest_rev, '')) def test_tags(self): self._git_init() self._add_repository('gitrepos') repos = self._repomgr.get_repository('gitrepos') repos.sync() self.assertEqual(['master'], self._get_quickjump_names(repos)) self._git('tag', 'v1.0', 'master') # add tag repos.sync() self.assertEqual(['master', 'v1.0'], self._get_quickjump_names(repos)) self._git('tag', '-d', 'v1.0') # delete tag repos.sync() self.assertEqual(['master'], self._get_quickjump_names(repos)) def test_branchs(self): self._git_init() self._add_repository('gitrepos') repos = self._repomgr.get_repository('gitrepos') repos.sync() self.assertEqual(['master'], self._get_quickjump_names(repos)) self._git('branch', 'alpha', 'master') # add branch repos.sync() self.assertEqual(['master', 'alpha'], self._get_quickjump_names(repos)) self._git('branch', '-m', 'alpha', 'beta') # rename branch repos.sync() self.assertEqual(['master', 'beta'], self._get_quickjump_names(repos)) self._git('branch', '-D', 'beta') # delete branch repos.sync() self.assertEqual(['master'], self._get_quickjump_names(repos)) def test_changeset_branches_tags(self): self._git_init() self._git('tag', '0.0.1', 'master') self._git('tag', '-m', 'Root commit', 'initial', 'master') self._git('branch', 'root', 'master') self._git('checkout', '-b', 'dev', 'master') self._git_commit('-m', 'Summary', '--allow-empty') self._git('tag', '0.1.0dev', 'dev') self._git('tag', '0.1.0a', 'dev') self._add_repository('gitrepos') repos = self._repomgr.get_repository('gitrepos') repos.sync() def get_branches(repos, rev): rev = repos.normalize_rev(rev) return list(repos.get_changeset(rev).get_branches()) def get_tags(repos, rev): rev = repos.normalize_rev(rev) return list(repos.get_changeset(rev).get_tags()) self.assertEqual([('dev', False), ('master', True), ('root', True)], get_branches(repos, '0.0.1')) self.assertEqual([('dev', True)], get_branches(repos, '0.1.0dev')) self.assertEqual(['0.0.1', 'initial'], get_tags(repos, '0.0.1')) self.assertEqual(['0.0.1', 'initial'], get_tags(repos, 'initial')) self.assertEqual(['0.1.0a', '0.1.0dev'], get_tags(repos, '0.1.0dev')) def test_parent_child_revs(self): self._git_init() self._git('branch', 'initial') # root commit self._create_merge_commit() self._git('branch', 'latest') self._add_repository('gitrepos') repos = self._repomgr.get_repository('gitrepos') repos.sync() rev = repos.normalize_rev('initial') children = repos.child_revs(rev) self.assertEqual(2, len(children), 'child_revs: %r' % children) parents = repos.parent_revs(rev) self.assertEqual(0, len(parents), 'parent_revs: %r' % parents) self.assertEqual(1, len(repos.child_revs(children[0]))) self.assertEqual(1, len(repos.child_revs(children[1]))) self.assertEqual([('.gitignore', Node.FILE, Changeset.ADD, None, None)], sorted(repos.get_changeset(rev).get_changes())) rev = repos.normalize_rev('latest') cset = repos.get_changeset(rev) children = repos.child_revs(rev) self.assertEqual(0, len(children), 'child_revs: %r' % children) parents = repos.parent_revs(rev) self.assertEqual(2, len(parents), 'parent_revs: %r' % parents) self.assertEqual(1, len(repos.parent_revs(parents[0]))) self.assertEqual(1, len(repos.parent_revs(parents[1]))) # check the differences against the first parent def fn_repos_changes(entry): old_node, new_node, kind, change = entry if old_node: old_path, old_rev = old_node.path, old_node.rev else: old_path, old_rev = None, None return new_node.path, kind, change, old_path, old_rev self.assertEqual(sorted(map(fn_repos_changes, repos.get_changes('/', parents[0], '/', rev))), sorted(cset.get_changes())) _data_annotations = b"""\ blob mark :1 data 14 one two three reset refs/heads/master commit refs/heads/master mark :2 author Joe 1467172510 +0000 committer Joe 1467172510 +0000 data 6 blame M 100644 :1 test.txt blob mark :3 data 49 one two three four five six seven eight nine ten commit refs/heads/master mark :4 author Joe 1467172511 +0000 committer Joe 1467172511 +0000 data 10 add lines from :2 M 100644 :3 test.txt blob mark :5 data 40 one two 3 four five 6 seven eight 9 ten commit refs/heads/master mark :6 author Joe 1467172512 +0000 committer Joe 1467172512 +0000 data 13 modify lines from :4 M 100644 :5 test.txt reset refs/heads/master from :6 """ def test_get_annotations(self): self._git_init(data=False) self._git_fast_import(self._data_annotations) self._add_repository('gitrepos') repos = self._repomgr.get_repository('gitrepos') repos.sync() rev1 = 'a7efe353630d02139f255220d71b76fa68eb7132' # root commit rev2 = 'f928d1b36b8bedf64bcf08667428fdcccf36b21b' rev3 = '279a097f111c7cb1ef0b9da39735188051fd4f69' # HEAD self.assertEqual([rev1] * 3, repos.get_node('test.txt', rev1).get_annotations()) self.assertEqual([rev1] * 3 + [rev2] * 7, repos.get_node('test.txt', rev2).get_annotations()) expected = [rev1, rev1, rev3, rev2, rev2, rev3, rev2, rev2, rev3, rev2] self.assertEqual(expected, repos.get_node('test.txt', rev3).get_annotations()) self.assertEqual(expected, repos.get_node('test.txt', 'HEAD').get_annotations()) self.assertEqual(expected, repos.get_node('test.txt').get_annotations()) # * 79dff4ccf842f8e2d2da2ee3e7a2149df63b099b Merge branch 'A' # |\ # | * 86387120095e9e43573bce61b9da70a8c5d1c1b9 Merge branch 'B' into A # | |\ # | | * 64e12f96b6b3040cd9edc225734ab2b26a03758b Changed a1 # | * | 67fdcf11e2d083b123b9a79be4fce0600f313f81 Changed a2 # * | | 42fbe758709b2a65aba33e56b2f53cd126c190e3 Changed b2 # | |/ # |/| # * | 24d94dc08eb77438e4ead192b3f7d1c7bdf1a9e1 Changed b2 # * | 998bf23843c8fd982bbc23f88ec33c4d08114557 Changed b1 # |/ # * c5b01c74e125aa034a1d4ae31dc16f1897a73779 First commit _data_iter_nodes = b"""\ blob mark :1 data 2 a1 blob mark :2 data 2 a2 blob mark :3 data 2 b1 blob mark :4 data 2 b2 reset refs/heads/A commit refs/heads/A mark :5 author Joe 1470744252 +0000 committer Joe 1470744252 +0000 data 13 First commit M 100644 :1 A/a1.txt M 100644 :2 A/a2.txt M 100644 :3 B/b1.txt M 100644 :4 B/b2.txt blob mark :6 data 4 b1-1 commit refs/heads/master mark :7 author Joe 1470744253 +0000 committer Joe 1470744253 +0000 data 11 Changed b1 from :5 M 100644 :6 B/b1.txt blob mark :8 data 4 b2-1 commit refs/heads/master mark :9 author Joe 1470744254 +0000 committer Joe 1470744254 +0000 data 11 Changed b2 from :7 M 100644 :8 B/b2.txt blob mark :10 data 4 b2-2 commit refs/heads/master mark :11 author Joe 1470744255 +0000 committer Joe 1470744255 +0000 data 11 Changed b2 from :9 M 100644 :10 B/b2.txt blob mark :12 data 4 a2-1 commit refs/heads/A mark :13 author Joe 1470744256 +0000 committer Joe 1470744256 +0000 data 11 Changed a2 from :5 M 100644 :12 A/a2.txt blob mark :14 data 4 a1-1 commit refs/heads/B mark :15 author Joe 1470744257 +0000 committer Joe 1470744257 +0000 data 11 Changed a1 from :9 M 100644 :14 A/a1.txt commit refs/heads/A mark :16 author Joe 1470744258 +0000 committer Joe 1470744258 +0000 data 24 Merge branch 'B' into A from :13 merge :15 M 100644 :14 A/a1.txt M 100644 :6 B/b1.txt M 100644 :8 B/b2.txt commit refs/heads/master mark :17 author Joe 1470744259 +0000 committer Joe 1470744259 +0000 data 17 Merge branch 'A' from :11 merge :16 M 100644 :14 A/a1.txt M 100644 :12 A/a2.txt reset refs/heads/master from :17 """ def test_iter_nodes(self): self._git_init(data=False) self._git_fast_import(self._data_iter_nodes) self._add_repository('gitrepos') repos = self._repomgr.get_repository('gitrepos') repos.sync() mod = BrowserModule(self.env) root_node = repos.get_node('') nodes = list(mod._iter_nodes(root_node)) self.assertEqual(['79dff4ccf842f8e2d2da2ee3e7a2149df63b099b'] * 7, [node.rev for node in nodes]) self.assertEqual([ ('79dff4ccf842f8e2d2da2ee3e7a2149df63b099b', ''), ('64e12f96b6b3040cd9edc225734ab2b26a03758b', 'A'), ('64e12f96b6b3040cd9edc225734ab2b26a03758b', 'A/a1.txt'), ('67fdcf11e2d083b123b9a79be4fce0600f313f81', 'A/a2.txt'), ('42fbe758709b2a65aba33e56b2f53cd126c190e3', 'B'), ('998bf23843c8fd982bbc23f88ec33c4d08114557', 'B/b1.txt'), ('42fbe758709b2a65aba33e56b2f53cd126c190e3', 'B/b2.txt'), ], [(node.created_rev, node.path) for node in nodes]) root_node = repos.get_node('', '86387120095e9e43573bce61b9da70a8c5d1c1b9') nodes = list(mod._iter_nodes(root_node)) self.assertEqual(['86387120095e9e43573bce61b9da70a8c5d1c1b9'] * 7, [node.rev for node in nodes]) self.assertEqual([ ('86387120095e9e43573bce61b9da70a8c5d1c1b9', ''), ('64e12f96b6b3040cd9edc225734ab2b26a03758b', 'A'), ('64e12f96b6b3040cd9edc225734ab2b26a03758b', 'A/a1.txt'), ('67fdcf11e2d083b123b9a79be4fce0600f313f81', 'A/a2.txt'), ('24d94dc08eb77438e4ead192b3f7d1c7bdf1a9e1', 'B'), ('998bf23843c8fd982bbc23f88ec33c4d08114557', 'B/b1.txt'), ('24d94dc08eb77438e4ead192b3f7d1c7bdf1a9e1', 'B/b2.txt'), ], [(node.created_rev, node.path) for node in nodes]) root_commit = 'c5b01c74e125aa034a1d4ae31dc16f1897a73779' root_node = repos.get_node('', root_commit) nodes = list(mod._iter_nodes(root_node)) self.assertEqual([root_commit] * 7, [node.rev for node in nodes]) self.assertEqual([ (root_commit, ''), (root_commit, 'A'), (root_commit, 'A/a1.txt'), (root_commit, 'A/a2.txt'), (root_commit, 'B'), (root_commit, 'B/b1.txt'), (root_commit, 'B/b2.txt'), ], [(node.created_rev, node.path) for node in nodes]) def test_colon_character_in_filename(self): self._git_init(data=False) self._git_fast_import(self._data_colon_character_in_filename) self._add_repository('gitrepos') repos = self._repomgr.get_repository('gitrepos') repos.sync() rev1 = '382e1e6b85ba20ce8a84af1a875eaa50b8e1e092' # root commit rev2 = 'd8001832aad079f85a39a54a388a8b15fe31093d' ADD = Changeset.ADD MOVE = Changeset.MOVE FILE = Node.FILE cset = repos.get_changeset(rev1) self.assertEqual({('0100644', FILE, ADD, None, None), ('0100644.txt', FILE, ADD, None, None), (':100644', FILE, ADD, None, None), (':100644.txt', FILE, ADD, None, None), ('a100644', FILE, ADD, None, None), ('a100644.txt', FILE, ADD, None, None)}, set(cset.get_changes())) cset = repos.get_changeset(rev2) self.assertEqual({(':100666', FILE, MOVE, ':100644', rev1)}, set(cset.get_changes())) _data_colon_character_in_filename = b"""\ blob mark :1 data 0 blob mark :2 data 16 ............... reset refs/heads/master commit refs/heads/master mark :3 author Joe 1491387182 +0000 committer Joe 1491387182 +0000 data 9 (#12758) M 100644 :1 0100644.txt M 100644 :1 0100644 M 100644 :1 :100644.txt M 100644 :2 :100644 M 100644 :1 a100644.txt M 100644 :1 a100644 commit refs/heads/master mark :4 author Joe 1491387183 +0000 committer Joe 1491387183 +0000 data 16 (#12758) rename from :3 D :100644 M 100644 :2 :100666 reset refs/heads/master from :4 """ def test_submodule(self): subrepos_path = os.path.join(self.tmpdir, 'subrepos') submodule_dir = os.path.join(self.repos_path, 'sub') os.mkdir(subrepos_path) self._git_init(data=False, bare=True, cwd=subrepos_path) self._git_fast_import(self._data_submodule, cwd=subrepos_path) submodule_rev1 = '3e733d786b3529d750ee39edacea2f1c4daadca4' self._git_init() self._git('submodule', 'add', subrepos_path, 'sub') self._git('checkout', submodule_rev1, cwd=submodule_dir) self._git('add', '.gitmodules') self._git_commit('-a', '-m', 'init submodule') self._git('tag', 'v1', 'master') self._add_repository('gitrepos') repos = self._repomgr.get_repository('gitrepos') repos.sync() ADD = Changeset.ADD EDIT = Changeset.EDIT FILE = Node.FILE DIRECTORY = Node.DIRECTORY rev1 = repos.normalize_rev('v1') cset1 = repos.get_changeset(rev1) self.assertEqual([('.gitmodules', FILE, ADD, None, None), ('sub', DIRECTORY, ADD, None, None)], sorted(cset1.get_changes())) node1 = repos.get_node('sub', rev1) self.assertIsNone(node1.get_content()) self.assertIsNone(node1.get_content_length()) self.assertEqual(DIRECTORY, node1.kind) self.assertEqual({'mode': '160000', 'commit': submodule_rev1}, node1.get_properties()) self.assertEqual([], list(node1.get_entries())) submodule_rev2 = '409058dc98500b5685c52a091cc9f44f3975113e' self._git('checkout', submodule_rev2, cwd=submodule_dir) self._git_commit('-a', '-m', 'change rev of the submodule') self._git('tag', 'v2', 'master') repos.sync() rev2 = repos.normalize_rev('v2') cset2 = repos.get_changeset(rev2) self.assertEqual([('sub', DIRECTORY, EDIT, 'sub', rev1)], sorted(cset2.get_changes())) node2 = repos.get_node('sub', rev2) self.assertEqual({'mode': '160000', 'commit': submodule_rev2}, node2.get_properties()) _data_submodule = b"""\ blob mark :1 data 0 blob mark :2 data 16 ............... # <= 3e733d786b3529d750ee39edacea2f1c4daadca4 reset refs/heads/master commit refs/heads/master mark :3 author Joe 1512643825 +0000 committer Joe 1512643825 +0000 data 12 root commit M 100644 :1 001-001.txt M 100644 :2 001-002.txt M 100644 :2 001-003.txt # <= 409058dc98500b5685c52a091cc9f44f3975113e commit refs/heads/master mark :4 author Joe 1512643826 +0000 committer Joe 1512643826 +0000 data 10 2nd commit from :3 M 100644 :1 002-001.txt M 100644 :2 002-002.txt reset refs/heads/master from :4 """ def _get_quickjump_names(self, repos): return list(name for type, name, path, rev in repos.get_quickjump_entries('HEAD')) class GitCachedRepositoryTestCase(GitRepositoryTestCase): cached_repository = 'enabled' def test_repository_instance(self): self._git_init() self._add_repository('gitrepos') self.assertEqual(GitCachedRepository, type(self._repomgr.get_repository('gitrepos'))) def test_sync(self): self._git_init() for idx in range(3): filename = 'file%d.txt' % idx create_file(os.path.join(self.repos_path, filename)) self._git('add', filename) self._git_commit('-a', '-m', filename, date=datetime(2014, 2, 2, 17, 12, idx)) self._add_repository('gitrepos') repos = self._repomgr.get_repository('gitrepos') revs = [entry[1] for entry in repos.repos.get_node('').get_history()] revs.reverse() revs2 = [] def feedback(rev): revs2.append(rev) repos.sync(feedback=feedback) self.assertEqual(revs, revs2) self.assertEqual(4, len(revs2)) revs2 = [] def feedback_1(rev): revs2.append(rev) if len(revs2) == 2: raise StopSync def feedback_2(rev): revs2.append(rev) try: repos.sync(feedback=feedback_1, clean=True) except StopSync: self.assertEqual(revs[:2], revs2) repos.sync(feedback=feedback_2) # restart sync self.assertEqual(revs, revs2) def test_sync_file_with_invalid_byte_sequence(self): self._git_init(data=False) self._git_fast_import(b"""\ blob mark :1 data 0 reset refs/heads/master commit refs/heads/master mark :2 author Ryan Ollos 1463639119 +0200 committer Ryan Ollos 1463639119 +0200 data 9 (#12322) M 100644 :1 "\\312\\326\\267\\347\\307\\331.txt" reset refs/heads/master from :2 """) self._add_repository('gitrepos') repos = self._repomgr.get_repository('gitrepos') revs = [] def feedback(rev): revs.append(rev) repos.sync(feedback=feedback) changes = list(repos.repos.get_changeset(revs[0]).get_changes()) self.assertEqual(1, len(changes)) self.assertEqual('\ufffd\u05b7\ufffd\ufffd\ufffd.txt', changes[0][0]) def test_sync_merge(self): self._git_init() self._create_merge_commit() self._add_repository('gitrepos') repos = self._repomgr.get_repository('gitrepos') youngest_rev = repos.repos.youngest_rev oldest_rev = repos.repos.oldest_rev revs = [] def feedback(rev): revs.append(rev) repos.sync(feedback=feedback) self.assertEqual(6, len(revs)) self.assertEqual(youngest_rev, revs[-1]) self.assertEqual(oldest_rev, revs[0]) revs2 = [] def feedback_1(rev): revs2.append(rev) if len(revs2) == 3: raise StopSync def feedback_2(rev): revs2.append(rev) try: repos.sync(feedback=feedback_1, clean=True) except StopSync: self.assertEqual(revs[:3], revs2) repos.sync(feedback=feedback_2) # restart sync self.assertEqual(revs, revs2) def test_sync_too_many_merges(self): n_merges = 2000 data = self._generate_data_many_merges(n_merges) self._git_init(data=False, bare=True) self._git_fast_import(data) self._add_repository('gitrepos', bare=True) repos = self._repomgr.get_repository('gitrepos') repos.sync() rows = self.env.db_query("SELECT COUNT(*) FROM revision " "WHERE repos=%s", (repos.id,)) self.assertEqual(n_merges * 2 + 2, rows[0][0]) def _generate_data_many_merges(self, n, timestamp=1400000000): init = b"""\ blob mark :1 data 0 reset refs/heads/dev commit refs/heads/dev mark :2 author Joe %(timestamp)d +0000 committer Joe %(timestamp)d +0000 data 5 root M 100644 :1 .gitignore commit refs/heads/master mark :3 author Joe %(timestamp)d +0000 committer Joe %(timestamp)d +0000 data 7 master from :2 M 100644 :1 master.txt """ merge = b"""\ commit refs/heads/dev mark :%(dev)d author Joe %(timestamp)d +0000 committer Joe %(timestamp)d +0000 data 4 dev from :2 M 100644 :1 dev%(dev)08d.txt commit refs/heads/master mark :%(merge)d author Joe %(timestamp)d +0000 committer Joe %(timestamp)d +0000 data 19 Merge branch 'dev' from :%(from)d merge :%(dev)d M 100644 :1 dev%(dev)08d.txt """ data = io.BytesIO() data.write(init % {b'timestamp': timestamp}) for idx in range(n): data.write(merge % {b'timestamp': timestamp, b'dev': 4 + idx * 2, b'merge': 5 + idx * 2, b'from': 3 + idx * 2}) return data.getvalue() def test_sync_many_refs(self): n_refs = 1500 data = self._generate_data_many_refs(n_refs) self._git_init(data=False, bare=True) self._git_fast_import(data) self._add_repository('gitrepos', bare=True) repos = self._repomgr.get_repository('gitrepos') revs = [] def feedback(rev): revs.append(rev) repos.sync(feedback) # create cache self.assertEqual(n_refs + 1, len(revs)) revs[:] = () repos.sync(feedback) # check whether all refs are cached self.assertEqual(0, len(revs)) rows = self.env.db_query("SELECT COUNT(*) FROM revision " "WHERE repos=%s", (repos.id,)) self.assertEqual(n_refs + 1, rows[0][0]) def _generate_data_many_refs(self, n, timestamp=1400000000): root_commit = """\ blob mark :1 data 0 reset refs/heads/master commit refs/heads/master mark :2 author Joe %(timestamp)d +0000 committer Joe %(timestamp)d +0000 data 12 root commit M 100644 :1 .gitignore """ ref_commit = """ commit refs/heads/ref-%(ref)08d mark :%(mark)d author Joe %(timestamp)d +0000 committer Joe %(timestamp)d +0000 data 13 ref-%(ref)08d from :2 """ data = io.BytesIO() data.write(to_utf8(root_commit % {'timestamp': timestamp})) for idx in range(n): data.write(to_utf8(ref_commit % {'timestamp': timestamp + idx, 'mark': idx + 3, 'ref': idx})) return data.getvalue() class GitwebProjectsRepositoryProviderTestCase(unittest.TestCase): def setUp(self): self.env = EnvironmentStub() self.projects_base = mkdtemp() self.projects_list = os.path.join(self.projects_base, 'projects_list') with open(self.projects_list, 'w', encoding='utf-8') as f: f.write(""" repos1 user1 repos2.git user+2+ repos3 """) self.env.config.set('gitweb-repositories', 'projects_list', self.projects_list) self.env.config.set('gitweb-repositories', 'projects_base', self.projects_base) self.env.config.set('gitweb-repositories', 'projects_url', 'https://example.com/%s') def tearDown(self): self.env.shutdown() rmtree(self.projects_base) def test_project_list_path_not_found(self): """Warning is logged when projects_list file is not found, but exception is not raised. """ os.remove(self.projects_list) provider = GitwebProjectsRepositoryProvider(self.env) repositories = list(provider.get_repositories()) self.assertEqual([], repositories) def test_get_repositories(self): provider = GitwebProjectsRepositoryProvider(self.env) repositories = list(provider.get_repositories()) self.assertEqual(3, len(repositories)) self.assertEqual('repos1', repositories[0][0]) self.assertEqual('git', repositories[0][1]['type']) self.assertEqual(os.path.join(self.projects_base, 'repos1'), repositories[0][1]['dir']) self.assertEqual('https://example.com/repos1', repositories[0][1]['url']) self.assertEqual('repos2', repositories[1][0]) self.assertEqual('git', repositories[1][1]['type']) self.assertEqual(os.path.join(self.projects_base, 'repos2.git'), repositories[1][1]['dir']) self.assertEqual('https://example.com/repos2', repositories[1][1]['url']) self.assertEqual('repos3', repositories[2][0]) self.assertEqual('git', repositories[2][1]['type']) self.assertEqual(os.path.join(self.projects_base, 'repos3'), repositories[2][1]['dir']) self.assertEqual('https://example.com/repos3', repositories[2][1]['url']) class StopSync(Exception): pass class GitConnectorTestCase(BaseTestCase): def _git_version_from_system_info(self): git_version = None for name, version in self.env.system_info: if name == 'GIT': git_version = version return git_version def test_get_system_info(self): self.assertIsNotNone(self._git_version_from_system_info()) def test_suite(): suite = unittest.TestSuite() if GitCommandMixin.git_bin: suite.addTest(makeSuite(SanityCheckingTestCase)) suite.addTest(makeSuite(PersistentCacheTestCase)) suite.addTest(makeSuite(HistoryTimeRangeTestCase)) suite.addTest(makeSuite(GitNormalTestCase)) suite.addTest(makeSuite(GitRepositoryTestCase)) suite.addTest(makeSuite(GitCachedRepositoryTestCase)) suite.addTest(makeSuite(GitConnectorTestCase)) suite.addTest(makeSuite(GitwebProjectsRepositoryProviderTestCase)) else: print("SKIP: tracopt/versioncontrol/git/tests/git_fs.py (git cli " "binary, 'git', not found)") return suite if __name__ == '__main__': unittest.main(defaultTest='test_suite')