# RhodeCode VCSServer provides access to different vcs backends via network. # Copyright (C) 2014-2017 RodeCode 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 subprocess import StringIO import time import pytest from fixture import ContextINI @pytest.mark.parametrize("arguments, expected_texts", [ (['--threadpool=192'], [ 'threadpool_size: 192', 'worker pool of size 192 created', 'Threadpool size set to 192']), (['--locale=fake'], [ 'Cannot set locale, not configuring the locale system']), (['--timeout=5'], [ 'Timeout for RPC calls set to 5.0 seconds']), (['--log-level=info'], [ 'log_level: info']), (['--port={port}'], [ 'port: {port}', 'created daemon on localhost:{port}']), (['--host=127.0.0.1', '--port={port}'], [ 'port: {port}', 'host: 127.0.0.1', 'created daemon on 127.0.0.1:{port}']), (['--config=/bad/file'], ['OSError: File /bad/file does not exist']), ]) def test_vcsserver_calls(arguments, expected_texts, vcsserver_port): port_argument = '--port={port}' if port_argument not in arguments: arguments.append(port_argument) arguments = _replace_port(arguments, vcsserver_port) expected_texts = _replace_port(expected_texts, vcsserver_port) output = call_vcs_server_with_arguments(arguments) for text in expected_texts: assert text in output def _replace_port(values, port): return [value.format(port=port) for value in values] def test_vcsserver_with_config(vcsserver_port): ini_def = [ {'DEFAULT': {'host': '127.0.0.1'}}, {'DEFAULT': {'threadpool_size': '111'}}, {'DEFAULT': {'port': vcsserver_port}}, ] with ContextINI('test.ini', ini_def) as new_test_ini_path: output = call_vcs_server_with_arguments( ['--config=' + new_test_ini_path]) expected_texts = [ 'host: 127.0.0.1', 'Threadpool size set to 111', ] for text in expected_texts: assert text in output def test_vcsserver_with_config_cli_overwrite(vcsserver_port): ini_def = [ {'DEFAULT': {'host': '127.0.0.1'}}, {'DEFAULT': {'port': vcsserver_port}}, {'DEFAULT': {'threadpool_size': '111'}}, {'DEFAULT': {'timeout': '0'}}, ] with ContextINI('test.ini', ini_def) as new_test_ini_path: output = call_vcs_server_with_arguments([ '--config=' + new_test_ini_path, '--host=128.0.0.1', '--threadpool=256', '--timeout=5']) expected_texts = [ 'host: 128.0.0.1', 'Threadpool size set to 256', 'Timeout for RPC calls set to 5.0 seconds', ] for text in expected_texts: assert text in output def call_vcs_server_with_arguments(args): vcs = subprocess.Popen( ["vcsserver"] + args, stdout=subprocess.PIPE, stderr=subprocess.STDOUT) output = read_output_until( "Starting vcsserver.main", vcs.stdout) vcs.terminate() return output def call_vcs_server_with_non_existing_config_file(args): vcs = subprocess.Popen( ["vcsserver", "--config=/tmp/bad"] + args, stdout=subprocess.PIPE, stderr=subprocess.STDOUT) output = read_output_until( "Starting vcsserver.main", vcs.stdout) vcs.terminate() return output def read_output_until(expected, source, timeout=5): ts = time.time() buf = StringIO.StringIO() while time.time() - ts < timeout: line = source.readline() buf.write(line) if expected in line: break return buf.getvalue()