diff --git a/heptapod/hooks/perm.py b/heptapod/hooks/perm.py index 5f58ee03b4790e2b63bbfad357516c00b123f0dd_aGVwdGFwb2QvaG9va3MvcGVybS5weQ==..4c68254d61e9b2e0950b8928e7d51adb916cfb22_aGVwdGFwb2QvaG9va3MvcGVybS5weQ== 100644 --- a/heptapod/hooks/perm.py +++ b/heptapod/hooks/perm.py @@ -85,6 +85,30 @@ return False +def check_write(repo, *args, **kwargs): + """Check that remote user has write privileges. + + In this hook, the very fact to be called serves as detection that a + write operation will happen. + + Therefore the implementation is very straightforward. + """ + validate_hook_type('pretxnopen', **kwargs) + + remote_user = get_remote_user(repo) + if remote_user is not None: + repo.ui.debug( + 'check_write detected push from user: %r\n' % remote_user) + + if allowed(repo, remote_user, 'web', 'allow-push'): + return 0 + + msg = 'user %r does not have write permission' % remote_user + repo.ui.note(msg) + repo.ui.status(msg) + return 1 + + def check_publish(repo, *args, **kwargs): validate_hook_type('pretxnclose', **kwargs) remoteuser = get_remote_user(repo) diff --git a/heptapod/hooks/tests/test_check_write.py b/heptapod/hooks/tests/test_check_write.py new file mode 100644 index 0000000000000000000000000000000000000000..4c68254d61e9b2e0950b8928e7d51adb916cfb22_aGVwdGFwb2QvaG9va3MvdGVzdHMvdGVzdF9jaGVja193cml0ZS5weQ== --- /dev/null +++ b/heptapod/hooks/tests/test_check_write.py @@ -0,0 +1,60 @@ +from __future__ import absolute_import +import pytest +from mercurial import ( + error, +) +from heptapod.testhelpers import ( + LocalRepoWrapper, + ) + +from .utils import switch_user + + +def init_repo(basedir): + return LocalRepoWrapper.init( + basedir, + config=dict( + hooks={'pretxnopen.check_write': + 'python:heptapod.hooks.perm.check_write' + }, + web={'allow-push': 'developer', + }, + )) + + +def test_write_empty_repo(tmpdir): + wrapper = init_repo(tmpdir) + with pytest.raises(error.HookAbort) as exc_info: + wrapper.write_commit('foo', content='Foo', return_ctx=True) + + assert 'pretxnopen.check_write' in exc_info.value.args[0] + switch_user(wrapper, 'developer') + ctx = wrapper.write_commit('foo', message='developer is accepted', + return_ctx=True) + assert ctx.description() == 'developer is accepted' + + +def test_phase_change(tmpdir): + wrapper = init_repo(tmpdir) + switch_user(wrapper, 'developer') + ctx = wrapper.write_commit('foo', content='Foo', return_ctx=True) + + switch_user(wrapper, 'guest') + with pytest.raises(error.HookAbort) as exc_info: + wrapper.set_phase('public', [ctx.hex()]) + assert 'pretxnopen.check_write' in exc_info.value.args[0] + + +def test_wrong_hook(tmpdir): + wrapper = init_repo(tmpdir) + ui = wrapper.repo.ui + pretxn = 'pretxnopen.check_write' + hookdef = ui.config('hooks', pretxn) + ui.setconfig('hooks', pretxn, '') + ui.setconfig('hooks', 'precommit.check_write', hookdef) + # precommit because that one does not swallow exceptions other + # than abort + with pytest.raises(error.ProgrammingError) as exc_info: + wrapper.write_commit('foo') + + assert 'precommit' in exc_info.value.args[0]