Add shell hook tests

dulwich/tests/__init__.py

@@ -117,6 +117,7 @@ def self_test_suite():
         'diff_tree',
         'fastexport',
         'file',
+        'hooks',
         'index',
         'lru_cache',
         'objects',

dulwich/tests/test_hooks.py

@@ -0,0 +1,147 @@
+# test_hooks.py -- Tests for executing hooks
+#
+# 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; either version 2
+# or (at your option) a later version of the License.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program; if not, write to the Free Software
+# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston,
+# MA  02110-1301, USA.
+
+"""Tests for executing hooks."""
+
+import os
+import stat
+import shutil
+import tempfile
+import warnings
+
+from dulwich import errors
+
+from dulwich.hooks import (
+    PreCommitShellHook,
+    PostCommitShellHook,
+    CommitMsgShellHook,
+)
+
+from dulwich.tests import TestCase
+
+
+class ShellHookTests(TestCase):
+
+    def setUp(self):
+        if os.name != 'posix':
+            self.skipTest('shell hook tests requires POSIX shell')
+
+    def test_hook_pre_commit(self):
+        pre_commit_fail = """#!/bin/sh
+exit 1
+"""
+
+        pre_commit_success = """#!/bin/sh
+exit 0
+"""
+
+        repo_dir = os.path.join(tempfile.mkdtemp())
+        os.mkdir(os.path.join(repo_dir, 'hooks'))
+        self.addCleanup(shutil.rmtree, repo_dir)
+
+        pre_commit = os.path.join(repo_dir, 'hooks', 'pre-commit')
+        hook = PreCommitShellHook(repo_dir)
+
+        f = open(pre_commit, 'wb')
+        try:
+            f.write(pre_commit_fail)
+        finally:
+            f.close()
+        os.chmod(pre_commit, stat.S_IREAD | stat.S_IWRITE | stat.S_IEXEC)
+
+        self.assertRaises(errors.HookError, hook.execute)
+
+        f = open(pre_commit, 'wb')
+        try:
+            f.write(pre_commit_success)
+        finally:
+            f.close()
+        os.chmod(pre_commit, stat.S_IREAD | stat.S_IWRITE | stat.S_IEXEC)
+
+        hook.execute()
+
+    def test_hook_commit_msg(self):
+
+        commit_msg_fail = """#!/bin/sh
+exit 1
+"""
+
+        commit_msg_success = """#!/bin/sh
+exit 0
+"""
+
+        repo_dir = os.path.join(tempfile.mkdtemp())
+        os.mkdir(os.path.join(repo_dir, 'hooks'))
+        self.addCleanup(shutil.rmtree, repo_dir)
+
+        commit_msg = os.path.join(repo_dir, 'hooks', 'commit-msg')
+        hook = CommitMsgShellHook(repo_dir)
+
+        f = open(commit_msg, 'wb')
+        try:
+            f.write(commit_msg_fail)
+        finally:
+            f.close()
+        os.chmod(commit_msg, stat.S_IREAD | stat.S_IWRITE | stat.S_IEXEC)
+
+        self.assertRaises(errors.HookError, hook.execute, 'failed commit')
+
+        f = open(commit_msg, 'wb')
+        try:
+            f.write(commit_msg_success)
+        finally:
+            f.close()
+        os.chmod(commit_msg, stat.S_IREAD | stat.S_IWRITE | stat.S_IEXEC)
+
+        hook.execute('empty commit')
+
+    def test_hook_post_commit(self):
+
+        (fd, path) = tempfile.mkstemp()
+        post_commit_msg = """#!/bin/sh
+unlink %(file)s
+""" % {'file': path}
+
+        post_commit_msg_fail = """#!/bin/sh
+exit 1
+"""
+
+        repo_dir = os.path.join(tempfile.mkdtemp())
+        os.mkdir(os.path.join(repo_dir, 'hooks'))
+        self.addCleanup(shutil.rmtree, repo_dir)
+
+        post_commit = os.path.join(repo_dir, 'hooks', 'post-commit')
+        hook = PostCommitShellHook(repo_dir)
+
+        f = open(post_commit, 'wb')
+        try:
+            f.write(post_commit_msg_fail)
+        finally:
+            f.close()
+        os.chmod(post_commit, stat.S_IREAD | stat.S_IWRITE | stat.S_IEXEC)
+
+        self.assertRaises(errors.HookError, hook.execute)
+
+        f = open(post_commit, 'wb')
+        try:
+            f.write(post_commit_msg)
+        finally:
+            f.close()
+        os.chmod(post_commit, stat.S_IREAD | stat.S_IWRITE | stat.S_IEXEC)
+
+        hook.execute()
+        self.assertFalse(os.path.exists(path))

