|
|
# RhodeCode VCSServer provides access to different vcs backends via network.
|
|
|
# Copyright (C) 2014-2020 RhodeCode GmbH
|
|
|
#
|
|
|
# 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 3 of the License, or
|
|
|
# (at your option) any later version.
|
|
|
#
|
|
|
# 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
|
|
|
|
|
|
import io
|
|
|
|
|
|
import dulwich.protocol
|
|
|
import mock
|
|
|
import pytest
|
|
|
import webob
|
|
|
import webtest
|
|
|
|
|
|
from vcsserver import hooks, pygrack
|
|
|
|
|
|
# pylint: disable=redefined-outer-name,protected-access
|
|
|
|
|
|
|
|
|
@pytest.fixture()
|
|
|
def pygrack_instance(tmpdir):
|
|
|
"""
|
|
|
Creates a pygrack app instance.
|
|
|
|
|
|
Right now, it does not much helpful regarding the passed directory.
|
|
|
It just contains the required folders to pass the signature test.
|
|
|
"""
|
|
|
for dir_name in ('config', 'head', 'info', 'objects', 'refs'):
|
|
|
tmpdir.mkdir(dir_name)
|
|
|
|
|
|
return pygrack.GitRepository('repo_name', str(tmpdir), 'git', False, {})
|
|
|
|
|
|
|
|
|
@pytest.fixture()
|
|
|
def pygrack_app(pygrack_instance):
|
|
|
"""
|
|
|
Creates a pygrack app wrapped in webtest.TestApp.
|
|
|
"""
|
|
|
return webtest.TestApp(pygrack_instance)
|
|
|
|
|
|
|
|
|
def test_invalid_service_info_refs_returns_403(pygrack_app):
|
|
|
response = pygrack_app.get('/info/refs?service=git-upload-packs',
|
|
|
expect_errors=True)
|
|
|
|
|
|
assert response.status_int == 403
|
|
|
|
|
|
|
|
|
def test_invalid_endpoint_returns_403(pygrack_app):
|
|
|
response = pygrack_app.post('/git-upload-packs', expect_errors=True)
|
|
|
|
|
|
assert response.status_int == 403
|
|
|
|
|
|
|
|
|
@pytest.mark.parametrize('sideband', [
|
|
|
'side-band-64k',
|
|
|
'side-band',
|
|
|
'side-band no-progress',
|
|
|
])
|
|
|
def test_pre_pull_hook_fails_with_sideband(pygrack_app, sideband):
|
|
|
request = ''.join([
|
|
|
'0054want 74730d410fcb6603ace96f1dc55ea6196122532d ',
|
|
|
'multi_ack %s ofs-delta\n' % sideband,
|
|
|
'0000',
|
|
|
'0009done\n',
|
|
|
])
|
|
|
with mock.patch('vcsserver.hooks.git_pre_pull',
|
|
|
return_value=hooks.HookResponse(1, 'foo')):
|
|
|
response = pygrack_app.post(
|
|
|
'/git-upload-pack', params=request,
|
|
|
content_type='application/x-git-upload-pack')
|
|
|
|
|
|
data = io.BytesIO(response.body)
|
|
|
proto = dulwich.protocol.Protocol(data.read, None)
|
|
|
packets = list(proto.read_pkt_seq())
|
|
|
|
|
|
expected_packets = [
|
|
|
'NAK\n', '\x02foo', '\x02Pre pull hook failed: aborting\n',
|
|
|
'\x01' + pygrack.GitRepository.EMPTY_PACK,
|
|
|
]
|
|
|
assert packets == expected_packets
|
|
|
|
|
|
|
|
|
def test_pre_pull_hook_fails_no_sideband(pygrack_app):
|
|
|
request = ''.join([
|
|
|
'0054want 74730d410fcb6603ace96f1dc55ea6196122532d ' +
|
|
|
'multi_ack ofs-delta\n'
|
|
|
'0000',
|
|
|
'0009done\n',
|
|
|
])
|
|
|
with mock.patch('vcsserver.hooks.git_pre_pull',
|
|
|
return_value=hooks.HookResponse(1, 'foo')):
|
|
|
response = pygrack_app.post(
|
|
|
'/git-upload-pack', params=request,
|
|
|
content_type='application/x-git-upload-pack')
|
|
|
|
|
|
assert response.body == pygrack.GitRepository.EMPTY_PACK
|
|
|
|
|
|
|
|
|
def test_pull_has_hook_messages(pygrack_app):
|
|
|
request = ''.join([
|
|
|
'0054want 74730d410fcb6603ace96f1dc55ea6196122532d ' +
|
|
|
'multi_ack side-band-64k ofs-delta\n'
|
|
|
'0000',
|
|
|
'0009done\n',
|
|
|
])
|
|
|
with mock.patch('vcsserver.hooks.git_pre_pull',
|
|
|
return_value=hooks.HookResponse(0, 'foo')):
|
|
|
with mock.patch('vcsserver.hooks.git_post_pull',
|
|
|
return_value=hooks.HookResponse(1, 'bar')):
|
|
|
with mock.patch('vcsserver.subprocessio.SubprocessIOChunker',
|
|
|
return_value=['0008NAK\n0009subp\n0000']):
|
|
|
response = pygrack_app.post(
|
|
|
'/git-upload-pack', params=request,
|
|
|
content_type='application/x-git-upload-pack')
|
|
|
|
|
|
data = io.BytesIO(response.body)
|
|
|
proto = dulwich.protocol.Protocol(data.read, None)
|
|
|
packets = list(proto.read_pkt_seq())
|
|
|
|
|
|
assert packets == ['NAK\n', '\x02foo', 'subp\n', '\x02bar']
|
|
|
|
|
|
|
|
|
def test_get_want_capabilities(pygrack_instance):
|
|
|
data = io.BytesIO(
|
|
|
'0054want 74730d410fcb6603ace96f1dc55ea6196122532d ' +
|
|
|
'multi_ack side-band-64k ofs-delta\n00000009done\n')
|
|
|
|
|
|
request = webob.Request({
|
|
|
'wsgi.input': data,
|
|
|
'REQUEST_METHOD': 'POST',
|
|
|
'webob.is_body_seekable': True
|
|
|
})
|
|
|
|
|
|
capabilities = pygrack_instance._get_want_capabilities(request)
|
|
|
|
|
|
assert capabilities == frozenset(
|
|
|
('ofs-delta', 'multi_ack', 'side-band-64k'))
|
|
|
assert data.tell() == 0
|
|
|
|
|
|
|
|
|
@pytest.mark.parametrize('data,capabilities,expected', [
|
|
|
('foo', [], []),
|
|
|
('', ['side-band-64k'], []),
|
|
|
('', ['side-band'], []),
|
|
|
('foo', ['side-band-64k'], ['0008\x02foo']),
|
|
|
('foo', ['side-band'], ['0008\x02foo']),
|
|
|
('f'*1000, ['side-band-64k'], ['03ed\x02' + 'f' * 1000]),
|
|
|
('f'*1000, ['side-band'], ['03e8\x02' + 'f' * 995, '000a\x02fffff']),
|
|
|
('f'*65520, ['side-band-64k'], ['fff0\x02' + 'f' * 65515, '000a\x02fffff']),
|
|
|
('f'*65520, ['side-band'], ['03e8\x02' + 'f' * 995] * 65 + ['0352\x02' + 'f' * 845]),
|
|
|
], ids=[
|
|
|
'foo-empty',
|
|
|
'empty-64k', 'empty',
|
|
|
'foo-64k', 'foo',
|
|
|
'f-1000-64k', 'f-1000',
|
|
|
'f-65520-64k', 'f-65520'])
|
|
|
def test_get_messages(pygrack_instance, data, capabilities, expected):
|
|
|
messages = pygrack_instance._get_messages(data, capabilities)
|
|
|
|
|
|
assert messages == expected
|
|
|
|
|
|
|
|
|
@pytest.mark.parametrize('response,capabilities,pre_pull_messages,post_pull_messages', [
|
|
|
# Unexpected response
|
|
|
('unexpected_response', ['side-band-64k'], 'foo', 'bar'),
|
|
|
# No sideband
|
|
|
('no-sideband', [], 'foo', 'bar'),
|
|
|
# No messages
|
|
|
('no-messages', ['side-band-64k'], '', ''),
|
|
|
])
|
|
|
def test_inject_messages_to_response_nothing_to_do(
|
|
|
pygrack_instance, response, capabilities, pre_pull_messages,
|
|
|
post_pull_messages):
|
|
|
new_response = pygrack_instance._inject_messages_to_response(
|
|
|
response, capabilities, pre_pull_messages, post_pull_messages)
|
|
|
|
|
|
assert new_response == response
|
|
|
|
|
|
|
|
|
@pytest.mark.parametrize('capabilities', [
|
|
|
['side-band'],
|
|
|
['side-band-64k'],
|
|
|
])
|
|
|
def test_inject_messages_to_response_single_element(pygrack_instance,
|
|
|
capabilities):
|
|
|
response = ['0008NAK\n0009subp\n0000']
|
|
|
new_response = pygrack_instance._inject_messages_to_response(
|
|
|
response, capabilities, 'foo', 'bar')
|
|
|
|
|
|
expected_response = [
|
|
|
'0008NAK\n', '0008\x02foo', '0009subp\n', '0008\x02bar', '0000']
|
|
|
|
|
|
assert new_response == expected_response
|
|
|
|
|
|
|
|
|
@pytest.mark.parametrize('capabilities', [
|
|
|
['side-band'],
|
|
|
['side-band-64k'],
|
|
|
])
|
|
|
def test_inject_messages_to_response_multi_element(pygrack_instance,
|
|
|
capabilities):
|
|
|
response = [
|
|
|
'0008NAK\n000asubp1\n', '000asubp2\n', '000asubp3\n', '000asubp4\n0000']
|
|
|
new_response = pygrack_instance._inject_messages_to_response(
|
|
|
response, capabilities, 'foo', 'bar')
|
|
|
|
|
|
expected_response = [
|
|
|
'0008NAK\n', '0008\x02foo', '000asubp1\n', '000asubp2\n', '000asubp3\n',
|
|
|
'000asubp4\n', '0008\x02bar', '0000'
|
|
|
]
|
|
|
|
|
|
assert new_response == expected_response
|
|
|
|
|
|
|
|
|
def test_build_failed_pre_pull_response_no_sideband(pygrack_instance):
|
|
|
response = pygrack_instance._build_failed_pre_pull_response([], 'foo')
|
|
|
|
|
|
assert response == [pygrack.GitRepository.EMPTY_PACK]
|
|
|
|
|
|
|
|
|
@pytest.mark.parametrize('capabilities', [
|
|
|
['side-band'],
|
|
|
['side-band-64k'],
|
|
|
['side-band-64k', 'no-progress'],
|
|
|
])
|
|
|
def test_build_failed_pre_pull_response(pygrack_instance, capabilities):
|
|
|
response = pygrack_instance._build_failed_pre_pull_response(
|
|
|
capabilities, 'foo')
|
|
|
|
|
|
expected_response = [
|
|
|
'0008NAK\n', '0008\x02foo', '0024\x02Pre pull hook failed: aborting\n',
|
|
|
'%04x\x01%s' % (len(pygrack.GitRepository.EMPTY_PACK) + 5,
|
|
|
pygrack.GitRepository.EMPTY_PACK),
|
|
|
'0000',
|
|
|
]
|
|
|
|
|
|
assert response == expected_response
|
|
|
|
