Run repository tests with the path set as unicode strings as well as byte strings.

dulwich/tests/test_repository.py

@@ -1,3 +1,4 @@
+# -*- coding: utf-8 -*-
 # test_repository.py -- tests for repository.py
 # Copyright (C) 2007 James Westby <jw+debian@jameswestby.net>
 #
@@ -25,6 +26,7 @@ import shutil
 import sys
 import tempfile
 import warnings
+import sys
 
 from dulwich import errors
 from dulwich.object_store import (
@@ -48,6 +50,23 @@ 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
+
+
+
 class CreateRepositoryTests(TestCase):
 
     def assertFileContentsEqual(self, expected, repo, path):
@@ -68,48 +87,73 @@ class CreateRepositoryTests(TestCase):
             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 = tempfile.mkdtemp()
+        tmp_dir, tmp_dir_bytes = self.mkdtemp()
         self.addCleanup(shutil.rmtree, tmp_dir)
         repo = Repo.init_bare(tmp_dir)
-        self.assertEqual(tmp_dir, repo._controldir)
+        self.assertEqual(tmp_dir_bytes, repo._controldir)
         self._check_repo_contents(repo, True)
 
     def test_create_disk_non_bare(self):
-        tmp_dir = tempfile.mkdtemp()
+        tmp_dir, tmp_dir_bytes = self.mkdtemp()
         self.addCleanup(shutil.rmtree, tmp_dir)
         repo = Repo.init(tmp_dir)
-        self.assertEqual(os.path.join(tmp_dir, '.git'), repo._controldir)
+        self.assertEqual(os.path.join(tmp_dir_bytes, b'.git'), repo._controldir)
         self._check_repo_contents(repo, False)
 
-    def test_create_memory(self):
-        repo = MemoryRepo.init_bare([], {})
-        self._check_repo_contents(repo, True)
 
+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 RepositoryTests(TestCase):
+
+class RepositoryBytesRootTests(TestCase):
 
     def setUp(self):
-        super(RepositoryTests, self).setUp()
+        super(RepositoryBytesRootTests, self).setUp()
         self._repo = None
 
     def tearDown(self):
         if self._repo is not None:
             tear_down_repo(self._repo)
-        super(RepositoryTests, self).tearDown()
+        super(RepositoryBytesRootTests, self).tearDown()
+
+    def mkdtemp(self):
+        return mkdtemp_bytes()
+
+    def open_repo(self, name):
+        temp_dir = self.mkdtemp()
+        return open_repo(name, temp_dir)
 
     def test_simple_props(self):
-        r = self._repo = open_repo('a.git')
-        self.assertEqual(r.controldir(), r.path)
+        r = self._repo = self.open_repo('a.git')
+        self.assertEqual(r.controldir(), r._path_bytes)
 
     def test_setitem(self):
-        r = self._repo = open_repo('a.git')
+        r = self._repo = 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 = open_repo('a.git')
+        r = self._repo = self.open_repo('a.git')
 
         test_keys = [
             (b'refs/heads/master', True),
@@ -127,7 +171,7 @@ class RepositoryTests(TestCase):
             )
 
     def test_delitem(self):
-        r = self._repo = open_repo('a.git')
+        r = self._repo = self.open_repo('a.git')
 
         del r[b'refs/heads/master']
         self.assertRaises(KeyError, lambda: r[b'refs/heads/master'])
@@ -138,7 +182,7 @@ class RepositoryTests(TestCase):
         self.assertRaises(ValueError, r.__delitem__, b'notrefs/foo')
 
     def test_get_refs(self):
-        r = self._repo = open_repo('a.git')
+        r = self._repo = self.open_repo('a.git')
         self.assertEqual({
             b'HEAD': b'a90fa2d900a17e99b433217e988c4eb4a2e9a097',
             b'refs/heads/master': b'a90fa2d900a17e99b433217e988c4eb4a2e9a097',
@@ -147,49 +191,49 @@ class RepositoryTests(TestCase):
             }, r.get_refs())
 
     def test_head(self):
-        r = self._repo = open_repo('a.git')
+        r = self._repo = self.open_repo('a.git')
         self.assertEqual(r.head(), b'a90fa2d900a17e99b433217e988c4eb4a2e9a097')
 
     def test_get_object(self):
-        r = self._repo = open_repo('a.git')
+        r = self._repo = 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 = open_repo('a.git')
+        r = self._repo = self.open_repo('a.git')
         self.assertRaises(KeyError, r.get_object, missing_sha)
 
     def test_contains_object(self):
-        r = self._repo = open_repo('a.git')
+        r = self._repo = self.open_repo('a.git')
         self.assertTrue(r.head() in r)
 
     def test_contains_ref(self):
-        r = self._repo = open_repo('a.git')
+        r = self._repo = self.open_repo('a.git')
         self.assertTrue(b"HEAD" in r)
 
     def test_get_no_description(self):
-        r = self._repo = open_repo('a.git')
+        r = self._repo = self.open_repo('a.git')
         self.assertIs(None, r.get_description())
 
     def test_get_description(self):
-        r = self._repo = open_repo('a.git')
+        r = self._repo = 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 = open_repo('a.git')
+        r = self._repo = 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 = open_repo('a.git')
+        r = self._repo = self.open_repo('a.git')
         self.assertFalse(b"bar" in r)
 
     def test_get_peeled(self):
         # unpacked ref
-        r = self._repo = open_repo('a.git')
+        r = self._repo = 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())
@@ -203,11 +247,11 @@ class RepositoryTests(TestCase):
         # TODO: add more corner cases to test repo
 
     def test_get_peeled_not_tag(self):