dulwich/tests/test_repository.py

@@ -21,6 +21,7 @@
 
 from cStringIO import StringIO
 import os
+import stat
 import shutil
 import tempfile
 import warnings
@@ -51,6 +52,7 @@ from dulwich.tests import (
 from dulwich.tests.utils import (
     open_repo,
     tear_down_repo,
+    setup_warning_catcher,
     )
 
 missing_sha = 'b91fa4d900e17e99b433218e988c4eb4a3e9a097'
@@ -412,6 +414,163 @@ class RepositoryTests(TestCase):
             shutil.rmtree(r1_dir)
             shutil.rmtree(r2_dir)
 
+    def test_shell_hook_pre_commit(self):
+        if os.name != 'posix':
+            self.skipTest('shell hook tests requires POSIX shell')
+
+        pre_commit_fail = """#!/bin/sh
+exit 1
+"""
+
+        pre_commit_success = """#!/bin/sh
+exit 0
+"""
+
+        repo_dir = os.path.join(tempfile.mkdtemp())
+        r = Repo.init(repo_dir)
+        self.addCleanup(shutil.rmtree, repo_dir)
+
+        pre_commit = os.path.join(r.controldir(), 'hooks', 'pre-commit')
+
+        f = open(pre_commit, 'wb')
+        try:
+            f.write(pre_commit_fail)
+        finally:
+            f.close()
+        os.chmod(pre_commit, stat.S_IREAD | stat.S_IWRITE | stat.S_IEXEC)
+
+        self.assertRaises(errors.CommitError, r.do_commit, 'failed commit',
+                          committer='Test Committer <test@nodomain.com>',
+                          author='Test Author <test@nodomain.com>',
+                          commit_timestamp=12345, commit_timezone=0,
+                          author_timestamp=12345, author_timezone=0)
+
+        f = open(pre_commit, 'wb')
+        try:
+            f.write(pre_commit_success)
+        finally:
+            f.close()
+        os.chmod(pre_commit, stat.S_IREAD | stat.S_IWRITE | stat.S_IEXEC)
+
+        commit_sha = r.do_commit(
+            'empty commit',
+            committer='Test Committer <test@nodomain.com>',
+            author='Test Author <test@nodomain.com>',
+            commit_timestamp=12395, commit_timezone=0,
+            author_timestamp=12395, author_timezone=0)
+        self.assertEqual([], r[commit_sha].parents)
+
+    def test_shell_hook_commit_msg(self):
+        if os.name != 'posix':
+            self.skipTest('shell hook tests requires POSIX shell')
+
+        commit_msg_fail = """#!/bin/sh
+exit 1
+"""
+
+        commit_msg_success = """#!/bin/sh
+exit 0
+"""
+
+        repo_dir = os.path.join(tempfile.mkdtemp())
+        r = Repo.init(repo_dir)
+        self.addCleanup(shutil.rmtree, repo_dir)
+
+        commit_msg = os.path.join(r.controldir(), 'hooks', 'commit-msg')
+
+        f = open(commit_msg, 'wb')
+        try:
+            f.write(commit_msg_fail)
+        finally:
+            f.close()
+        os.chmod(commit_msg, stat.S_IREAD | stat.S_IWRITE | stat.S_IEXEC)
+
+        self.assertRaises(errors.CommitError, r.do_commit, 'failed commit',
+                          committer='Test Committer <test@nodomain.com>',
+                          author='Test Author <test@nodomain.com>',
+                          commit_timestamp=12345, commit_timezone=0,
+                          author_timestamp=12345, author_timezone=0)
+
+        f = open(commit_msg, 'wb')
+        try:
+            f.write(commit_msg_success)
+        finally:
+            f.close()
+        os.chmod(commit_msg, stat.S_IREAD | stat.S_IWRITE | stat.S_IEXEC)
+
+        commit_sha = r.do_commit(
+            'empty commit',
+            committer='Test Committer <test@nodomain.com>',
+            author='Test Author <test@nodomain.com>',
+            commit_timestamp=12395, commit_timezone=0,
+            author_timestamp=12395, author_timezone=0)
+        self.assertEqual([], r[commit_sha].parents)
+
+    def test_shell_hook_post_commit(self):
+        if os.name != 'posix':
+            self.skipTest('shell hook tests requires POSIX shell')
+
+        repo_dir = os.path.join(tempfile.mkdtemp())
+        r = Repo.init(repo_dir)
+        self.addCleanup(shutil.rmtree, repo_dir)
+
+        (fd, path) = tempfile.mkstemp(dir=repo_dir)
+        post_commit_msg = """#!/bin/sh
+unlink %(file)s
+""" % {'file': path}
+
+        root_sha = r.do_commit(
+            'empty commit',
+            committer='Test Committer <test@nodomain.com>',
+            author='Test Author <test@nodomain.com>',
+            commit_timestamp=12345, commit_timezone=0,
+            author_timestamp=12345, author_timezone=0)
+        self.assertEqual([], r[root_sha].parents)
+
+        post_commit = os.path.join(r.controldir(), 'hooks', 'post-commit')
+
+        f = open(post_commit, 'wb')
+        try:
+            f.write(post_commit_msg)
+        finally:
+            f.close()
+        os.chmod(post_commit, stat.S_IREAD | stat.S_IWRITE | stat.S_IEXEC)
+
+        commit_sha = r.do_commit(
+            'empty commit',
+            committer='Test Committer <test@nodomain.com>',
+            author='Test Author <test@nodomain.com>',
+            commit_timestamp=12345, commit_timezone=0,
+            author_timestamp=12345, author_timezone=0)
+        self.assertEqual([root_sha], r[commit_sha].parents)
+
+        self.assertFalse(os.path.exists(path))
+
+        post_commit_msg_fail = """#!/bin/sh
+exit 1
+"""
+        f = open(post_commit, 'wb')
+        try:
+            f.write(post_commit_msg_fail)
+        finally:
+            f.close()
+        os.chmod(post_commit, stat.S_IREAD | stat.S_IWRITE | stat.S_IEXEC)
+
+        warnings.simplefilter("always", UserWarning)
+        self.addCleanup(warnings.resetwarnings)
+        warnings_list = setup_warning_catcher()
+
+        commit_sha2 = r.do_commit(
+            'empty commit',
+            committer='Test Committer <test@nodomain.com>',
+            author='Test Author <test@nodomain.com>',
+            commit_timestamp=12345, commit_timezone=0,
+            author_timestamp=12345, author_timezone=0)
+        self.assertEqual(len(warnings_list), 1)
+        self.assertIsInstance(warnings_list[-1], UserWarning)
+        self.assertTrue("post-commit hook failed: " in str(warnings_list[-1]))
+        self.assertEqual([commit_sha], r[commit_sha2].parents)
+
 
 class BuildRepoTests(TestCase):
     """Tests that build on-disk repos from scratch.

dulwich/tests/utils.py

@@ -26,6 +26,7 @@ import shutil
 import tempfile
 import time
 import types
+import warnings
 
 from dulwich.index import (
     commit_tree,
@@ -310,3 +311,16 @@ def build_commit_graph(object_store, commit_spec, trees=None, attrs=None):
         commits.append(commit_obj)
 
     return commits
+
+
+def setup_warning_catcher():
+    """Wrap warnings.showwarning with code that records warnings."""
+
+    caught_warnings = []
+    original_showwarning = warnings.showwarning
+
+    def custom_showwarning(*args,  **kwargs):
+        caught_warnings.append(args[0])
+
+    warnings.showwarning = custom_showwarning
+    return caught_warnings