##// END OF EJS Templates
release: Bump version 5.1.0 to 5.1.1
release: Bump version 5.1.0 to 5.1.1

File last commit:

r1249:5745b11f default
r1269:20bfc127 v5.1.1 stable
Show More
test_wsgi_app_caller.py
98 lines | 3.0 KiB | text/x-python | PythonLexer
/ vcsserver / tests / test_wsgi_app_caller.py
# RhodeCode VCSServer provides access to different vcs backends via network.
# Copyright (C) 2014-2023 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 wsgiref.simple_server
import wsgiref.validate
from vcsserver import wsgi_app_caller
from vcsserver.lib.str_utils import ascii_bytes, safe_str
@wsgiref.validate.validator
def demo_app(environ, start_response):
"""WSGI app used for testing."""
input_data = safe_str(environ['wsgi.input'].read(1024))
data = [
'Hello World!\n',
f'input_data={input_data}\n',
]
for key, value in sorted(environ.items()):
data.append(f'{key}={value}\n')
write = start_response("200 OK", [('Content-Type', 'text/plain')])
write(b'Old school write method\n')
write(b'***********************\n')
return list(map(ascii_bytes, data))
BASE_ENVIRON = {
'REQUEST_METHOD': 'GET',
'SERVER_NAME': 'localhost',
'SERVER_PORT': '80',
'SCRIPT_NAME': '',
'PATH_INFO': '/',
'QUERY_STRING': '',
'foo.var': 'bla',
}
def test_complete_environ():
environ = dict(BASE_ENVIRON)
data = b"data"
wsgi_app_caller._complete_environ(environ, data)
wsgiref.validate.check_environ(environ)
assert data == environ['wsgi.input'].read(1024)
def test_start_response():
start_response = wsgi_app_caller._StartResponse()
status = '200 OK'
headers = [('Content-Type', 'text/plain')]
start_response(status, headers)
assert status == start_response.status
assert headers == start_response.headers
def test_start_response_with_error():
start_response = wsgi_app_caller._StartResponse()
status = '500 Internal Server Error'
headers = [('Content-Type', 'text/plain')]
start_response(status, headers, (None, None, None))
assert status == start_response.status
assert headers == start_response.headers
def test_wsgi_app_caller():
environ = dict(BASE_ENVIRON)
input_data = 'some text'
caller = wsgi_app_caller.WSGIAppCaller(demo_app)
responses, status, headers = caller.handle(environ, input_data)
response = b''.join(responses)
assert status == '200 OK'
assert headers == [('Content-Type', 'text/plain')]
assert response.startswith(b'Old school write method\n***********************\n')
assert b'Hello World!\n' in response
assert b'foo.var=bla\n' in response
assert ascii_bytes(f'input_data={input_data}\n') in response