technocloud-public/nginx-unit
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

Tests: reworked python tests with application.

Browse Source
This commit is contained in:
Andrey Zelenkov
2018-02-20 20:34:41 +03:00
parent babf67712e
commit 09f2009df5
12 changed files with 104 additions and 191 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
test/python/204_no_content/wsgi.py Normal file
View File

@@ -0,0 +1,4 @@
def application(environ, start_response):
start_response('204 No Content', [])
return []

12
test/python/atexit/wsgi.py Normal file
View File

@@ -0,0 +1,12 @@
import atexit
def application(environ, start_response):
test_dir = environ.get('HTTP_TEST_DIR')
def create_file():
open(test_dir + '/atexit', 'w')
atexit.register(create_file)
start_response('200', [('Content-Length', '0')])
return []

3
test/python/empty/wsgi.py Normal file
View File

@@ -0,0 +1,3 @@
def application(env, start_response):
start_response('200', [('Content-Length', '0')])
return []

10
test/python/mirror/wsgi.py Normal file
View File

@@ -0,0 +1,10 @@
def application(environ, start_response):
content_length = int(environ.get('CONTENT_LENGTH', 0))
body = bytes(environ['wsgi.input'].read(content_length))
start_response('200', [
('Content-Type', environ.get('CONTENT_TYPE')),
('Content-Length', str(len(body)))
])
return [body]

7
test/python/query_string/wsgi.py Normal file
View File

@@ -0,0 +1,7 @@
def application(environ, start_response):
start_response('200', [
('Content-Length', '0'),
('Query-String', environ.get('QUERY_STRING'))
])
return []

7
test/python/server_port/wsgi.py Normal file
View File

@@ -0,0 +1,7 @@
def application(environ, start_response):
start_response('200', [
('Content-Length', '0'),
('Server-Port', environ.get('SERVER_PORT'))
])
return []

15
test/python/variables/wsgi.py Normal file
View File

@@ -0,0 +1,15 @@
def application(environ, start_response):
content_length = int(environ.get('CONTENT_LENGTH', 0))
body = bytes(environ['wsgi.input'].read(content_length))
start_response('200', [
('Content-Type', environ.get('CONTENT_TYPE')),
('Content-Length', str(len(body))),
('Request-Method', environ.get('REQUEST_METHOD')),
('Request-Uri', environ.get('REQUEST_URI')),
('Http-Host', environ.get('HTTP_HOST')),
('Server-Protocol', environ.get('SERVER_PROTOCOL')),
('Custom-Header', environ.get('HTTP_CUSTOM_HEADER'))
])
return [body]

86
test/test_python_application.py
View File

@@ -2,51 +2,13 @@ import time
import unittest
import unit
class TestUnitPythonApplication(unit.TestUnitControl):
class TestUnitPythonApplication(unit.TestUnitApplicationPython):
def setUpClass():
unit.TestUnit().check_modules('python')
def conf_with_name(self, name):
self.conf({
"listeners": {
"*:7080": {
"application": "app"
}
},
"applications": {
"app": {
"type": "python",
"processes": { "spare": 0 },
"path": self.testdir + '/' + name,
"module": "wsgi"
}
}
})
def test_python_application_simple(self):
code, name = """
def application(environ, start_response):
content_length = int(environ.get('CONTENT_LENGTH', 0))
body = bytes(environ['wsgi.input'].read(content_length))
start_response('200', [
('Content-Type', environ.get('CONTENT_TYPE')),
('Content-Length', str(len(body))),
('Request-Method', environ.get('REQUEST_METHOD')),
('Request-Uri', environ.get('REQUEST_URI')),
('Http-Host', environ.get('HTTP_HOST')),
('Server-Protocol', environ.get('SERVER_PROTOCOL')),
('Custom-Header', environ.get('HTTP_CUSTOM_HEADER'))
])
return [body]
""", 'py_app'
self.python_application(name, code)
self.conf_with_name(name)
def test_python_application_variables(self):
self.load('variables')
body = 'Test body string.'
@@ -75,20 +37,7 @@ def application(environ, start_response):
self.assertEqual(resp['body'], body, 'body')
def test_python_application_query_string(self):
code, name = """
def application(environ, start_response):
start_response('200', [
('Content-Length', '0'),
('Query-String', environ.get('QUERY_STRING'))
])
return []
""", 'py_app'
self.python_application(name, code)
self.conf_with_name(name)
self.load('query_string')
resp = self.get(url='/?var1=val1&var2=val2')
@@ -97,37 +46,14 @@ def application(environ, start_response):
@unittest.expectedFailure
def test_python_application_server_port(self):
code, name = """
def application(environ, start_response):
start_response('200', [
('Content-Length', '0'),
('Server-Port', environ.get('SERVER_PORT'))
])
return []
""", 'py_app'
self.python_application(name, code)
self.conf_with_name(name)
self.load('server_port')
self.assertEqual(self.get()['headers']['Server-Port'], '7080',
'Server-Port header')
@unittest.expectedFailure
def test_python_application_204_transfer_encoding(self):
code, name = """
def application(environ, start_response):
start_response('204 No Content', [])
return []
""", 'py_app'
self.python_application(name, code)
self.conf_with_name(name)
self.load('204_no_content')
self.assertNotIn('Transfer-Encoding', self.get()['headers'],
'204 header transfer encoding')

39
test/test_python_atexit.py
View File

