vcsserver/tests/test_pygrack.py

# 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