mirror of
https://gerrit.googlesource.com/git-repo
synced 2024-12-21 07:16:21 +00:00
74737da1ab
Now that we don't need to support Python 2, we can switch to this API for better contextmanager logic. Change-Id: I2d03e391121886547e7808a3b5c3b470c411533f Reviewed-on: https://gerrit-review.googlesource.com/c/git-repo/+/337515 Reviewed-by: LaMont Jones <lamontjones@google.com> Tested-by: Mike Frysinger <vapier@google.com>
406 lines
13 KiB
Python
406 lines
13 KiB
Python
# Copyright (C) 2019 The Android Open Source Project
|
|
#
|
|
# Licensed under the Apache License, Version 2.0 (the "License");
|
|
# you may not use this file except in compliance with the License.
|
|
# You may obtain a copy of the License at
|
|
#
|
|
# http://www.apache.org/licenses/LICENSE-2.0
|
|
#
|
|
# Unless required by applicable law or agreed to in writing, software
|
|
# distributed under the License is distributed on an "AS IS" BASIS,
|
|
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
|
# See the License for the specific language governing permissions and
|
|
# limitations under the License.
|
|
|
|
"""Unittests for the project.py module."""
|
|
|
|
import contextlib
|
|
import os
|
|
from pathlib import Path
|
|
import subprocess
|
|
import tempfile
|
|
import unittest
|
|
|
|
import error
|
|
import git_command
|
|
import git_config
|
|
import platform_utils
|
|
import project
|
|
|
|
|
|
@contextlib.contextmanager
|
|
def TempGitTree():
|
|
"""Create a new empty git checkout for testing."""
|
|
with tempfile.TemporaryDirectory(prefix='repo-tests') as tempdir:
|
|
# Tests need to assume, that main is default branch at init,
|
|
# which is not supported in config until 2.28.
|
|
cmd = ['git', 'init']
|
|
if git_command.git_require((2, 28, 0)):
|
|
cmd += ['--initial-branch=main']
|
|
else:
|
|
# Use template dir for init.
|
|
templatedir = tempfile.mkdtemp(prefix='.test-template')
|
|
with open(os.path.join(templatedir, 'HEAD'), 'w') as fp:
|
|
fp.write('ref: refs/heads/main\n')
|
|
cmd += ['--template', templatedir]
|
|
subprocess.check_call(cmd, cwd=tempdir)
|
|
yield tempdir
|
|
|
|
|
|
class FakeProject(object):
|
|
"""A fake for Project for basic functionality."""
|
|
|
|
def __init__(self, worktree):
|
|
self.worktree = worktree
|
|
self.gitdir = os.path.join(worktree, '.git')
|
|
self.name = 'fakeproject'
|
|
self.work_git = project.Project._GitGetByExec(
|
|
self, bare=False, gitdir=self.gitdir)
|
|
self.bare_git = project.Project._GitGetByExec(
|
|
self, bare=True, gitdir=self.gitdir)
|
|
self.config = git_config.GitConfig.ForRepository(gitdir=self.gitdir)
|
|
|
|
|
|
class ReviewableBranchTests(unittest.TestCase):
|
|
"""Check ReviewableBranch behavior."""
|
|
|
|
def test_smoke(self):
|
|
"""A quick run through everything."""
|
|
with TempGitTree() as tempdir:
|
|
fakeproj = FakeProject(tempdir)
|
|
|
|
# Generate some commits.
|
|
with open(os.path.join(tempdir, 'readme'), 'w') as fp:
|
|
fp.write('txt')
|
|
fakeproj.work_git.add('readme')
|
|
fakeproj.work_git.commit('-mAdd file')
|
|
fakeproj.work_git.checkout('-b', 'work')
|
|
fakeproj.work_git.rm('-f', 'readme')
|
|
fakeproj.work_git.commit('-mDel file')
|
|
|
|
# Start off with the normal details.
|
|
rb = project.ReviewableBranch(
|
|
fakeproj, fakeproj.config.GetBranch('work'), 'main')
|
|
self.assertEqual('work', rb.name)
|
|
self.assertEqual(1, len(rb.commits))
|
|
self.assertIn('Del file', rb.commits[0])
|
|
d = rb.unabbrev_commits
|
|
self.assertEqual(1, len(d))
|
|
short, long = next(iter(d.items()))
|
|
self.assertTrue(long.startswith(short))
|
|
self.assertTrue(rb.base_exists)
|
|
# Hard to assert anything useful about this.
|
|
self.assertTrue(rb.date)
|
|
|
|
# Now delete the tracking branch!
|
|
fakeproj.work_git.branch('-D', 'main')
|
|
rb = project.ReviewableBranch(
|
|
fakeproj, fakeproj.config.GetBranch('work'), 'main')
|
|
self.assertEqual(0, len(rb.commits))
|
|
self.assertFalse(rb.base_exists)
|
|
# Hard to assert anything useful about this.
|
|
self.assertTrue(rb.date)
|
|
|
|
|
|
class CopyLinkTestCase(unittest.TestCase):
|
|
"""TestCase for stub repo client checkouts.
|
|
|
|
It'll have a layout like:
|
|
tempdir/ # self.tempdir
|
|
checkout/ # self.topdir
|
|
git-project/ # self.worktree
|
|
|
|
Attributes:
|
|
tempdir: A dedicated temporary directory.
|
|
worktree: The top of the repo client checkout.
|
|
topdir: The top of a project checkout.
|
|
"""
|
|
|
|
def setUp(self):
|
|
self.tempdirobj = tempfile.TemporaryDirectory(prefix='repo_tests')
|
|
self.tempdir = self.tempdirobj.name
|
|
self.topdir = os.path.join(self.tempdir, 'checkout')
|
|
self.worktree = os.path.join(self.topdir, 'git-project')
|
|
os.makedirs(self.topdir)
|
|
os.makedirs(self.worktree)
|
|
|
|
def tearDown(self):
|
|
self.tempdirobj.cleanup()
|
|
|
|
@staticmethod
|
|
def touch(path):
|
|
with open(path, 'w'):
|
|
pass
|
|
|
|
def assertExists(self, path, msg=None):
|
|
"""Make sure |path| exists."""
|
|
if os.path.exists(path):
|
|
return
|
|
|
|
if msg is None:
|
|
msg = ['path is missing: %s' % path]
|
|
while path != '/':
|
|
path = os.path.dirname(path)
|
|
if not path:
|
|
# If we're given something like "foo", abort once we get to "".
|
|
break
|
|
result = os.path.exists(path)
|
|
msg.append('\tos.path.exists(%s): %s' % (path, result))
|
|
if result:
|
|
msg.append('\tcontents: %r' % os.listdir(path))
|
|
break
|
|
msg = '\n'.join(msg)
|
|
|
|
raise self.failureException(msg)
|
|
|
|
|
|
class CopyFile(CopyLinkTestCase):
|
|
"""Check _CopyFile handling."""
|
|
|
|
def CopyFile(self, src, dest):
|
|
return project._CopyFile(self.worktree, src, self.topdir, dest)
|
|
|
|
def test_basic(self):
|
|
"""Basic test of copying a file from a project to the toplevel."""
|
|
src = os.path.join(self.worktree, 'foo.txt')
|
|
self.touch(src)
|
|
cf = self.CopyFile('foo.txt', 'foo')
|
|
cf._Copy()
|
|
self.assertExists(os.path.join(self.topdir, 'foo'))
|
|
|
|
def test_src_subdir(self):
|
|
"""Copy a file from a subdir of a project."""
|
|
src = os.path.join(self.worktree, 'bar', 'foo.txt')
|
|
os.makedirs(os.path.dirname(src))
|
|
self.touch(src)
|
|
cf = self.CopyFile('bar/foo.txt', 'new.txt')
|
|
cf._Copy()
|
|
self.assertExists(os.path.join(self.topdir, 'new.txt'))
|
|
|
|
def test_dest_subdir(self):
|
|
"""Copy a file to a subdir of a checkout."""
|
|
src = os.path.join(self.worktree, 'foo.txt')
|
|
self.touch(src)
|
|
cf = self.CopyFile('foo.txt', 'sub/dir/new.txt')
|
|
self.assertFalse(os.path.exists(os.path.join(self.topdir, 'sub')))
|
|
cf._Copy()
|
|
self.assertExists(os.path.join(self.topdir, 'sub', 'dir', 'new.txt'))
|
|
|
|
def test_update(self):
|
|
"""Make sure changed files get copied again."""
|
|
src = os.path.join(self.worktree, 'foo.txt')
|
|
dest = os.path.join(self.topdir, 'bar')
|
|
with open(src, 'w') as f:
|
|
f.write('1st')
|
|
cf = self.CopyFile('foo.txt', 'bar')
|
|
cf._Copy()
|
|
self.assertExists(dest)
|
|
with open(dest) as f:
|
|
self.assertEqual(f.read(), '1st')
|
|
|
|
with open(src, 'w') as f:
|
|
f.write('2nd!')
|
|
cf._Copy()
|
|
with open(dest) as f:
|
|
self.assertEqual(f.read(), '2nd!')
|
|
|
|
def test_src_block_symlink(self):
|
|
"""Do not allow reading from a symlinked path."""
|
|
src = os.path.join(self.worktree, 'foo.txt')
|
|
sym = os.path.join(self.worktree, 'sym')
|
|
self.touch(src)
|
|
platform_utils.symlink('foo.txt', sym)
|
|
self.assertExists(sym)
|
|
cf = self.CopyFile('sym', 'foo')
|
|
self.assertRaises(error.ManifestInvalidPathError, cf._Copy)
|
|
|
|
def test_src_block_symlink_traversal(self):
|
|
"""Do not allow reading through a symlink dir."""
|
|
realfile = os.path.join(self.tempdir, 'file.txt')
|
|
self.touch(realfile)
|
|
src = os.path.join(self.worktree, 'bar', 'file.txt')
|
|
platform_utils.symlink(self.tempdir, os.path.join(self.worktree, 'bar'))
|
|
self.assertExists(src)
|
|
cf = self.CopyFile('bar/file.txt', 'foo')
|
|
self.assertRaises(error.ManifestInvalidPathError, cf._Copy)
|
|
|
|
def test_src_block_copy_from_dir(self):
|
|
"""Do not allow copying from a directory."""
|
|
src = os.path.join(self.worktree, 'dir')
|
|
os.makedirs(src)
|
|
cf = self.CopyFile('dir', 'foo')
|
|
self.assertRaises(error.ManifestInvalidPathError, cf._Copy)
|
|
|
|
def test_dest_block_symlink(self):
|
|
"""Do not allow writing to a symlink."""
|
|
src = os.path.join(self.worktree, 'foo.txt')
|
|
self.touch(src)
|
|
platform_utils.symlink('dest', os.path.join(self.topdir, 'sym'))
|
|
cf = self.CopyFile('foo.txt', 'sym')
|
|
self.assertRaises(error.ManifestInvalidPathError, cf._Copy)
|
|
|
|
def test_dest_block_symlink_traversal(self):
|
|
"""Do not allow writing through a symlink dir."""
|
|
src = os.path.join(self.worktree, 'foo.txt')
|
|
self.touch(src)
|
|
platform_utils.symlink(tempfile.gettempdir(),
|
|
os.path.join(self.topdir, 'sym'))
|
|
cf = self.CopyFile('foo.txt', 'sym/foo.txt')
|
|
self.assertRaises(error.ManifestInvalidPathError, cf._Copy)
|
|
|
|
def test_src_block_copy_to_dir(self):
|
|
"""Do not allow copying to a directory."""
|
|
src = os.path.join(self.worktree, 'foo.txt')
|
|
self.touch(src)
|
|
os.makedirs(os.path.join(self.topdir, 'dir'))
|
|
cf = self.CopyFile('foo.txt', 'dir')
|
|
self.assertRaises(error.ManifestInvalidPathError, cf._Copy)
|
|
|
|
|
|
class LinkFile(CopyLinkTestCase):
|
|
"""Check _LinkFile handling."""
|
|
|
|
def LinkFile(self, src, dest):
|
|
return project._LinkFile(self.worktree, src, self.topdir, dest)
|
|
|
|
def test_basic(self):
|
|
"""Basic test of linking a file from a project into the toplevel."""
|
|
src = os.path.join(self.worktree, 'foo.txt')
|
|
self.touch(src)
|
|
lf = self.LinkFile('foo.txt', 'foo')
|
|
lf._Link()
|
|
dest = os.path.join(self.topdir, 'foo')
|
|
self.assertExists(dest)
|
|
self.assertTrue(os.path.islink(dest))
|
|
self.assertEqual(os.path.join('git-project', 'foo.txt'), os.readlink(dest))
|
|
|
|
def test_src_subdir(self):
|
|
"""Link to a file in a subdir of a project."""
|
|
src = os.path.join(self.worktree, 'bar', 'foo.txt')
|
|
os.makedirs(os.path.dirname(src))
|
|
self.touch(src)
|
|
lf = self.LinkFile('bar/foo.txt', 'foo')
|
|
lf._Link()
|
|
self.assertExists(os.path.join(self.topdir, 'foo'))
|
|
|
|
def test_src_self(self):
|
|
"""Link to the project itself."""
|
|
dest = os.path.join(self.topdir, 'foo', 'bar')
|
|
lf = self.LinkFile('.', 'foo/bar')
|
|
lf._Link()
|
|
self.assertExists(dest)
|
|
self.assertEqual(os.path.join('..', 'git-project'), os.readlink(dest))
|
|
|
|
def test_dest_subdir(self):
|
|
"""Link a file to a subdir of a checkout."""
|
|
src = os.path.join(self.worktree, 'foo.txt')
|
|
self.touch(src)
|
|
lf = self.LinkFile('foo.txt', 'sub/dir/foo/bar')
|
|
self.assertFalse(os.path.exists(os.path.join(self.topdir, 'sub')))
|
|
lf._Link()
|
|
self.assertExists(os.path.join(self.topdir, 'sub', 'dir', 'foo', 'bar'))
|
|
|
|
def test_src_block_relative(self):
|
|
"""Do not allow relative symlinks."""
|
|
BAD_SOURCES = (
|
|
'./',
|
|
'..',
|
|
'../',
|
|
'foo/.',
|
|
'foo/./bar',
|
|
'foo/..',
|
|
'foo/../foo',
|
|
)
|
|
for src in BAD_SOURCES:
|
|
lf = self.LinkFile(src, 'foo')
|
|
self.assertRaises(error.ManifestInvalidPathError, lf._Link)
|
|
|
|
def test_update(self):
|
|
"""Make sure changed targets get updated."""
|
|
dest = os.path.join(self.topdir, 'sym')
|
|
|
|
src = os.path.join(self.worktree, 'foo.txt')
|
|
self.touch(src)
|
|
lf = self.LinkFile('foo.txt', 'sym')
|
|
lf._Link()
|
|
self.assertEqual(os.path.join('git-project', 'foo.txt'), os.readlink(dest))
|
|
|
|
# Point the symlink somewhere else.
|
|
os.unlink(dest)
|
|
platform_utils.symlink(self.tempdir, dest)
|
|
lf._Link()
|
|
self.assertEqual(os.path.join('git-project', 'foo.txt'), os.readlink(dest))
|
|
|
|
|
|
class MigrateWorkTreeTests(unittest.TestCase):
|
|
"""Check _MigrateOldWorkTreeGitDir handling."""
|
|
|
|
_SYMLINKS = {
|
|
'config', 'description', 'hooks', 'info', 'logs', 'objects',
|
|
'packed-refs', 'refs', 'rr-cache', 'shallow', 'svn',
|
|
}
|
|
_FILES = {
|
|
'COMMIT_EDITMSG', 'FETCH_HEAD', 'HEAD', 'index', 'ORIG_HEAD',
|
|
'unknown-file-should-be-migrated',
|
|
}
|
|
_CLEAN_FILES = {
|
|
'a-vim-temp-file~', '#an-emacs-temp-file#',
|
|
}
|
|
|
|
@classmethod
|
|
@contextlib.contextmanager
|
|
def _simple_layout(cls):
|
|
"""Create a simple repo client checkout to test against."""
|
|
with tempfile.TemporaryDirectory() as tempdir:
|
|
tempdir = Path(tempdir)
|
|
|
|
gitdir = tempdir / '.repo/projects/src/test.git'
|
|
gitdir.mkdir(parents=True)
|
|
cmd = ['git', 'init', '--bare', str(gitdir)]
|
|
subprocess.check_call(cmd)
|
|
|
|
dotgit = tempdir / 'src/test/.git'
|
|
dotgit.mkdir(parents=True)
|
|
for name in cls._SYMLINKS:
|
|
(dotgit / name).symlink_to(f'../../../.repo/projects/src/test.git/{name}')
|
|
for name in cls._FILES | cls._CLEAN_FILES:
|
|
(dotgit / name).write_text(name)
|
|
|
|
yield tempdir
|
|
|
|
def test_standard(self):
|
|
"""Migrate a standard checkout that we expect."""
|
|
with self._simple_layout() as tempdir:
|
|
dotgit = tempdir / 'src/test/.git'
|
|
project.Project._MigrateOldWorkTreeGitDir(str(dotgit))
|
|
|
|
# Make sure the dir was transformed into a symlink.
|
|
self.assertTrue(dotgit.is_symlink())
|
|
self.assertEqual(os.readlink(dotgit), '../../.repo/projects/src/test.git')
|
|
|
|
# Make sure files were moved over.
|
|
gitdir = tempdir / '.repo/projects/src/test.git'
|
|
for name in self._FILES:
|
|
self.assertEqual(name, (gitdir / name).read_text())
|
|
# Make sure files were removed.
|
|
for name in self._CLEAN_FILES:
|
|
self.assertFalse((gitdir / name).exists())
|
|
|
|
def test_unknown(self):
|
|
"""A checkout with unknown files should abort."""
|
|
with self._simple_layout() as tempdir:
|
|
dotgit = tempdir / 'src/test/.git'
|
|
(tempdir / '.repo/projects/src/test.git/random-file').write_text('one')
|
|
(dotgit / 'random-file').write_text('two')
|
|
with self.assertRaises(error.GitError):
|
|
project.Project._MigrateOldWorkTreeGitDir(str(dotgit))
|
|
|
|
# Make sure no content was actually changed.
|
|
self.assertTrue(dotgit.is_dir())
|
|
for name in self._FILES:
|
|
self.assertTrue((dotgit / name).is_file())
|
|
for name in self._CLEAN_FILES:
|
|
self.assertTrue((dotgit / name).is_file())
|
|
for name in self._SYMLINKS:
|
|
self.assertTrue((dotgit / name).is_symlink())
|