@@ -3,46 +3,20 @@ import time
import unittest
import unit
class TestUnitPythonAtExit(unit.TestUnitControl):
class TestUnitPythonAtExit(unit.TestUnitApplicationPython):
def setUpClass():
unit.TestUnit().check_modules('python')
def test_python_atexit(self):
code, name = """
import atexit
self.load('atexit')
def create_file():
open('%s', 'w')
atexit.register(create_file)
def application(env, start_response):
start_response('200', [('Content-Length', '0')])
return []
""" % (self.testdir + '/atexit'), 'py_app'
self.python_application(name, code)
self.conf({
"listeners": {
"*:7080": {
"application": "app"
}
},
"applications": {
"app": {
"type": "python",
"processes": { "spare": 0 },
"path": self.testdir + '/' + name,
"module": "wsgi"
}
}
self.get(headers={
'Host': 'localhost',
'Test-Dir': self.testdir,
'Connection': 'close'
})
self.get()
self.conf({
"listeners": {},
"applications": {}
@@ -53,6 +27,5 @@ def application(env, start_response):
self.assertEqual(os.path.exists(self.testdir + '/atexit'), True,
'python atexit')
if __name__ == '__main__':
unittest.main()

35
test/test_python_keepalive.py
View File

@@ -1,44 +1,13 @@
import unittest
import unit
class TestUnitPythonKeepalive(unit.TestUnitControl):
class TestUnitPythonKeepalive(unit.TestUnitApplicationPython):
def setUpClass():
unit.TestUnit().check_modules('python')
def test_python_keepalive_body(self):
code, name = """
def application(environ, start_response):
content_length = int(environ.get('CONTENT_LENGTH', 0))
body = bytes(environ['wsgi.input'].read(content_length))
start_response('200', [
('Content-Type', environ.get('CONTENT_TYPE')),
('Content-Length', str(len(body)))
])
return [body]
""", 'py_app'
self.python_application(name, code)
self.conf({
"listeners": {
"*:7080": {
"application": "app"
}
},
"applications": {
"app": {
"type": "python",
"processes": { "spare": 0 },
"path": self.testdir + '/' + name,
"module": "wsgi"
}
}
})
self.load('mirror')
(resp, sock) = self.post(headers={
'Connection': 'keep-alive',

36
test/test_python_procman.py
View File

@@ -4,21 +4,18 @@ import subprocess
import unittest
import unit
class TestUnitPythonProcman(unit.TestUnitControl):
class TestUnitPythonProcman(unit.TestUnitApplicationPython):
def setUpClass():
unit.TestUnit().check_modules('python')
def pids_for_process(self, process=None):
if process is None:
process = self.app_name
def pids_for_process(self):
time.sleep(0.2)
output = subprocess.check_output(['ps', 'ax'])
pids = set()
for m in re.findall('.*' + process, output.decode()):
for m in re.findall('.*' + self.app_name, output.decode()):
pids.add(re.search('^\s*(\d+)', m).group(1))
return pids
@@ -26,33 +23,8 @@ class TestUnitPythonProcman(unit.TestUnitControl):
def setUp(self):
super().setUp()
code, name = """
def application(env, start_response):
start_response('200', [('Content-Length', '0')])
return []
""", 'py_app'
self.app_name = "app-" + self.testdir.split('/')[-1]
self.python_application(name, code)
self.conf({
"listeners": {
"*:7080": {
"application": self.app_name
}
},
"applications": {
self.app_name: {
"type": "python",
"processes": { "spare": 0 },
"path": self.testdir + '/' + name,
"module": "wsgi"
}
}
})
self.load('empty', self.app_name)
def test_python_processes_access(self):
self.conf('1', '/applications/' + self.app_name + '/processes')

41
test/unit.py
View File

@@ -77,12 +77,6 @@ class TestUnit(unittest.TestCase):
self.testdir + '/unit.log', self.testdir + '/control.unit.sock'):
exit("Could not start unit")
def python_application(self, name, code):
os.mkdir(self.testdir + '/' + name)
with open(self.testdir + '/' + name + '/wsgi.py', 'w') as f:
f.write(code)
def _stop(self):
with open(self.testdir + '/unit.pid', 'r') as f:
pid = f.read().rstrip()
@@ -283,20 +277,41 @@ class TestUnitApplicationProto(TestUnitControl):
current_dir = os.path.dirname(os.path.abspath(__file__))
class TestUnitApplicationPerl(TestUnitApplicationProto):
def load(self, dir, name='psgi.pl'):
class TestUnitApplicationPython(TestUnitApplicationProto):
def load(self, script, name=None):
if name is None:
name = script
self.conf({
"listeners": {
"*:7080": {
"application": dir
"application": name
}
},
"applications": {
dir: {
"type": "perl",
name: {
"type": "python",
"processes": { "spare": 0 },
"working_directory": self.current_dir + '/perl/' + dir,
"script": self.current_dir + '/perl/' + dir + '/' + name
"path": self.current_dir + '/python/' + script,
"module": "wsgi"
}
}
})
class TestUnitApplicationPerl(TestUnitApplicationProto):
def load(self, script, name='psgi.pl'):
self.conf({
"listeners": {
"*:7080": {
"application": script
}
},
"applications": {
script: {
"type": "perl",
"processes": { "spare": 0 },
"working_directory": self.current_dir + '/perl/' + script,
"script": self.current_dir + '/perl/' + script + '/' + name
}
}
})