|
@@ -20,13 +20,14 @@
|
|
|
|
|
|
"""Tests for the repository."""
|
|
|
|
|
|
+from contextlib import closing
|
|
|
+import locale
|
|
|
import os
|
|
|
import stat
|
|
|
import shutil
|
|
|
import sys
|
|
|
import tempfile
|
|
|
import warnings
|
|
|
-import sys
|
|
|
|
|
|
from dulwich import errors
|
|
|
from dulwich.object_store import (
|
|
@@ -40,6 +41,7 @@ from dulwich.repo import (
|
|
|
)
|
|
|
from dulwich.tests import (
|
|
|
TestCase,
|
|
|
+ skipIf,
|
|
|
)
|
|
|
from dulwich.tests.utils import (
|
|
|
open_repo,
|
|
@@ -50,21 +52,9 @@ from dulwich.tests.utils import (
|
|
|
missing_sha = b'b91fa4d900e17e99b433218e988c4eb4a3e9a097'
|
|
|
|
|
|
|
|
|
-def mkdtemp_bytes():
|
|
|
- tmp_dir = tempfile.mkdtemp()
|
|
|
- if sys.version_info[0] > 2:
|
|
|
- tmp_dir = tmp_dir.encode(sys.getfilesystemencoding())
|
|
|
- return tmp_dir
|
|
|
-
|
|
|
def mkdtemp_unicode():
|
|
|
- suffix = u'déłwíçh'
|
|
|
- if sys.version_info[0] == 2:
|
|
|
- suffix = suffix.encode(sys.getfilesystemencoding())
|
|
|
- tmp_dir = tempfile.mkdtemp(suffix=suffix)
|
|
|
- if sys.version_info[0] == 2:
|
|
|
- tmp_dir = tmp_dir.decode(sys.getfilesystemencoding())
|
|
|
- return tmp_dir
|
|
|
-
|
|
|
+ suffix = u'délwíçh'
|
|
|
+ return tempfile.mkdtemp(suffix=suffix)
|
|
|
|
|
|
|
|
|
class CreateRepositoryTests(TestCase):
|
|
@@ -79,81 +69,56 @@ class CreateRepositoryTests(TestCase):
|
|
|
|
|
|
def _check_repo_contents(self, repo, expect_bare):
|
|
|
self.assertEqual(expect_bare, repo.bare)
|
|
|
- self.assertFileContentsEqual(b'Unnamed repository', repo, b'description')
|
|
|
- self.assertFileContentsEqual(b'', repo, os.path.join(b'info', b'exclude'))
|
|
|
- self.assertFileContentsEqual(None, repo, b'nonexistent file')
|
|
|
+ self.assertFileContentsEqual(b'Unnamed repository', repo, 'description')
|
|
|
+ self.assertFileContentsEqual(b'', repo, os.path.join('info', 'exclude'))
|
|
|
+ self.assertFileContentsEqual(None, repo, 'nonexistent file')
|
|
|
barestr = b'bare = ' + str(expect_bare).lower().encode('ascii')
|
|
|
- with repo.get_named_file(b'config') as f:
|
|
|
+ with repo.get_named_file('config') as f:
|
|
|
config_text = f.read()
|
|
|
self.assertTrue(barestr in config_text, "%r" % config_text)
|
|
|
|
|
|
-
|
|
|
-class CreateMemoryRepositoryTests(CreateRepositoryTests):
|
|
|
-
|
|
|
def test_create_memory(self):
|
|
|
repo = MemoryRepo.init_bare([], {})
|
|
|
self._check_repo_contents(repo, True)
|
|
|
|
|
|
-
|
|
|
-class CreateRepositoryBytesRootTests(CreateRepositoryTests):
|
|
|
-
|
|
|
- def mkdtemp(self):
|
|
|
- tmp_dir = mkdtemp_bytes()
|
|
|
- return tmp_dir, tmp_dir
|
|
|
-
|
|
|
def test_create_disk_bare(self):
|
|
|
- tmp_dir, tmp_dir_bytes = self.mkdtemp()
|
|
|
+ tmp_dir = mkdtemp_unicode()
|
|
|
self.addCleanup(shutil.rmtree, tmp_dir)
|
|
|
repo = Repo.init_bare(tmp_dir)
|
|
|
- self.assertEqual(tmp_dir_bytes, repo._controldir)
|
|
|
+ self.assertEqual(tmp_dir, repo._controldir)
|
|
|
self._check_repo_contents(repo, True)
|
|
|
|
|
|
def test_create_disk_non_bare(self):
|
|
|
- tmp_dir, tmp_dir_bytes = self.mkdtemp()
|
|
|
+ tmp_dir = mkdtemp_unicode()
|
|
|
self.addCleanup(shutil.rmtree, tmp_dir)
|
|
|
repo = Repo.init(tmp_dir)
|
|
|
- self.assertEqual(os.path.join(tmp_dir_bytes, b'.git'), repo._controldir)
|
|
|
+ self.assertEqual(os.path.join(tmp_dir, '.git'), repo._controldir)
|
|
|
self._check_repo_contents(repo, False)
|
|
|
|
|
|
|
|
|
-class CreateRepositoryUnicodeRootTests(CreateRepositoryBytesRootTests):
|
|
|
-
|
|
|
- def mktemp(self):
|
|
|
- tmp_dir = mkdtemp_unicode()
|
|
|
- tmp_dir_bytes = tmp_dir.encode(sys.getfilesystemencoding())
|
|
|
- return tmp_dir, tmp_dir_bytes
|
|
|
-
|
|
|
-
|
|
|
-class RepositoryBytesRootTests(TestCase):
|
|
|
-
|
|
|
- def setUp(self):
|
|
|
- super(RepositoryBytesRootTests, self).setUp()
|
|
|
- self._repo = None
|
|
|
-
|
|
|
- def tearDown(self):
|
|
|
- if self._repo is not None:
|
|
|
- tear_down_repo(self._repo)
|
|
|
- super(RepositoryBytesRootTests, self).tearDown()
|
|
|
+class RepositoryRootTests(TestCase):
|
|
|
|
|
|
def mkdtemp(self):
|
|
|
- return mkdtemp_bytes()
|
|
|
+ return mkdtemp_unicode()
|
|
|
|
|
|
def open_repo(self, name):
|
|
|
temp_dir = self.mkdtemp()
|
|
|
- return open_repo(name, temp_dir)
|
|
|
+ repo = open_repo(name, temp_dir)
|
|
|
+ self.addCleanup(tear_down_repo, repo)
|
|
|
+ return repo
|
|
|
|
|
|
def test_simple_props(self):
|
|
|
- r = self._repo = self.open_repo('a.git')
|
|
|
- self.assertEqual(r.controldir(), r._path_bytes)
|
|
|
+ r = self.open_repo('a.git')
|
|
|
+ self.assertEqual(r.controldir(), r.path)
|
|
|
|
|
|
def test_setitem(self):
|
|
|
- r = self._repo = self.open_repo('a.git')
|
|
|
+ r = self.open_repo('a.git')
|
|
|
r[b"refs/tags/foo"] = b'a90fa2d900a17e99b433217e988c4eb4a2e9a097'
|
|
|
self.assertEqual(b'a90fa2d900a17e99b433217e988c4eb4a2e9a097',
|
|
|
r[b"refs/tags/foo"].id)
|
|
|
|
|
|
def test_getitem_unicode(self):
|
|
|
- r = self._repo = self.open_repo('a.git')
|
|
|
+ r = self.open_repo('a.git')
|
|
|
|
|
|
test_keys = [
|
|
|
(b'refs/heads/master', True),
|
|
@@ -171,7 +136,7 @@ class RepositoryBytesRootTests(TestCase):
|
|
|
)
|
|
|
|
|
|
def test_delitem(self):
|
|
|
- r = self._repo = self.open_repo('a.git')
|
|
|
+ r = self.open_repo('a.git')
|
|
|
|
|
|
del r[b'refs/heads/master']
|
|
|
self.assertRaises(KeyError, lambda: r[b'refs/heads/master'])
|
|
@@ -182,7 +147,7 @@ class RepositoryBytesRootTests(TestCase):
|
|
|
self.assertRaises(ValueError, r.__delitem__, b'notrefs/foo')
|
|
|
|
|
|
def test_get_refs(self):
|
|
|
- r = self._repo = self.open_repo('a.git')
|
|
|
+ r = self.open_repo('a.git')
|
|
|
self.assertEqual({
|
|
|
b'HEAD': b'a90fa2d900a17e99b433217e988c4eb4a2e9a097',
|
|
|
b'refs/heads/master': b'a90fa2d900a17e99b433217e988c4eb4a2e9a097',
|
|
@@ -191,49 +156,49 @@ class RepositoryBytesRootTests(TestCase):
|
|
|
}, r.get_refs())
|
|
|
|
|
|
def test_head(self):
|
|
|
- r = self._repo = self.open_repo('a.git')
|
|
|
+ r = self.open_repo('a.git')
|
|
|
self.assertEqual(r.head(), b'a90fa2d900a17e99b433217e988c4eb4a2e9a097')
|
|
|
|
|
|
def test_get_object(self):
|
|
|
- r = self._repo = self.open_repo('a.git')
|
|
|
+ r = self.open_repo('a.git')
|
|
|
obj = r.get_object(r.head())
|
|
|
self.assertEqual(obj.type_name, b'commit')
|
|
|
|
|
|
def test_get_object_non_existant(self):
|
|
|
- r = self._repo = self.open_repo('a.git')
|
|
|
+ r = self.open_repo('a.git')
|
|
|
self.assertRaises(KeyError, r.get_object, missing_sha)
|
|
|
|
|
|
def test_contains_object(self):
|
|
|
- r = self._repo = self.open_repo('a.git')
|
|
|
+ r = self.open_repo('a.git')
|
|
|
self.assertTrue(r.head() in r)
|
|
|
|
|
|
def test_contains_ref(self):
|
|
|
- r = self._repo = self.open_repo('a.git')
|
|
|
+ r = self.open_repo('a.git')
|
|
|
self.assertTrue(b"HEAD" in r)
|
|
|
|
|
|
def test_get_no_description(self):
|
|
|
- r = self._repo = self.open_repo('a.git')
|
|
|
+ r = self.open_repo('a.git')
|
|
|
self.assertIs(None, r.get_description())
|
|
|
|
|
|
def test_get_description(self):
|
|
|
- r = self._repo = self.open_repo('a.git')
|
|
|
+ r = self.open_repo('a.git')
|
|
|
with open(os.path.join(r.path, 'description'), 'wb') as f:
|
|
|
f.write(b"Some description")
|
|
|
self.assertEqual(b"Some description", r.get_description())
|
|
|
|
|
|
def test_set_description(self):
|
|
|
- r = self._repo = self.open_repo('a.git')
|
|
|
+ r = self.open_repo('a.git')
|
|
|
description = b"Some description"
|
|
|
r.set_description(description)
|
|
|
self.assertEqual(description, r.get_description())
|
|
|
|
|
|
def test_contains_missing(self):
|
|
|
- r = self._repo = self.open_repo('a.git')
|
|
|
+ r = self.open_repo('a.git')
|
|
|
self.assertFalse(b"bar" in r)
|
|
|
|
|
|
def test_get_peeled(self):
|
|
|
# unpacked ref
|
|
|
- r = self._repo = self.open_repo('a.git')
|
|
|
+ r = self.open_repo('a.git')
|
|
|
tag_sha = b'28237f4dc30d0d462658d6b937b08a0f0b6ef55a'
|
|
|
self.assertNotEqual(r[tag_sha].sha().hexdigest(), r.head())
|
|
|
self.assertEqual(r.get_peeled(b'refs/tags/mytag'), r.head())
|
|
@@ -247,11 +212,11 @@ class RepositoryBytesRootTests(TestCase):
|
|
|
# TODO: add more corner cases to test repo
|
|
|
|
|
|
def test_get_peeled_not_tag(self):
|
|
|
- r = self._repo = self.open_repo('a.git')
|
|
|
+ r = self.open_repo('a.git')
|
|
|
self.assertEqual(r.get_peeled(b'HEAD'), r.head())
|
|
|
|
|
|
def test_get_walker(self):
|
|
|
- r = self._repo = self.open_repo('a.git')
|
|
|
+ r = self.open_repo('a.git')
|
|
|
# include defaults to [r.head()]
|
|
|
self.assertEqual([e.commit.id for e in r.get_walker()],
|
|
|
[r.head(), b'2a72d929692c41d8554c07f6301757ba18a65d91'])
|
|
@@ -263,38 +228,34 @@ class RepositoryBytesRootTests(TestCase):
|
|
|
[b'2a72d929692c41d8554c07f6301757ba18a65d91'])
|
|
|
|
|
|
def test_clone(self):
|
|
|
- r = self._repo = self.open_repo('a.git')
|
|
|
+ r = self.open_repo('a.git')
|
|
|
tmp_dir = self.mkdtemp()
|
|
|
self.addCleanup(shutil.rmtree, tmp_dir)
|
|
|
- t = r.clone(tmp_dir, mkdir=False)
|
|
|
- self.assertEqual({
|
|
|
- b'HEAD': b'a90fa2d900a17e99b433217e988c4eb4a2e9a097',
|
|
|
- b'refs/remotes/origin/master':
|
|
|
- b'a90fa2d900a17e99b433217e988c4eb4a2e9a097',
|
|
|
- b'refs/heads/master': b'a90fa2d900a17e99b433217e988c4eb4a2e9a097',
|
|
|
- b'refs/tags/mytag': b'28237f4dc30d0d462658d6b937b08a0f0b6ef55a',
|
|
|
- b'refs/tags/mytag-packed':
|
|
|
- b'b0931cadc54336e78a1d980420e3268903b57a50',
|
|
|
- }, t.refs.as_dict())
|
|
|
- shas = [e.commit.id for e in r.get_walker()]
|
|
|
- self.assertEqual(shas, [t.head(),
|
|
|
- b'2a72d929692c41d8554c07f6301757ba18a65d91'])
|
|
|
+ with closing(r.clone(tmp_dir, mkdir=False)) as t:
|
|
|
+ self.assertEqual({
|
|
|
+ b'HEAD': b'a90fa2d900a17e99b433217e988c4eb4a2e9a097',
|
|
|
+ b'refs/remotes/origin/master':
|
|
|
+ b'a90fa2d900a17e99b433217e988c4eb4a2e9a097',
|
|
|
+ b'refs/heads/master': b'a90fa2d900a17e99b433217e988c4eb4a2e9a097',
|
|
|
+ b'refs/tags/mytag': b'28237f4dc30d0d462658d6b937b08a0f0b6ef55a',
|
|
|
+ b'refs/tags/mytag-packed':
|
|
|
+ b'b0931cadc54336e78a1d980420e3268903b57a50',
|
|
|
+ }, t.refs.as_dict())
|
|
|
+ shas = [e.commit.id for e in r.get_walker()]
|
|
|
+ self.assertEqual(shas, [t.head(),
|
|
|
+ b'2a72d929692c41d8554c07f6301757ba18a65d91'])
|
|
|
|
|
|
def test_clone_no_head(self):
|
|
|
temp_dir = self.mkdtemp()
|
|
|
- if isinstance(temp_dir, bytes):
|
|
|
- temp_dir_str = temp_dir.decode(sys.getfilesystemencoding())
|
|
|
- else:
|
|
|
- temp_dir_str = temp_dir
|
|
|
self.addCleanup(shutil.rmtree, temp_dir)
|
|
|
repo_dir = os.path.join(os.path.dirname(__file__), 'data', 'repos')
|
|
|
- dest_dir = os.path.join(temp_dir_str, 'a.git')
|
|
|
+ dest_dir = os.path.join(temp_dir, 'a.git')
|
|
|
shutil.copytree(os.path.join(repo_dir, 'a.git'),
|
|
|
dest_dir, symlinks=True)
|
|
|
r = Repo(dest_dir)
|
|
|
del r.refs[b"refs/heads/master"]
|
|
|
del r.refs[b"HEAD"]
|
|
|
- t = r.clone(os.path.join(temp_dir_str, 'b.git'), mkdir=True)
|
|
|
+ t = r.clone(os.path.join(temp_dir, 'b.git'), mkdir=True)
|
|
|
self.assertEqual({
|
|
|
b'refs/tags/mytag': b'28237f4dc30d0d462658d6b937b08a0f0b6ef55a',
|
|
|
b'refs/tags/mytag-packed':
|
|
@@ -309,13 +270,13 @@ class RepositoryBytesRootTests(TestCase):
|
|
|
to the server.
|
|
|
Non-bare repo HEAD always points to an existing ref.
|
|
|
"""
|
|
|
- r = self._repo = self.open_repo('empty.git')
|
|
|
+ r = self.open_repo('empty.git')
|
|
|
tmp_dir = self.mkdtemp()
|
|
|
self.addCleanup(shutil.rmtree, tmp_dir)
|
|
|
r.clone(tmp_dir, mkdir=False, bare=True)
|
|
|
|
|
|
def test_merge_history(self):
|
|
|
- r = self._repo = self.open_repo('simple_merge.git')
|
|
|
+ r = self.open_repo('simple_merge.git')
|
|
|
shas = [e.commit.id for e in r.get_walker()]
|
|
|
self.assertEqual(shas, [b'5dac377bdded4c9aeb8dff595f0faeebcc8498cc',
|
|
|
b'ab64bbdcc51b170d21588e5c5d391ee5c0c96dfd',
|
|
@@ -325,7 +286,7 @@ class RepositoryBytesRootTests(TestCase):
|
|
|
|
|
|
def test_out_of_order_merge(self):
|
|
|
"""Test that revision history is ordered by date, not parent order."""
|
|
|
- r = self._repo = self.open_repo('ooo_merge.git')
|
|
|
+ r = self.open_repo('ooo_merge.git')
|
|
|
shas = [e.commit.id for e in r.get_walker()]
|
|
|
self.assertEqual(shas, [b'7601d7f6231db6a57f7bbb79ee52e4d462fd44d1',
|
|
|
b'f507291b64138b875c28e03469025b1ea20bc614',
|
|
@@ -333,30 +294,28 @@ class RepositoryBytesRootTests(TestCase):
|
|
|
b'f9e39b120c68182a4ba35349f832d0e4e61f485c'])
|
|
|
|
|
|
def test_get_tags_empty(self):
|
|
|
- r = self._repo = self.open_repo('ooo_merge.git')
|
|
|
+ r = self.open_repo('ooo_merge.git')
|
|
|
self.assertEqual({}, r.refs.as_dict(b'refs/tags'))
|
|
|
|
|
|
def test_get_config(self):
|
|
|
- r = self._repo = self.open_repo('ooo_merge.git')
|
|
|
+ r = self.open_repo('ooo_merge.git')
|
|
|
self.assertIsInstance(r.get_config(), Config)
|
|
|
|
|
|
def test_get_config_stack(self):
|
|
|
- r = self._repo = self.open_repo('ooo_merge.git')
|
|
|
+ r = self.open_repo('ooo_merge.git')
|
|
|
self.assertIsInstance(r.get_config_stack(), Config)
|
|
|
|
|
|
+ @skipIf(not getattr(os, 'symlink', None), 'Requires symlink support')
|
|
|
def test_submodule(self):
|
|
|
temp_dir = self.mkdtemp()
|
|
|
+ self.addCleanup(shutil.rmtree, temp_dir)
|
|
|
repo_dir = os.path.join(os.path.dirname(__file__), 'data', 'repos')
|
|
|
- if isinstance(temp_dir, bytes):
|
|
|
- temp_dir_str = temp_dir.decode(sys.getfilesystemencoding())
|
|
|
- else:
|
|
|
- temp_dir_str = temp_dir
|
|
|
shutil.copytree(os.path.join(repo_dir, 'a.git'),
|
|
|
- os.path.join(temp_dir_str, 'a.git'), symlinks=True)
|
|
|
- rel = os.path.relpath(os.path.join(repo_dir, 'submodule'), temp_dir_str)
|
|
|
- os.symlink(os.path.join(rel, 'dotgit'), os.path.join(temp_dir_str, '.git'))
|
|
|
- r = Repo(temp_dir)
|
|
|
- self.assertEqual(r.head(), b'a90fa2d900a17e99b433217e988c4eb4a2e9a097')
|
|
|
+ os.path.join(temp_dir, 'a.git'), symlinks=True)
|
|
|
+ rel = os.path.relpath(os.path.join(repo_dir, 'submodule'), temp_dir)
|
|
|
+ os.symlink(os.path.join(rel, 'dotgit'), os.path.join(temp_dir, '.git'))
|
|
|
+ with closing(Repo(temp_dir)) as r:
|
|
|
+ self.assertEqual(r.head(), b'a90fa2d900a17e99b433217e988c4eb4a2e9a097')
|
|
|
|
|
|
def test_common_revisions(self):
|
|
|
"""
|
|
@@ -377,45 +336,43 @@ class RepositoryBytesRootTests(TestCase):
|
|
|
# corrupted, but we're only checking for commits for the purpose of this
|
|
|
# test, so it's immaterial.
|
|
|
r1_dir = self.mkdtemp()
|
|
|
+ self.addCleanup(shutil.rmtree, r1_dir)
|
|
|
r1_commits = [b'ab64bbdcc51b170d21588e5c5d391ee5c0c96dfd', # HEAD
|
|
|
b'60dacdc733de308bb77bb76ce0fb0f9b44c9769e',
|
|
|
b'0d89f20333fbb1d2f3a94da77f4981373d8f4310']
|
|
|
|
|
|
r2_dir = self.mkdtemp()
|
|
|
+ self.addCleanup(shutil.rmtree, r2_dir)
|
|
|
r2_commits = [b'4cffe90e0a41ad3f5190079d7c8f036bde29cbe6', # HEAD
|
|
|
b'60dacdc733de308bb77bb76ce0fb0f9b44c9769e',
|
|
|
b'0d89f20333fbb1d2f3a94da77f4981373d8f4310']
|
|
|
|
|
|
- try:
|
|
|
- r1 = Repo.init_bare(r1_dir)
|
|
|
- for c in r1_commits:
|
|
|
- r1.object_store.add_object(r_base.get_object(c))
|
|
|
- r1.refs[b'HEAD'] = r1_commits[0]
|
|
|
+ r1 = Repo.init_bare(r1_dir)
|
|
|
+ for c in r1_commits:
|
|
|
+ r1.object_store.add_object(r_base.get_object(c))
|
|
|
+ r1.refs[b'HEAD'] = r1_commits[0]
|
|
|
|
|
|
- r2 = Repo.init_bare(r2_dir)
|
|
|
- for c in r2_commits:
|
|
|
- r2.object_store.add_object(r_base.get_object(c))
|
|
|
- r2.refs[b'HEAD'] = r2_commits[0]
|
|
|
+ r2 = Repo.init_bare(r2_dir)
|
|
|
+ for c in r2_commits:
|
|
|
+ r2.object_store.add_object(r_base.get_object(c))
|
|
|
+ r2.refs[b'HEAD'] = r2_commits[0]
|
|
|
|
|
|
- # Finally, the 'real' testing!
|
|
|
- shas = r2.object_store.find_common_revisions(r1.get_graph_walker())
|
|
|
- self.assertEqual(set(shas), expected_shas)
|
|
|
+ # Finally, the 'real' testing!
|
|
|
+ shas = r2.object_store.find_common_revisions(r1.get_graph_walker())
|
|
|
+ self.assertEqual(set(shas), expected_shas)
|
|
|
|
|
|
- shas = r1.object_store.find_common_revisions(r2.get_graph_walker())
|
|
|
- self.assertEqual(set(shas), expected_shas)
|
|
|
- finally:
|
|
|
- shutil.rmtree(r1_dir)
|
|
|
- shutil.rmtree(r2_dir)
|
|
|
+ shas = r1.object_store.find_common_revisions(r2.get_graph_walker())
|
|
|
+ self.assertEqual(set(shas), expected_shas)
|
|
|
|
|
|
def test_shell_hook_pre_commit(self):
|
|
|
if os.name != 'posix':
|
|
|
self.skipTest('shell hook tests requires POSIX shell')
|
|
|
|
|
|
- pre_commit_fail = b"""#!/bin/sh
|
|
|
+ pre_commit_fail = """#!/bin/sh
|
|
|
exit 1
|
|
|
"""
|
|
|
|
|
|
- pre_commit_success = b"""#!/bin/sh
|
|
|
+ pre_commit_success = """#!/bin/sh
|
|
|
exit 0
|
|
|
"""
|
|
|
|
|
@@ -423,9 +380,9 @@ exit 0
|
|
|
r = Repo.init(repo_dir)
|
|
|
self.addCleanup(shutil.rmtree, repo_dir)
|
|
|
|
|
|
- pre_commit = os.path.join(r.controldir(), b'hooks', b'pre-commit')
|
|
|
+ pre_commit = os.path.join(r.controldir(), 'hooks', 'pre-commit')
|
|
|
|
|
|
- with open(pre_commit, 'wb') as f:
|
|
|
+ with open(pre_commit, 'w') as f:
|
|
|
f.write(pre_commit_fail)
|
|
|
os.chmod(pre_commit, stat.S_IREAD | stat.S_IWRITE | stat.S_IEXEC)
|
|
|
|
|
@@ -435,7 +392,7 @@ exit 0
|
|
|
commit_timestamp=12345, commit_timezone=0,
|
|
|
author_timestamp=12345, author_timezone=0)
|
|
|
|
|
|
- with open(pre_commit, 'wb') as f:
|
|
|
+ with open(pre_commit, 'w') as f:
|
|
|
f.write(pre_commit_success)
|
|
|
os.chmod(pre_commit, stat.S_IREAD | stat.S_IWRITE | stat.S_IEXEC)
|
|
|
|
|
@@ -451,21 +408,21 @@ exit 0
|
|
|
if os.name != 'posix':
|
|
|
self.skipTest('shell hook tests requires POSIX shell')
|
|
|
|
|
|
- commit_msg_fail = b"""#!/bin/sh
|
|
|
+ commit_msg_fail = """#!/bin/sh
|
|
|
exit 1
|
|
|
"""
|
|
|
|
|
|
- commit_msg_success = b"""#!/bin/sh
|
|
|
+ commit_msg_success = """#!/bin/sh
|
|
|
exit 0
|
|
|
"""
|
|
|
|
|
|
- repo_dir = os.path.join(self.mkdtemp())
|
|
|
+ repo_dir = self.mkdtemp()
|
|
|
r = Repo.init(repo_dir)
|
|
|
self.addCleanup(shutil.rmtree, repo_dir)
|
|
|
|
|
|
- commit_msg = os.path.join(r.controldir(), b'hooks', b'commit-msg')
|
|
|
+ commit_msg = os.path.join(r.controldir(), 'hooks', 'commit-msg')
|
|
|
|
|
|
- with open(commit_msg, 'wb') as f:
|
|
|
+ with open(commit_msg, 'w') as f:
|
|
|
f.write(commit_msg_fail)
|
|
|
os.chmod(commit_msg, stat.S_IREAD | stat.S_IWRITE | stat.S_IEXEC)
|
|
|
|
|
@@ -475,7 +432,7 @@ exit 0
|
|
|
commit_timestamp=12345, commit_timezone=0,
|
|
|
author_timestamp=12345, author_timezone=0)
|
|
|
|
|
|
- with open(commit_msg, 'wb') as f:
|
|
|
+ with open(commit_msg, 'w') as f:
|
|
|
f.write(commit_msg_success)
|
|
|
os.chmod(commit_msg, stat.S_IREAD | stat.S_IWRITE | stat.S_IEXEC)
|
|
|
|
|
@@ -492,15 +449,12 @@ exit 0
|
|
|
self.skipTest('shell hook tests requires POSIX shell')
|
|
|
|
|
|
repo_dir = self.mkdtemp()
|
|
|
- if isinstance(repo_dir, bytes):
|
|
|
- repo_dir_str = repo_dir.decode(sys.getfilesystemencoding())
|
|
|
- else:
|
|
|
- repo_dir_str = repo_dir
|
|
|
|
|
|
r = Repo.init(repo_dir)
|
|
|
self.addCleanup(shutil.rmtree, repo_dir)
|
|
|
|
|
|
- (fd, path) = tempfile.mkstemp(dir=repo_dir_str)
|
|
|
+ (fd, path) = tempfile.mkstemp(dir=repo_dir)
|
|
|
+ os.close(fd)
|
|
|
post_commit_msg = """#!/bin/sh
|
|
|
rm """ + path + """
|
|
|
"""
|
|
@@ -513,10 +467,10 @@ rm """ + path + """
|
|
|
author_timestamp=12345, author_timezone=0)
|
|
|
self.assertEqual([], r[root_sha].parents)
|
|
|
|
|
|
- post_commit = os.path.join(r.controldir(), b'hooks', b'post-commit')
|
|
|
+ post_commit = os.path.join(r.controldir(), 'hooks', 'post-commit')
|
|
|
|
|
|
- with open(post_commit, 'w') as f:
|
|
|
- f.write(post_commit_msg)
|
|
|
+ with open(post_commit, 'wb') as f:
|
|
|
+ f.write(post_commit_msg.encode(locale.getpreferredencoding()))
|
|
|
os.chmod(post_commit, stat.S_IREAD | stat.S_IWRITE | stat.S_IEXEC)
|
|
|
|
|
|
commit_sha = r.do_commit(
|
|
@@ -553,13 +507,7 @@ exit 1
|
|
|
self.assertEqual([commit_sha], r[commit_sha2].parents)
|
|
|
|
|
|
|
|
|
-class RepositoryUnicodeRootTests(RepositoryBytesRootTests):
|
|
|
-
|
|
|
- def mktemp(self):
|
|
|
- return mkdtemp_unicode()
|
|
|
-
|
|
|
-
|
|
|
-class BuildRepoBytesRootTests(TestCase):
|
|
|
+class BuildRepoRootTests(TestCase):
|
|
|
"""Tests that build on-disk repos from scratch.
|
|
|
|
|
|
Repos live in a temp dir and are torn down after each test. They start with
|
|
@@ -567,21 +515,19 @@ class BuildRepoBytesRootTests(TestCase):
|
|
|
"""
|
|
|
|
|
|
def get_repo_dir(self):
|
|
|
- return os.path.join(mkdtemp_bytes(), b'test')
|
|
|
-
|
|
|
- def get_a_filename(self):
|
|
|
- return b'a'
|
|
|
+ return os.path.join(mkdtemp_unicode(), 'test')
|
|
|
|
|
|
def setUp(self):
|
|
|
- super(BuildRepoBytesRootTests, self).setUp()
|
|
|
+ super(BuildRepoRootTests, self).setUp()
|
|
|
self._repo_dir = self.get_repo_dir()
|
|
|
os.makedirs(self._repo_dir)
|
|
|
r = self._repo = Repo.init(self._repo_dir)
|
|
|
+ self.addCleanup(tear_down_repo, r)
|
|
|
self.assertFalse(r.bare)
|
|
|
self.assertEqual(b'ref: refs/heads/master', r.refs.read_ref(b'HEAD'))
|
|
|
self.assertRaises(KeyError, lambda: r.refs[b'refs/heads/master'])
|
|
|
|
|
|
- with open(os.path.join(r._path_bytes, b'a'), 'wb') as f:
|
|
|
+ with open(os.path.join(r.path, 'a'), 'wb') as f:
|
|
|
f.write(b'file contents')
|
|
|
r.stage(['a'])
|
|
|
commit_sha = r.do_commit(b'msg',
|
|
@@ -592,10 +538,6 @@ class BuildRepoBytesRootTests(TestCase):
|
|
|
self.assertEqual([], r[commit_sha].parents)
|
|
|
self._root_commit = commit_sha
|
|
|
|
|
|
- def tearDown(self):
|
|
|
- tear_down_repo(self._repo)
|
|
|
- super(BuildRepoBytesRootTests, self).tearDown()
|
|
|
-
|
|
|
def test_build_repo(self):
|
|
|
r = self._repo
|
|
|
self.assertEqual(b'ref: refs/heads/master', r.refs.read_ref(b'HEAD'))
|
|
@@ -607,10 +549,9 @@ class BuildRepoBytesRootTests(TestCase):
|
|
|
|
|
|
def test_commit_modified(self):
|
|
|
r = self._repo
|
|
|
- with open(os.path.join(r._path_bytes, b'a'), 'wb') as f:
|
|
|
+ with open(os.path.join(r.path, 'a'), 'wb') as f:
|
|
|
f.write(b'new contents')
|
|
|
- os.symlink('a', os.path.join(r._path_bytes, b'b'))
|
|
|
- r.stage(['a', 'b'])
|
|
|
+ r.stage(['a'])
|
|
|
commit_sha = r.do_commit(b'modified a',
|
|
|
committer=b'Test Committer <test@nodomain.com>',
|
|
|
author=b'Test Author <test@nodomain.com>',
|
|
@@ -620,13 +561,25 @@ class BuildRepoBytesRootTests(TestCase):
|
|
|
a_mode, a_id = tree_lookup_path(r.get_object, r[commit_sha].tree, b'a')
|
|
|
self.assertEqual(stat.S_IFREG | 0o644, a_mode)
|
|
|
self.assertEqual(b'new contents', r[a_id].data)
|
|
|
+
|
|
|
+ @skipIf(not getattr(os, 'symlink', None), 'Requires symlink support')
|
|
|
+ def test_commit_symlink(self):
|
|
|
+ r = self._repo
|
|
|
+ os.symlink('a', os.path.join(r.path, 'b'))
|
|
|
+ r.stage(['a', 'b'])
|
|
|
+ commit_sha = r.do_commit(b'Symlink b',
|
|
|
+ committer=b'Test Committer <test@nodomain.com>',
|
|
|
+ author=b'Test Author <test@nodomain.com>',
|
|
|
+ commit_timestamp=12395, commit_timezone=0,
|
|
|
+ author_timestamp=12395, author_timezone=0)
|
|
|
+ self.assertEqual([self._root_commit], r[commit_sha].parents)
|
|
|
b_mode, b_id = tree_lookup_path(r.get_object, r[commit_sha].tree, b'b')
|
|
|
self.assertTrue(stat.S_ISLNK(b_mode))
|
|
|
self.assertEqual(b'a', r[b_id].data)
|
|
|
|
|
|
def test_commit_deleted(self):
|
|
|
r = self._repo
|
|
|
- os.remove(os.path.join(r._path_bytes, b'a'))
|
|
|
+ os.remove(os.path.join(r.path, 'a'))
|
|
|
r.stage(['a'])
|
|
|
commit_sha = r.do_commit(b'deleted a',
|
|
|
committer=b'Test Committer <test@nodomain.com>',
|
|
@@ -788,17 +741,32 @@ class BuildRepoBytesRootTests(TestCase):
|
|
|
|
|
|
def test_stage_deleted(self):
|
|
|
r = self._repo
|
|
|
- os.remove(os.path.join(r._path_bytes, b'a'))
|
|
|
+ os.remove(os.path.join(r.path, 'a'))
|
|
|
r.stage(['a'])
|
|
|
r.stage(['a']) # double-stage a deleted path
|
|
|
|
|
|
+ def test_commit_no_encode_decode(self):
|
|
|
+ r = self._repo
|
|
|
+ repo_path_bytes = r.path.encode(sys.getfilesystemencoding())
|
|
|
+ encodings = ('utf8', 'latin1')
|
|
|
+ names = [u'À'.encode(encoding) for encoding in encodings]
|
|
|
+ for name, encoding in zip(names, encodings):
|
|
|
+ full_path = os.path.join(repo_path_bytes, name)
|
|
|
+ with open(full_path, 'wb') as f:
|
|
|
+ f.write(encoding.encode('ascii'))
|
|
|
+ # These files are break tear_down_repo, so cleanup these files
|
|
|
+ # ourselves.
|
|
|
+ self.addCleanup(os.remove, full_path)
|
|
|
+
|
|
|
+ r.stage(names)
|
|
|
+ commit_sha = r.do_commit(b'Files with different encodings',
|
|
|
+ committer=b'Test Committer <test@nodomain.com>',
|
|
|
+ author=b'Test Author <test@nodomain.com>',
|
|
|
+ commit_timestamp=12395, commit_timezone=0,
|
|
|
+ author_timestamp=12395, author_timezone=0,
|
|
|
+ ref=None, merge_heads=[self._root_commit])
|
|
|
|
|
|
-class BuildRepoUnicodeRootTests(TestCase):
|
|
|
- """Tests that build on-disk repos from scratch.
|
|
|
-
|
|
|
- Repos live in a temp dir and are torn down after each test. They start with
|
|
|
- a single commit in master having single file named 'a'.
|
|
|
- """
|
|
|
-
|
|
|
- def get_repo_dir(self):
|
|
|
- return os.path.join(mkdtemp_unicode(), 'test')
|
|
|
+ for name, encoding in zip(names, encodings):
|
|
|
+ mode, id = tree_lookup_path(r.get_object, r[commit_sha].tree, name)
|
|
|
+ self.assertEqual(stat.S_IFREG | 0o644, mode)
|
|
|
+ self.assertEqual(encoding.encode('ascii'), r[id].data)
|