|
|
# -*- coding: utf-8 -*-
|
|
|
|
|
|
# Copyright (C) 2016-2017 RhodeCode GmbH
|
|
|
#
|
|
|
# This program is free software: you can redistribute it and/or modify
|
|
|
# it under the terms of the GNU Affero General Public License, version 3
|
|
|
# (only), as published by the Free Software Foundation.
|
|
|
#
|
|
|
# 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 Affero General Public License
|
|
|
# along with this program. If not, see <http://www.gnu.org/licenses/>.
|
|
|
#
|
|
|
# This program is dual-licensed. If you wish to learn more about the
|
|
|
# RhodeCode Enterprise Edition, including its added features, Support services,
|
|
|
# and proprietary license terms, please see https://rhodecode.com/licenses/
|
|
|
|
|
|
|
|
|
import subprocess
|
|
|
from io import BytesIO
|
|
|
from time import sleep
|
|
|
|
|
|
import pytest
|
|
|
from mock import patch, Mock, MagicMock, call
|
|
|
|
|
|
from rhodecode.apps.ssh_support.lib.ssh_wrapper import SubversionTunnelWrapper
|
|
|
from rhodecode.tests import no_newline_id_generator
|
|
|
|
|
|
|
|
|
class TestSubversionTunnelWrapper(object):
|
|
|
@pytest.mark.parametrize(
|
|
|
'input_string, output_string', [
|
|
|
[None, ''],
|
|
|
['abcde', '5:abcde '],
|
|
|
['abcdefghijk', '11:abcdefghijk ']
|
|
|
])
|
|
|
def test_svn_string(self, input_string, output_string):
|
|
|
wrapper = SubversionTunnelWrapper(timeout=5)
|
|
|
assert wrapper._svn_string(input_string) == output_string
|
|
|
|
|
|
def test_read_first_client_response(self):
|
|
|
wrapper = SubversionTunnelWrapper(timeout=5)
|
|
|
buffer_ = '( abcd ( efg hij ) ) '
|
|
|
wrapper.stdin = BytesIO(buffer_)
|
|
|
result = wrapper._read_first_client_response()
|
|
|
assert result == buffer_
|
|
|
|
|
|
def test_parse_first_client_response_returns_dict(self):
|
|
|
response = (
|
|
|
'( 2 ( edit-pipeline svndiff1 absent-entries depth mergeinfo'
|
|
|
' log-revprops ) 26:svn+ssh://vcs@vm/hello-svn 38:SVN/1.8.11'
|
|
|
' (x86_64-apple-darwin14.1.0) ( ) ) ')
|
|
|
wrapper = SubversionTunnelWrapper(timeout=5)
|
|
|
result = wrapper._parse_first_client_response(response)
|
|
|
assert result['version'] == '2'
|
|
|
assert (
|
|
|
result['capabilities'] ==
|
|
|
'edit-pipeline svndiff1 absent-entries depth mergeinfo'
|
|
|
' log-revprops')
|
|
|
assert result['url'] == 'svn+ssh://vcs@vm/hello-svn'
|
|
|
assert result['ra_client'] == 'SVN/1.8.11 (x86_64-apple-darwin14.1.0)'
|
|
|
assert result['client'] is None
|
|
|
|
|
|
def test_parse_first_client_response_returns_none_when_not_matched(self):
|
|
|
response = (
|
|
|
'( 2 ( edit-pipeline svndiff1 absent-entries depth mergeinfo'
|
|
|
' log-revprops ) ) ')
|
|
|
wrapper = SubversionTunnelWrapper(timeout=5)
|
|
|
result = wrapper._parse_first_client_response(response)
|
|
|
assert result is None
|
|
|
|
|
|
def test_interrupt(self):
|
|
|
wrapper = SubversionTunnelWrapper(timeout=5)
|
|
|
with patch.object(wrapper, 'fail') as fail_mock:
|
|
|
wrapper.interrupt(1, 'frame')
|
|
|
fail_mock.assert_called_once_with("Exited by timeout")
|
|
|
|
|
|
def test_fail(self):
|
|
|
process_mock = Mock()
|
|
|
wrapper = SubversionTunnelWrapper(timeout=5)
|
|
|
with patch.object(wrapper, 'remove_configs') as remove_configs_mock:
|
|
|
with patch('sys.stdout', new_callable=BytesIO) as stdout_mock:
|
|
|
with patch.object(wrapper, 'process') as process_mock:
|
|
|
wrapper.fail('test message')
|
|
|
assert (
|
|
|
stdout_mock.getvalue() ==
|
|
|
'( failure ( ( 210005 12:test message 0: 0 ) ) )\n')
|
|
|
process_mock.kill.assert_called_once_with()
|
|
|
remove_configs_mock.assert_called_once_with()
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
|
'client, expected_client', [
|
|
|
['test ', 'test '],
|
|
|
['', ''],
|
|
|
[None, '']
|
|
|
])
|
|
|
def test_client_in_patch_first_client_response(
|
|
|
self, client, expected_client):
|
|
|
response = {
|
|
|
'version': 2,
|
|
|
'capabilities': 'edit-pipeline svndiff1 absent-entries depth',
|
|
|
'url': 'svn+ssh://example.com/svn',
|
|
|
'ra_client': 'SVN/1.8.11 (x86_64-apple-darwin14.1.0)',
|
|
|
'client': client
|
|
|
}
|
|
|
wrapper = SubversionTunnelWrapper(timeout=5)
|
|
|
stdin = BytesIO()
|
|
|
with patch.object(wrapper, 'process') as process_mock:
|
|
|
process_mock.stdin = stdin
|
|
|
wrapper.patch_first_client_response(response)
|
|
|
assert (
|
|
|
stdin.getvalue() ==
|
|
|
'( 2 ( edit-pipeline svndiff1 absent-entries depth )'
|
|
|
' 25:svn+ssh://example.com/svn 38:SVN/1.8.11'
|
|
|
' (x86_64-apple-darwin14.1.0) ( {expected_client}) ) '.format(
|
|
|
expected_client=expected_client))
|
|
|
|
|
|
def test_kwargs_override_data_in_patch_first_client_response(self):
|
|
|
response = {
|
|
|
'version': 2,
|
|
|
'capabilities': 'edit-pipeline svndiff1 absent-entries depth',
|
|
|
'url': 'svn+ssh://example.com/svn',
|
|
|
'ra_client': 'SVN/1.8.11 (x86_64-apple-darwin14.1.0)',
|
|
|
'client': 'test'
|
|
|
}
|
|
|
wrapper = SubversionTunnelWrapper(timeout=5)
|
|
|
stdin = BytesIO()
|
|
|
with patch.object(wrapper, 'process') as process_mock:
|
|
|
process_mock.stdin = stdin
|
|
|
wrapper.patch_first_client_response(
|
|
|
response, version=3, client='abcde ',
|
|
|
capabilities='absent-entries depth',
|
|
|
url='svn+ssh://example.org/test',
|
|
|
ra_client='SVN/1.8.12 (ubuntu 14.04)')
|
|
|
assert (
|
|
|
stdin.getvalue() ==
|
|
|
'( 3 ( absent-entries depth ) 26:svn+ssh://example.org/test'
|
|
|
' 25:SVN/1.8.12 (ubuntu 14.04) ( abcde ) ) ')
|
|
|
|
|
|
def test_patch_first_client_response_sets_environment(self):
|
|
|
response = {
|
|
|
'version': 2,
|
|
|
'capabilities': 'edit-pipeline svndiff1 absent-entries depth',
|
|
|
'url': 'svn+ssh://example.com/svn',
|
|
|
'ra_client': 'SVN/1.8.11 (x86_64-apple-darwin14.1.0)',
|
|
|
'client': 'test'
|
|
|
}
|
|
|
wrapper = SubversionTunnelWrapper(timeout=5)
|
|
|
stdin = BytesIO()
|
|
|
with patch.object(wrapper, 'create_hooks_env') as create_hooks_mock:
|
|
|
with patch.object(wrapper, 'process') as process_mock:
|
|
|
process_mock.stdin = stdin
|
|
|
wrapper.patch_first_client_response(response)
|
|
|
create_hooks_mock.assert_called_once_with()
|
|
|
|
|
|
def test_get_first_client_response_exits_by_signal(self):
|
|
|
wrapper = SubversionTunnelWrapper(timeout=1)
|
|
|
read_patch = patch.object(wrapper, '_read_first_client_response')
|
|
|
parse_patch = patch.object(wrapper, '_parse_first_client_response')
|
|
|
interrupt_patch = patch.object(wrapper, 'interrupt')
|
|
|
|
|
|
with read_patch as read_mock, parse_patch as parse_mock, \
|
|
|
interrupt_patch as interrupt_mock:
|
|
|
read_mock.side_effect = lambda: sleep(3)
|
|
|
wrapper.get_first_client_response()
|
|
|
|
|
|
assert parse_mock.call_count == 0
|
|
|
assert interrupt_mock.call_count == 1
|
|
|
|
|
|
def test_get_first_client_response_parses_data(self):
|
|
|
wrapper = SubversionTunnelWrapper(timeout=5)
|
|
|
response = (
|
|
|
'( 2 ( edit-pipeline svndiff1 absent-entries depth mergeinfo'
|
|
|
' log-revprops ) 26:svn+ssh://vcs@vm/hello-svn 38:SVN/1.8.11'
|
|
|
' (x86_64-apple-darwin14.1.0) ( ) ) ')
|
|
|
read_patch = patch.object(wrapper, '_read_first_client_response')
|
|
|
parse_patch = patch.object(wrapper, '_parse_first_client_response')
|
|
|
|
|
|
with read_patch as read_mock, parse_patch as parse_mock:
|
|
|
read_mock.return_value = response
|
|
|
wrapper.get_first_client_response()
|
|
|
|
|
|
parse_mock.assert_called_once_with(response)
|
|
|
|
|
|
def test_return_code(self):
|
|
|
wrapper = SubversionTunnelWrapper(timeout=5)
|
|
|
with patch.object(wrapper, 'process') as process_mock:
|
|
|
process_mock.returncode = 1
|
|
|
assert wrapper.return_code == 1
|
|
|
|
|
|
def test_sync_loop_breaks_when_process_cannot_be_polled(self):
|
|
|
self.counter = 0
|
|
|
buffer_ = 'abcdefghij'
|
|
|
|
|
|
wrapper = SubversionTunnelWrapper(timeout=5)
|
|
|
wrapper.stdin = BytesIO(buffer_)
|
|
|
with patch.object(wrapper, 'remove_configs') as remove_configs_mock:
|
|
|
with patch.object(wrapper, 'process') as process_mock:
|
|
|
process_mock.poll.side_effect = self._poll
|
|
|
process_mock.stdin = BytesIO()
|
|
|
wrapper.sync()
|
|
|
assert process_mock.stdin.getvalue() == 'abcde'
|
|
|
remove_configs_mock.assert_called_once_with()
|
|
|
|
|
|
def test_sync_loop_breaks_when_nothing_to_read(self):
|
|
|
self.counter = 0
|
|
|
buffer_ = 'abcdefghij'
|
|
|
|
|
|
wrapper = SubversionTunnelWrapper(timeout=5)
|
|
|
wrapper.stdin = BytesIO(buffer_)
|
|
|
with patch.object(wrapper, 'remove_configs') as remove_configs_mock:
|
|
|
with patch.object(wrapper, 'process') as process_mock:
|
|
|
process_mock.poll.return_value = None
|
|
|
process_mock.stdin = BytesIO()
|
|
|
wrapper.sync()
|
|
|
assert process_mock.stdin.getvalue() == buffer_
|
|
|
remove_configs_mock.assert_called_once_with()
|
|
|
|
|
|
def test_start_without_repositories_root(self):
|
|
|
svn_path = '/usr/local/bin/svnserve'
|
|
|
wrapper = SubversionTunnelWrapper(timeout=5, svn_path=svn_path)
|
|
|
with patch('rhodecode.apps.ssh_support.lib.ssh_wrapper.Popen') as popen_mock:
|
|
|
wrapper.start()
|
|
|
expected_command = [
|
|
|
svn_path, '-t', '--config-file', wrapper.svn_conf_path]
|
|
|
popen_mock.assert_called_once_with(
|
|
|
expected_command, stdin=subprocess.PIPE)
|
|
|
assert wrapper.process == popen_mock()
|
|
|
|
|
|
def test_start_with_repositories_root(self):
|
|
|
svn_path = '/usr/local/bin/svnserve'
|
|
|
repositories_root = '/home/repos'
|
|
|
wrapper = SubversionTunnelWrapper(
|
|
|
timeout=5, svn_path=svn_path, repositories_root=repositories_root)
|
|
|
with patch('rhodecode.apps.ssh_support.lib.ssh_wrapper.Popen') as popen_mock:
|
|
|
wrapper.start()
|
|
|
expected_command = [
|
|
|
svn_path, '-t', '--config-file', wrapper.svn_conf_path,
|
|
|
'-r', repositories_root]
|
|
|
popen_mock.assert_called_once_with(
|
|
|
expected_command, stdin=subprocess.PIPE)
|
|
|
assert wrapper.process == popen_mock()
|
|
|
|
|
|
def test_create_svn_config(self):
|
|
|
wrapper = SubversionTunnelWrapper(timeout=5)
|
|
|
file_mock = MagicMock(spec=file)
|
|
|
with patch('os.fdopen', create=True) as open_mock:
|
|
|
open_mock.return_value = file_mock
|
|
|
wrapper.create_svn_config()
|
|
|
open_mock.assert_called_once_with(wrapper.svn_conf_fd, 'w')
|
|
|
expected_content = '[general]\nhooks-env = {}\n'.format(
|
|
|
wrapper.hooks_env_path)
|
|
|
file_handle = file_mock.__enter__.return_value
|
|
|
file_handle.write.assert_called_once_with(expected_content)
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
|
'read_only, expected_content', [
|
|
|
[True, '[default]\nLANG = en_US.UTF-8\nSSH_READ_ONLY = 1\n'],
|
|
|
[False, '[default]\nLANG = en_US.UTF-8\n']
|
|
|
], ids=no_newline_id_generator)
|
|
|
def test_create_hooks_env(self, read_only, expected_content):
|
|
|
wrapper = SubversionTunnelWrapper(timeout=5)
|
|
|
wrapper.read_only = read_only
|
|
|
file_mock = MagicMock(spec=file)
|
|
|
with patch('os.fdopen', create=True) as open_mock:
|
|
|
open_mock.return_value = file_mock
|
|
|
wrapper.create_hooks_env()
|
|
|
open_mock.assert_called_once_with(wrapper.hooks_env_fd, 'w')
|
|
|
file_handle = file_mock.__enter__.return_value
|
|
|
file_handle.write.assert_called_once_with(expected_content)
|
|
|
|
|
|
def test_remove_configs(self):
|
|
|
wrapper = SubversionTunnelWrapper(timeout=5)
|
|
|
with patch('os.remove') as remove_mock:
|
|
|
wrapper.remove_configs()
|
|
|
expected_calls = [
|
|
|
call(wrapper.svn_conf_path), call(wrapper.hooks_env_path)]
|
|
|
assert sorted(remove_mock.call_args_list) == sorted(expected_calls)
|
|
|
|
|
|
def _poll(self):
|
|
|
self.counter += 1
|
|
|
return None if self.counter < 6 else 1
|
|
|
|