-        r = self._repo = open_repo('a.git')
+        r = self._repo = self.open_repo('a.git')
         self.assertEqual(r.get_peeled(b'HEAD'), r.head())
 
     def test_get_walker(self):
-        r = self._repo = open_repo('a.git')
+        r = self._repo = 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'])
@@ -219,8 +263,8 @@ class RepositoryTests(TestCase):
             [b'2a72d929692c41d8554c07f6301757ba18a65d91'])
 
     def test_clone(self):
-        r = self._repo = open_repo('a.git')
-        tmp_dir = tempfile.mkdtemp()
+        r = self._repo = self.open_repo('a.git')
+        tmp_dir = self.mkdtemp()
         self.addCleanup(shutil.rmtree, tmp_dir)
         t = r.clone(tmp_dir, mkdir=False)
         self.assertEqual({
@@ -237,16 +281,20 @@ class RepositoryTests(TestCase):
                          b'2a72d929692c41d8554c07f6301757ba18a65d91'])
 
     def test_clone_no_head(self):
-        temp_dir = tempfile.mkdtemp()
+        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, 'a.git')
+        dest_dir = os.path.join(temp_dir_str, '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, 'b.git'), mkdir=True)
+        t = r.clone(os.path.join(temp_dir_str, 'b.git'), mkdir=True)
         self.assertEqual({
             b'refs/tags/mytag': b'28237f4dc30d0d462658d6b937b08a0f0b6ef55a',
             b'refs/tags/mytag-packed':
@@ -261,13 +309,13 @@ class RepositoryTests(TestCase):
         to the server.
         Non-bare repo HEAD always points to an existing ref.
         """
-        r = self._repo = open_repo('empty.git')
-        tmp_dir = tempfile.mkdtemp()
+        r = self._repo = 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 = open_repo('simple_merge.git')
+        r = self._repo = self.open_repo('simple_merge.git')
         shas = [e.commit.id for e in r.get_walker()]
         self.assertEqual(shas, [b'5dac377bdded4c9aeb8dff595f0faeebcc8498cc',
                                 b'ab64bbdcc51b170d21588e5c5d391ee5c0c96dfd',
@@ -277,7 +325,7 @@ class RepositoryTests(TestCase):
 
     def test_out_of_order_merge(self):
         """Test that revision history is ordered by date, not parent order."""
-        r = self._repo = open_repo('ooo_merge.git')
+        r = self._repo = self.open_repo('ooo_merge.git')
         shas = [e.commit.id for e in r.get_walker()]
         self.assertEqual(shas, [b'7601d7f6231db6a57f7bbb79ee52e4d462fd44d1',
                                 b'f507291b64138b875c28e03469025b1ea20bc614',
@@ -285,24 +333,28 @@ class RepositoryTests(TestCase):
                                 b'f9e39b120c68182a4ba35349f832d0e4e61f485c'])
 
     def test_get_tags_empty(self):
-        r = self._repo = open_repo('ooo_merge.git')
+        r = self._repo = self.open_repo('ooo_merge.git')
         self.assertEqual({}, r.refs.as_dict(b'refs/tags'))
 
     def test_get_config(self):
-        r = self._repo = open_repo('ooo_merge.git')
+        r = self._repo = self.open_repo('ooo_merge.git')
         self.assertIsInstance(r.get_config(), Config)
 
     def test_get_config_stack(self):
-        r = self._repo = open_repo('ooo_merge.git')
+        r = self._repo = self.open_repo('ooo_merge.git')
         self.assertIsInstance(r.get_config_stack(), Config)
 
     def test_submodule(self):
-        temp_dir = tempfile.mkdtemp()
+        temp_dir = self.mkdtemp()
         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, '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'))
+                        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')
 
@@ -317,19 +369,19 @@ class RepositoryTests(TestCase):
         expected_shas = set([b'60dacdc733de308bb77bb76ce0fb0f9b44c9769e'])
 
         # Source for objects.
-        r_base = open_repo('simple_merge.git')
+        r_base = self.open_repo('simple_merge.git')
 
         # Re-create each-side of the merge in simple_merge.git.
         #
         # Since the trees and blobs are missing, the repository created is
         # corrupted, but we're only checking for commits for the purpose of this
         # test, so it's immaterial.
-        r1_dir = tempfile.mkdtemp()
+        r1_dir = self.mkdtemp()
         r1_commits = [b'ab64bbdcc51b170d21588e5c5d391ee5c0c96dfd', # HEAD
                       b'60dacdc733de308bb77bb76ce0fb0f9b44c9769e',
                       b'0d89f20333fbb1d2f3a94da77f4981373d8f4310']
 
-        r2_dir = tempfile.mkdtemp()
+        r2_dir = self.mkdtemp()
         r2_commits = [b'4cffe90e0a41ad3f5190079d7c8f036bde29cbe6', # HEAD
                       b'60dacdc733de308bb77bb76ce0fb0f9b44c9769e',
                       b'0d89f20333fbb1d2f3a94da77f4981373d8f4310']
@@ -367,7 +419,7 @@ exit 1
 exit 0
 """
 
-        repo_dir = os.path.join(tempfile.mkdtemp())
+        repo_dir = os.path.join(self.mkdtemp())
         r = Repo.init(repo_dir)
         self.addCleanup(shutil.rmtree, repo_dir)
 
@@ -407,7 +459,7 @@ exit 1
 exit 0
 """
 
-        repo_dir = os.path.join(tempfile.mkdtemp())
+        repo_dir = os.path.join(self.mkdtemp())
         r = Repo.init(repo_dir)
         self.addCleanup(shutil.rmtree, repo_dir)
 
@@ -439,11 +491,16 @@ exit 0
         if os.name != 'posix':
             self.skipTest('shell hook tests requires POSIX shell')
 
-        repo_dir = os.path.join(tempfile.mkdtemp())
+        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)
+        (fd, path) = tempfile.mkstemp(dir=repo_dir_str)
         post_commit_msg = b"""#!/bin/sh
 rm """ + path.encode(sys.getfilesystemencoding()) + b"""
 """
@@ -496,16 +553,25 @@ exit 1
         self.assertEqual([commit_sha], r[commit_sha2].parents)
 
 
-class BuildRepoTests(TestCase):
+class RepositoryUnicodeRootTests(RepositoryBytesRootTests):
+
+    def mktemp(self):
+        return mkdtemp_unicode()
+
+
+class BuildRepoBytesRootTests(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_bytes(), b'test')
+
     def setUp(self):
-        super(BuildRepoTests, self).setUp()
-        self._repo_dir = os.path.join(tempfile.mkdtemp(), 'test')
+        super(BuildRepoBytesRootTests, self).setUp()
+        self._repo_dir = self.get_repo_dir()
         os.makedirs(self._repo_dir)
         r = self._repo = Repo.init(self._repo_dir)
         self.assertFalse(r.bare)
@@ -525,7 +591,7 @@ class BuildRepoTests(TestCase):
 
     def tearDown(self):
         tear_down_repo(self._repo)
-        super(BuildRepoTests, self).tearDown()
+        super(BuildRepoBytesRootTests, self).tearDown()
 
     def test_build_repo(self):
         r = self._repo
@@ -723,3 +789,13 @@ class BuildRepoTests(TestCase):
         r.stage(['a'])
         r.stage(['a'])  # double-stage a deleted path
 
+
+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')

dulwich/tests/utils.py

@@ -60,7 +60,7 @@ from dulwich.tests import (
 F = 0o100644  # Shorthand mode for Files.
 
 
-def open_repo(name):
+def open_repo(name, temp_dir=None):
     """Open a copy of a repo in a temporary directory.
 
     Use this function for accessing repos in dulwich/tests/data/repos to avoid
@@ -69,6 +69,8 @@ def open_repo(name):
 
     :param name: The name of the repository, relative to
         dulwich/tests/data/repos
+    :temp_dir: temporary directory to initialize to. If not provided, a
+        temporary directory will be created.
     :returns: An initialized Repo object that lives in a temporary directory.
     """
     temp_dir = tempfile.mkdtemp()