##// END OF EJS Templates
release: Finish preparation for 4.1.2
release: Finish preparation for 4.1.2

File last commit:

r0:0fb8cb8f default
r27:f38ed1e1 v4.1.2 stable
Show More
test_vcsserver.py
132 lines | 4.3 KiB | text/x-python | PythonLexer
# RhodeCode VCSServer provides access to different vcs backends via network.
# Copyright (C) 2014-2016 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 TestINI
@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 TestINI('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 TestINI('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()