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

Tests: style.

Browse Source
This commit is contained in:
Andrei Zeliankou
2021-04-05 14:03:05 +01:00
parent 46d8567dd7
commit 6c97a1a069
67 changed files with 689 additions and 614 deletions
Download Patch File Download Diff File Expand all files Collapse all files

38
test/conftest.py
View File

@@ -6,7 +6,6 @@ import platform
import re
import shutil
import signal
import socket
import stat
import subprocess
import sys
@@ -15,11 +14,12 @@ import time
from multiprocessing import Process
import pytest
from unit.check.go import check_go
from unit.check.isolation import check_isolation
from unit.check.node import check_node
from unit.check.tls import check_openssl
from unit.check.regex import check_regex
from unit.check.tls import check_openssl
from unit.http import TestHTTP
from unit.option import option
from unit.utils import public_dir
@@ -85,6 +85,7 @@ _fds_check = {
}
http = TestHTTP()
def pytest_configure(config):
option.config = config.option
@@ -115,9 +116,11 @@ def pytest_configure(config):
def pytest_generate_tests(metafunc):
cls = metafunc.cls
if (not hasattr(cls, 'application_type')
or cls.application_type == None
or cls.application_type == 'external'):
if (
not hasattr(cls, 'application_type')
or cls.application_type == None
or cls.application_type == 'external'
):
return
type = cls.application_type
@@ -216,6 +219,7 @@ def pytest_sessionstart(session):
elif option.save_log:
open(unit_instance['temp_dir'] + '/' + unit_log_copy, 'w').close()
@pytest.hookimpl(tryfirst=True, hookwrapper=True)
def pytest_runtest_makereport(item, call):
# execute all other hooks to obtain the report object
@@ -320,7 +324,9 @@ def run(request):
public_dir(path)
if os.path.isfile(path) or stat.S_ISSOCK(os.stat(path).st_mode):
if os.path.isfile(path) or stat.S_ISSOCK(
os.stat(path).st_mode
):
os.remove(path)
else:
shutil.rmtree(path)
@@ -384,6 +390,7 @@ def run(request):
_check_alerts(log=log)
def unit_run():
global unit_instance
@@ -482,7 +489,6 @@ def unit_stop():
return 'Could not terminate unit'
def _check_alerts(path=None, log=None):
if path is None:
path = unit_instance['log']
@@ -554,24 +560,21 @@ def _clear_conf(sock, log=None):
return
try:
certs = json.loads(http.get(
url='/certificates',
sock_type='unix',
addr=sock,
)['body']).keys()
certs = json.loads(
http.get(url='/certificates', sock_type='unix', addr=sock,)['body']
).keys()
except json.JSONDecodeError:
pytest.fail('Can\'t parse certificates list.')
for cert in certs:
resp = http.delete(
url='/certificates/' + cert,
sock_type='unix',
addr=sock,
url='/certificates/' + cert, sock_type='unix', addr=sock,
)['body']
check_success(resp)
def _count_fds(pid):
procfile = '/proc/%s/fd' % pid
if os.path.isdir(procfile):
@@ -606,6 +609,7 @@ def run_process(target, *args):
_processes.append(process)
def stop_processes():
if not _processes:
return
@@ -657,18 +661,22 @@ def skip_fds_check():
def temp_dir(request):
return unit_instance['temp_dir']
@pytest.fixture
def is_unsafe(request):
return request.config.getoption("--unsafe")
@pytest.fixture
def is_su(request):
return os.geteuid() == 0
@pytest.fixture
def unit_pid(request):
return unit_instance['process'].pid
def pytest_sessionfinish(session):
if not option.restart and option.save_log:
print('Path to unit.log:\n' + unit_instance['log'] + '\n')

1
test/python/atexit/wsgi.py
View File

@@ -1,5 +1,6 @@
import atexit
def application(environ, start_response):
def at_exit():
environ['wsgi.errors'].write('At exit called.\n')

1
test/python/body_io/wsgi.py
View File

@@ -1,5 +1,6 @@
import io
def application(env, start_response):
start_response('200', [('Content-Length', '10')])
f = io.BytesIO(b'0123456789')

1
test/python/callable/wsgi.py
View File

@@ -2,6 +2,7 @@ def application(env, start_response):
start_response('204', [('Content-Length', '0')])
return []
def app(env, start_response):
start_response('200', [('Content-Length', '0')])
return []

16
test/python/ctx_iter_atexit/wsgi.py
View File

@@ -1,5 +1,6 @@
import atexit
class application:
def __init__(self, environ, start_response):
self.environ = environ
@@ -11,13 +12,14 @@ class application:
content_length = int(self.environ.get('CONTENT_LENGTH', 0))
body = bytes(self.environ['wsgi.input'].read(content_length))
self.start('200', [
('Content-Type', self.environ.get('CONTENT_TYPE')),
('Content-Length', str(len(body)))
])
self.start(
'200',
[
('Content-Type', self.environ.get('CONTENT_TYPE')),
('Content-Length', str(len(body))),
],
)
yield body
def _atexit(self):
self.start('200', [
('Content-Length', '0')
])
self.start('200', [('Content-Length', '0')])

11
test/python/custom_header/wsgi.py
View File

@@ -1,7 +1,10 @@
def application(environ, start_response):
start_response('200', [
('Content-Length', '0'),
('Custom-Header', environ.get('HTTP_CUSTOM_HEADER'))
])
start_response(
'200',
[
('Content-Length', '0'),
('Custom-Header', environ.get('HTTP_CUSTOM_HEADER')),
],
)
return []

27
test/python/delayed/asgi.py
View File

@@ -1,5 +1,6 @@
import asyncio
async def application(scope, receive, send):
assert scope['type'] == 'http'
@@ -28,13 +29,13 @@ async def application(scope, receive, send):
loop.call_later(n, future.set_result, None)
await future
await send({
'type': 'http.response.start',
'status': 200,
'headers': [
(b'content-length', str(len(body)).encode()),
]
})
await send(
{
'type': 'http.response.start',
'status': 200,
'headers': [(b'content-length', str(len(body)).encode()),],
}
)
if not body:
await sleep(delay)
@@ -42,10 +43,12 @@ async def application(scope, receive, send):
step = int(len(body) / parts)
for i in range(0, len(body), step):
await send({
'type': 'http.response.body',
'body': body[i : i + step],
'more_body': True,
})
await send(
{
'type': 'http.response.body',
'body': body[i : i + step],
'more_body': True,
}
)
await sleep(delay)

14
test/python/empty/asgi.py
View File

@@ -1,10 +1,10 @@
async def application(scope, receive, send):
assert scope['type'] == 'http'
await send({
'type': 'http.response.start',
'status': 200,
'headers': [
(b'content-length', b'0'),
]
})
await send(
{
'type': 'http.response.start',
'status': 200,
'headers': [(b'content-length', b'0')],
}
)

1
test/python/environment/wsgi.py
View File

@@ -1,5 +1,6 @@
import os
def application(env, start_response):
body = ''
vars = env.get('HTTP_X_VARIABLES').split(',')

7
test/python/header_fields/wsgi.py
View File

@@ -2,8 +2,7 @@ def application(environ, start_response):
h = (k for k, v in environ.items() if k.startswith('HTTP_'))
start_response('200', [
('Content-Length', '0'),
('All-Headers', ','.join(h))
])
start_response(
'200', [('Content-Length', '0'), ('All-Headers', ','.join(h))]
)
return []

13
test/python/host/wsgi.py
View File

@@ -1,7 +1,10 @@
def application(env, start_response):
start_response('200', [
('Content-Length', '0'),
('X-Server-Name', env.get('SERVER_NAME')),
('X-Http-Host', str(env.get('HTTP_HOST')))
])
start_response(
'200',
[
('Content-Length', '0'),
('X-Server-Name', env.get('SERVER_NAME')),
('X-Http-Host', str(env.get('HTTP_HOST'))),
],
)
return []

4
test/python/iter_exception/wsgi.py
View File

@@ -8,7 +8,9 @@ class application:
def __iter__(self):
self.__i = 0
self._skip_level = int(self.environ.get('HTTP_X_SKIP', 0))
self._not_skip_close = int(self.environ.get('HTTP_X_NOT_SKIP_CLOSE', 0))
self._not_skip_close = int(
self.environ.get('HTTP_X_NOT_SKIP_CLOSE', 0)
)
self._is_chunked = self.environ.get('HTTP_X_CHUNKED')
headers = [(('Content-Length', '10'))]

15
test/python/legacy/asgi.py
View File

@@ -3,11 +3,12 @@ def application(scope):
return app_http
async def app_http(receive, send):
await send({
'type': 'http.response.start',
'status': 200,
'headers': [
(b'content-length', b'0'),
]
})
await send(
{
'type': 'http.response.start',
'status': 200,
'headers': [(b'content-length', b'0'),],
}
)

15
test/python/legacy_force/asgi.py
View File

@@ -7,11 +7,12 @@ def application(scope, receive=None, send=None):
else:
return app_http(receive, send)
async def app_http(receive, send):
await send({
'type': 'http.response.start',
'status': 200,
'headers': [
(b'content-length', b'0'),
]
})
await send(
{
'type': 'http.response.start',
'status': 200,
'headers': [(b'content-length', b'0'),],
}
)

19
test/python/mirror/asgi.py
View File

@@ -8,15 +8,12 @@ async def application(scope, receive, send):
if not m.get('more_body', False):
break
await send({
'type': 'http.response.start',
'status': 200,
'headers': [
(b'content-length', str(len(body)).encode()),
]
})
await send(
{
'type': 'http.response.start',
'status': 200,
'headers': [(b'content-length', str(len(body)).encode())],
}
)
await send({
'type': 'http.response.body',
'body': body,
})
await send({'type': 'http.response.body', 'body': body})

4
test/python/mirror/wsgi.py
View File

@@ -3,7 +3,5 @@ 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-Length', str(len(body)))
])
start_response('200', [('Content-Length', str(len(body)))])
return [body]

1
test/python/path/wsgi.py
View File

@@ -1,6 +1,7 @@
import os
import sys
def application(environ, start_response):
body = os.pathsep.join(sys.path).encode()

18
test/python/query_string/asgi.py
View File

@@ -1,11 +1,13 @@
async def application(scope, receive, send):
assert scope['type'] == 'http'
await send({
'type': 'http.response.start',
'status': 200,
'headers': [
(b'content-length', b'0'),
(b'query-string', scope['query_string']),
]
})
await send(
{
'type': 'http.response.start',
'status': 200,
'headers': [
(b'content-length', b'0'),
(b'query-string', scope['query_string']),
],
}
)

11
test/python/query_string/wsgi.py
View File

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

18
test/python/server_port/asgi.py
View File

@@ -1,11 +1,13 @@
async def application(scope, receive, send):
assert scope['type'] == 'http'
await send({
'type': 'http.response.start',
'status': 200,
'headers': [
(b'content-length', b'0'),
(b'server-port', str(scope['server'][1]).encode()),
]
})
await send(
{
'type': 'http.response.start',
'status': 200,
'headers': [
(b'content-length', b'0'),
(b'server-port', str(scope['server'][1]).encode()),
],
}
)

8
test/python/server_port/wsgi.py
View File

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

14
test/python/threading/asgi.py
View File

@@ -33,10 +33,10 @@ async def application(scope, receive, send):
Foo(Foo.num).start()
Foo.num += 10
await send({
'type': 'http.response.start',
'status': 200,
'headers': [
(b'content-length', b'0'),
]
})
await send(
{
'type': 'http.response.start',
'status': 200,
'headers': [(b'content-length', b'0')],
}
)

19
test/python/threads/asgi.py
View File

@@ -2,6 +2,7 @@ import asyncio
import time
import threading
async def application(scope, receive, send):
assert scope['type'] == 'http'
@@ -17,11 +18,13 @@ async def application(scope, receive, send):
time.sleep(delay)
await send({
'type': 'http.response.start',
'status': 200,
'headers': [
(b'content-length', b'0'),
(b'x-thread', str(threading.currentThread().ident).encode()),
]
})
await send(
{
'type': 'http.response.start',
'status': 200,
'headers': [
(b'content-length', b'0'),
(b'x-thread', str(threading.currentThread().ident).encode()),
],
}
)

14
test/python/threads/wsgi.py
View File

@@ -1,15 +1,19 @@
import time
import threading
def application(environ, start_response):
delay = float(environ.get('HTTP_X_DELAY', 0))
time.sleep(delay)
start_response('200', [
('Content-Length', '0'),
('Wsgi-Multithread', str(environ['wsgi.multithread'])),
('X-Thread', str(threading.currentThread().ident))
])
start_response(
'200',
[
('Content-Length', '0'),
('Wsgi-Multithread', str(environ['wsgi.multithread'])),
('X-Thread', str(threading.currentThread().ident)),
],
)
return []

10
test/python/upload/wsgi.py
View File

@@ -1,6 +1,7 @@
from tempfile import TemporaryFile
import os, cgi
def read(environ):
length = int(environ.get('CONTENT_LENGTH', 0))
@@ -11,6 +12,7 @@ def read(environ):
environ['wsgi.input'] = body
return body
def application(environ, start_response):
file = read(environ)
@@ -19,9 +21,9 @@ def application(environ, start_response):
filename = form['file'].filename
data = filename.encode() + form['file'].file.read()
start_response('200 OK', [
('Content-Type', 'text/plain'),
('Content-Length', str(len(data))),
])
start_response(
'200 OK',
[('Content-Type', 'text/plain'), ('Content-Length', str(len(data)))],
)
return data

17
test/python/user_group/wsgi.py
View File

@@ -1,18 +1,19 @@
import json
import os
def application(environ, start_response):
uid = os.geteuid()
gid = os.getegid()
out = json.dumps({
'UID': uid,
'GID': gid,
}).encode('utf-8')
out = json.dumps({'UID': uid, 'GID': gid,}).encode('utf-8')
start_response('200 OK', [
('Content-Length', str(len(out))),
('Content-Type', 'application/json')
])
start_response(
'200 OK',
[
('Content-Length', str(len(out))),
('Content-Type', 'application/json'),
],
)
return [out]

39
test/python/variables/asgi.py
View File

@@ -17,24 +17,23 @@ async def application(scope, receive, send):
res.append(h[1])
return b', '.join(res)
await send({
'type': 'http.response.start',
'status': 200,
'headers': [
(b'content-type', get_header(b'content-type')),
(b'content-length', str(len(body)).encode()),
(b'request-method', scope['method'].encode()),
(b'request-uri', scope['path'].encode()),
(b'http-host', get_header(b'host')),
(b'http-version', scope['http_version'].encode()),
(b'asgi-version', scope['asgi']['version'].encode()),
(b'asgi-spec-version', scope['asgi']['spec_version'].encode()),
(b'scheme', scope['scheme'].encode()),
(b'custom-header', get_header(b'custom-header')),
]
})
await send(
{
'type': 'http.response.start',
'status': 200,
'headers': [
(b'content-type', get_header(b'content-type')),
(b'content-length', str(len(body)).encode()),
(b'request-method', scope['method'].encode()),
(b'request-uri', scope['path'].encode()),
(b'http-host', get_header(b'host')),
(b'http-version', scope['http_version'].encode()),
(b'asgi-version', scope['asgi']['version'].encode()),
(b'asgi-spec-version', scope['asgi']['spec_version'].encode()),
(b'scheme', scope['scheme'].encode()),
(b'custom-header', get_header(b'custom-header')),
],
}
)
await send({
'type': 'http.response.body',
'body': body,
})
await send({'type': 'http.response.body', 'body': body})

33
test/python/variables/wsgi.py
View File

@@ -3,19 +3,22 @@ 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')),
('Server-Software', environ.get('SERVER_SOFTWARE')),
('Custom-Header', environ.get('HTTP_CUSTOM_HEADER')),
('Wsgi-Version', str(environ['wsgi.version'])),
('Wsgi-Url-Scheme', environ['wsgi.url_scheme']),
('Wsgi-Multithread', str(environ['wsgi.multithread'])),
('Wsgi-Multiprocess', str(environ['wsgi.multiprocess'])),
('Wsgi-Run-Once', str(environ['wsgi.run_once']))
])
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')),
('Server-Software', environ.get('SERVER_SOFTWARE')),
('Custom-Header', environ.get('HTTP_CUSTOM_HEADER')),
('Wsgi-Version', str(environ['wsgi.version'])),
('Wsgi-Url-Scheme', environ['wsgi.url_scheme']),
('Wsgi-Multithread', str(environ['wsgi.multithread'])),
('Wsgi-Multiprocess', str(environ['wsgi.multiprocess'])),
('Wsgi-Run-Once', str(environ['wsgi.run_once'])),
],
)
return [body]

18
test/python/websockets/mirror/asgi.py
View File

@@ -3,16 +3,16 @@ async def application(scope, receive, send):
while True:
m = await receive()
if m['type'] == 'websocket.connect':
await send({
'type': 'websocket.accept',
})
await send({'type': 'websocket.accept'})
if m['type'] == 'websocket.receive':
await send({
'type': 'websocket.send',
'bytes': m.get('bytes', None),
'text': m.get('text', None),
})
await send(
{
'type': 'websocket.send',
'bytes': m.get('bytes', None),
'text': m.get('text', None),
}
)
if m['type'] == 'websocket.disconnect':
break;
break

30
test/python/websockets/subprotocol/asgi.py
View File

@@ -6,20 +6,24 @@ async def application(scope, receive, send):
if m['type'] == 'websocket.connect':
subprotocols = scope['subprotocols']
await send({
'type': 'websocket.accept',
'headers': [
(b'x-subprotocols', str(subprotocols).encode()),
],
'subprotocol': subprotocols[0],
})
await send(
{
'type': 'websocket.accept',
'headers': [
(b'x-subprotocols', str(subprotocols).encode()),
],
'subprotocol': subprotocols[0],
}
)
if m['type'] == 'websocket.receive':
await send({
'type': 'websocket.send',
'bytes': m.get('bytes', None),
'text': m.get('text', None),
})
await send(
{
'type': 'websocket.send',
'bytes': m.get('bytes', None),
'text': m.get('text', None),
}
)
if m['type'] == 'websocket.disconnect':
break;
break

1
test/test_access_log.py
View File

@@ -1,6 +1,7 @@
import time
import pytest
from unit.applications.lang.python import TestApplicationPython
from unit.option import option

13
test/test_asgi_application.py
View File

@@ -3,13 +3,15 @@ import time
from distutils.version import LooseVersion
import pytest
from unit.applications.lang.python import TestApplicationPython
from unit.option import option
class TestASGIApplication(TestApplicationPython):
prerequisites = {'modules': {'python':
lambda v: LooseVersion(v) >= LooseVersion('3.5')}}
prerequisites = {
'modules': {'python': lambda v: LooseVersion(v) >= LooseVersion('3.5')}
}
load_module = 'asgi'
def findall(self, pattern):
@@ -31,7 +33,8 @@ Content-Type: text/html
Connection: close
custom-header: BLAH
%s""" % (len(body), body.encode()),
%s"""
% (len(body), body.encode()),
raw=True,
)
@@ -145,7 +148,7 @@ custom-header: BLAH
assert 'success' in self.conf(
'{"http":{"max_body_size": ' + str(max_body_size) + ' }}',
'settings'
'settings',
)
assert self.get()['status'] == 200, 'init'
@@ -398,7 +401,7 @@ Connection: close
socks.append(sock)
time.sleep(1.0) # required to avoid greedy request reading
time.sleep(1.0) # required to avoid greedy request reading
threads = set()

1
test/test_asgi_lifespan.py
View File

@@ -2,6 +2,7 @@ import os
from distutils.version import LooseVersion
import pytest
from conftest import unit_stop
from unit.applications.lang.python import TestApplicationPython
from unit.option import option

202
test/test_asgi_websockets.py
View File

@@ -3,14 +3,16 @@ import time
from distutils.version import LooseVersion
import pytest
from unit.applications.lang.python import TestApplicationPython
from unit.applications.websockets import TestApplicationWebsocket
from unit.option import option
class TestASGIWebsockets(TestApplicationPython):
prerequisites = {'modules': {'python':
lambda v: LooseVersion(v) >= LooseVersion('3.5')}}
prerequisites = {
'modules': {'python': lambda v: LooseVersion(v) >= LooseVersion('3.5')}
}
load_module = 'asgi'
ws = TestApplicationWebsocket()
@@ -74,7 +76,9 @@ class TestASGIWebsockets(TestApplicationPython):
sock.close()
assert resp['status'] == 101, 'status'
assert resp['headers']['x-subprotocols'] == "('chat', 'phone', 'video')", 'subprotocols'
assert (
resp['headers']['x-subprotocols'] == "('chat', 'phone', 'video')"
), 'subprotocols'
assert resp['headers']['sec-websocket-protocol'] == 'chat', 'key'
def test_asgi_websockets_mirror(self):
@@ -159,7 +163,7 @@ class TestASGIWebsockets(TestApplicationPython):
self.ws.frame_write(sock, self.ws.OP_TEXT, 'fragment1', fin=False)
self.ws.frame_write(
sock, self.ws.OP_CONT, 'fragment2', length=2**64 - 1
sock, self.ws.OP_CONT, 'fragment2', length=2 ** 64 - 1
)
self.check_close(sock, 1009) # 1009 - CLOSE_TOO_LARGE
@@ -231,7 +235,7 @@ class TestASGIWebsockets(TestApplicationPython):
@pytest.mark.skip('not yet')
def test_asgi_websockets_handshake_upgrade_absent(
self
self,
): # FAIL https://tools.ietf.org/html/rfc6455#section-4.2.1
self.load('websockets/mirror')
@@ -324,7 +328,9 @@ class TestASGIWebsockets(TestApplicationPython):
},
)
assert resp['status'] == 400, 'key double' # FAIL https://tools.ietf.org/html/rfc6455#section-11.3.1
assert (
resp['status'] == 400
), 'key double' # FAIL https://tools.ietf.org/html/rfc6455#section-11.3.1
def test_asgi_websockets_handshake_method_invalid(self):
self.load('websockets/mirror')
@@ -419,14 +425,14 @@ class TestASGIWebsockets(TestApplicationPython):
frame = self.ws.frame_read(sock)
self.check_frame(frame, True, opcode, payload)
check_length(0) # 1_1_1
check_length(125) # 1_1_2
check_length(126) # 1_1_3
check_length(127) # 1_1_4
check_length(128) # 1_1_5
check_length(65535) # 1_1_6
check_length(65536) # 1_1_7
check_length(65536, chopsize = 997) # 1_1_8
check_length(0) # 1_1_1
check_length(125) # 1_1_2
check_length(126) # 1_1_3
check_length(127) # 1_1_4
check_length(128) # 1_1_5
check_length(65535) # 1_1_6
check_length(65536) # 1_1_7
check_length(65536, chopsize=997) # 1_1_8
self.close_connection(sock)
@@ -445,14 +451,14 @@ class TestASGIWebsockets(TestApplicationPython):
self.check_frame(frame, True, opcode, payload)
check_length(0) # 1_2_1
check_length(125) # 1_2_2
check_length(126) # 1_2_3
check_length(127) # 1_2_4
check_length(128) # 1_2_5
check_length(65535) # 1_2_6
check_length(65536) # 1_2_7
check_length(65536, chopsize = 997) # 1_2_8
check_length(0) # 1_2_1
check_length(125) # 1_2_2
check_length(126) # 1_2_3
check_length(127) # 1_2_4
check_length(128) # 1_2_5
check_length(65535) # 1_2_6
check_length(65536) # 1_2_7
check_length(65536, chopsize=997) # 1_2_8
self.close_connection(sock)
@@ -470,11 +476,11 @@ class TestASGIWebsockets(TestApplicationPython):
self.check_frame(frame, True, op_pong, payload, decode=decode)
check_ping('') # 2_1
check_ping('Hello, world!') # 2_2
check_ping('') # 2_1
check_ping('Hello, world!') # 2_2
check_ping(b'\x00\xff\xfe\xfd\xfc\xfb\x00\xff', decode=False) # 2_3
check_ping(b'\xfe' * 125, decode=False) # 2_4
check_ping(b'\xfe' * 125, chopsize=1, decode=False) # 2_6
check_ping(b'\xfe' * 125, decode=False) # 2_4
check_ping(b'\xfe' * 125, chopsize=1, decode=False) # 2_6
self.close_connection(sock)
@@ -935,7 +941,9 @@ class TestASGIWebsockets(TestApplicationPython):
frame = self.ws.frame_read(sock)
if frame['opcode'] == self.ws.OP_TEXT:
self.check_frame(frame, True, self.ws.OP_TEXT, 'fragment1fragment2')
self.check_frame(
frame, True, self.ws.OP_TEXT, 'fragment1fragment2'
)
frame = None
self.check_close(sock, 1002, frame=frame)
@@ -1092,27 +1100,27 @@ class TestASGIWebsockets(TestApplicationPython):
self.close_connection(sock)
# Unit does not support UTF-8 validation
#
# # 6_3_1 FAIL
#
# payload_1 = '\xce\xba\xe1\xbd\xb9\xcf\x83\xce\xbc\xce\xb5'
# payload_2 = '\xed\xa0\x80'
# payload_3 = '\x65\x64\x69\x74\x65\x64'
#
# payload = payload_1 + payload_2 + payload_3
#
# self.ws.message(sock, self.ws.OP_TEXT, payload)
# self.check_close(sock, 1007)
#
# # 6_3_2 FAIL
#
# _, sock, _ = self.ws.upgrade()
#
# self.ws.message(sock, self.ws.OP_TEXT, payload, fragmention_size=1)
# self.check_close(sock, 1007)
#
# # 6_4_1 ... 6_4_4 FAIL
# Unit does not support UTF-8 validation
#
# # 6_3_1 FAIL
#
# payload_1 = '\xce\xba\xe1\xbd\xb9\xcf\x83\xce\xbc\xce\xb5'
# payload_2 = '\xed\xa0\x80'
# payload_3 = '\x65\x64\x69\x74\x65\x64'
#
# payload = payload_1 + payload_2 + payload_3
#
# self.ws.message(sock, self.ws.OP_TEXT, payload)
# self.check_close(sock, 1007)
#
# # 6_3_2 FAIL
#
# _, sock, _ = self.ws.upgrade()
#
# self.ws.message(sock, self.ws.OP_TEXT, payload, fragmention_size=1)
# self.check_close(sock, 1007)
#
# # 6_4_1 ... 6_4_4 FAIL
def test_asgi_websockets_7_1_1__7_5_1(self):
self.load('websockets/mirror')
@@ -1239,15 +1247,15 @@ class TestASGIWebsockets(TestApplicationPython):
self.ws.frame_write(sock, self.ws.OP_CLOSE, payload)
self.check_close(sock, 1002)
# # 7_5_1 FAIL Unit does not support UTF-8 validation
#
# _, sock, _ = self.ws.upgrade()
#
# payload = self.ws.serialize_close(reason = '\xce\xba\xe1\xbd\xb9\xcf' \
# '\x83\xce\xbc\xce\xb5\xed\xa0\x80\x65\x64\x69\x74\x65\x64')
#
# self.ws.frame_write(sock, self.ws.OP_CLOSE, payload)
# self.check_close(sock, 1007)
# # 7_5_1 FAIL Unit does not support UTF-8 validation
#
# _, sock, _ = self.ws.upgrade()
#
# payload = self.ws.serialize_close(reason = '\xce\xba\xe1\xbd\xb9\xcf' \
# '\x83\xce\xbc\xce\xb5\xed\xa0\x80\x65\x64\x69\x74\x65\x64')
#
# self.ws.frame_write(sock, self.ws.OP_CLOSE, payload)
# self.check_close(sock, 1007)
def test_asgi_websockets_7_7_X__7_9_X(self):
self.load('websockets/mirror')
@@ -1350,52 +1358,52 @@ class TestASGIWebsockets(TestApplicationPython):
frame = self.ws.frame_read(sock, read_timeout=5)
self.check_frame(frame, True, opcode, payload)
check_payload(op_text, 64 * 2 ** 10) # 9_1_1
check_payload(op_text, 256 * 2 ** 10) # 9_1_2
check_payload(op_text, 2 ** 20) # 9_1_3
check_payload(op_text, 4 * 2 ** 20) # 9_1_4
check_payload(op_text, 8 * 2 ** 20) # 9_1_5
check_payload(op_text, 16 * 2 ** 20) # 9_1_6
check_payload(op_text, 64 * 2 ** 10) # 9_1_1
check_payload(op_text, 256 * 2 ** 10) # 9_1_2
check_payload(op_text, 2 ** 20) # 9_1_3
check_payload(op_text, 4 * 2 ** 20) # 9_1_4
check_payload(op_text, 8 * 2 ** 20) # 9_1_5
check_payload(op_text, 16 * 2 ** 20) # 9_1_6
check_payload(op_binary, 64 * 2 ** 10) # 9_2_1
check_payload(op_binary, 256 * 2 ** 10) # 9_2_2
check_payload(op_binary, 2 ** 20) # 9_2_3
check_payload(op_binary, 4 * 2 ** 20) # 9_2_4
check_payload(op_binary, 8 * 2 ** 20) # 9_2_5
check_payload(op_binary, 16 * 2 ** 20) # 9_2_6
check_payload(op_binary, 64 * 2 ** 10) # 9_2_1
check_payload(op_binary, 256 * 2 ** 10) # 9_2_2
check_payload(op_binary, 2 ** 20) # 9_2_3
check_payload(op_binary, 4 * 2 ** 20) # 9_2_4
check_payload(op_binary, 8 * 2 ** 20) # 9_2_5
check_payload(op_binary, 16 * 2 ** 20) # 9_2_6
if option.system != 'Darwin' and option.system != 'FreeBSD':
check_message(op_text, 64) # 9_3_1
check_message(op_text, 256) # 9_3_2
check_message(op_text, 2 ** 10) # 9_3_3
check_message(op_text, 4 * 2 ** 10) # 9_3_4
check_message(op_text, 16 * 2 ** 10) # 9_3_5
check_message(op_text, 64 * 2 ** 10) # 9_3_6
check_message(op_text, 256 * 2 ** 10) # 9_3_7
check_message(op_text, 2 ** 20) # 9_3_8
check_message(op_text, 4 * 2 ** 20) # 9_3_9
check_message(op_text, 64) # 9_3_1
check_message(op_text, 256) # 9_3_2
check_message(op_text, 2 ** 10) # 9_3_3
check_message(op_text, 4 * 2 ** 10) # 9_3_4
check_message(op_text, 16 * 2 ** 10) # 9_3_5
check_message(op_text, 64 * 2 ** 10) # 9_3_6
check_message(op_text, 256 * 2 ** 10) # 9_3_7
check_message(op_text, 2 ** 20) # 9_3_8
check_message(op_text, 4 * 2 ** 20) # 9_3_9
check_message(op_binary, 64) # 9_4_1
check_message(op_binary, 256) # 9_4_2
check_message(op_binary, 2 ** 10) # 9_4_3
check_message(op_binary, 4 * 2 ** 10) # 9_4_4
check_message(op_binary, 16 * 2 ** 10) # 9_4_5
check_message(op_binary, 64 * 2 ** 10) # 9_4_6
check_message(op_binary, 256 * 2 ** 10) # 9_4_7
check_message(op_binary, 2 ** 20) # 9_4_8
check_message(op_binary, 4 * 2 ** 20) # 9_4_9
check_message(op_binary, 64) # 9_4_1
check_message(op_binary, 256) # 9_4_2
check_message(op_binary, 2 ** 10) # 9_4_3
check_message(op_binary, 4 * 2 ** 10) # 9_4_4
check_message(op_binary, 16 * 2 ** 10) # 9_4_5
check_message(op_binary, 64 * 2 ** 10) # 9_4_6
check_message(op_binary, 256 * 2 ** 10) # 9_4_7
check_message(op_binary, 2 ** 20) # 9_4_8
check_message(op_binary, 4 * 2 ** 20) # 9_4_9
check_payload(op_text, 2 ** 20, chopsize=64) # 9_5_1
check_payload(op_text, 2 ** 20, chopsize=128) # 9_5_2
check_payload(op_text, 2 ** 20, chopsize=256) # 9_5_3
check_payload(op_text, 2 ** 20, chopsize=512) # 9_5_4
check_payload(op_text, 2 ** 20, chopsize=1024) # 9_5_5
check_payload(op_text, 2 ** 20, chopsize=2048) # 9_5_6
check_payload(op_text, 2 ** 20, chopsize=64) # 9_5_1
check_payload(op_text, 2 ** 20, chopsize=128) # 9_5_2
check_payload(op_text, 2 ** 20, chopsize=256) # 9_5_3
check_payload(op_text, 2 ** 20, chopsize=512) # 9_5_4
check_payload(op_text, 2 ** 20, chopsize=1024) # 9_5_5
check_payload(op_text, 2 ** 20, chopsize=2048) # 9_5_6
check_payload(op_binary, 2 ** 20, chopsize=64) # 9_6_1
check_payload(op_binary, 2 ** 20, chopsize=128) # 9_6_2
check_payload(op_binary, 2 ** 20, chopsize=256) # 9_6_3
check_payload(op_binary, 2 ** 20, chopsize=512) # 9_6_4
check_payload(op_binary, 2 ** 20, chopsize=64) # 9_6_1
check_payload(op_binary, 2 ** 20, chopsize=128) # 9_6_2
check_payload(op_binary, 2 ** 20, chopsize=256) # 9_6_3
check_payload(op_binary, 2 ** 20, chopsize=512) # 9_6_4
check_payload(op_binary, 2 ** 20, chopsize=1024) # 9_6_5
check_payload(op_binary, 2 ** 20, chopsize=2048) # 9_6_6

1
test/test_configuration.py
View File

@@ -1,4 +1,5 @@
import pytest
from unit.control import TestControl

8
test/test_go_application.py
View File

@@ -129,11 +129,9 @@ class TestGoApplication(TestApplicationGo):
def test_go_application_command_line_arguments_type(self):
self.load('command_line_arguments')
assert 'error' in \
self.conf(
'' "a b c", 'applications/command_line_arguments/arguments'
), \
'arguments type'
assert 'error' in self.conf(
'' "a b c", 'applications/command_line_arguments/arguments'
), 'arguments type'
def test_go_application_command_line_arguments_0(self):
self.load('command_line_arguments')

14
test/test_go_isolation.py
View File

@@ -3,10 +3,12 @@ import os
import pwd
import pytest
from unit.applications.lang.go import TestApplicationGo
from unit.option import option
from unit.utils import getns
class TestGoIsolation(TestApplicationGo):
prerequisites = {'modules': {'go': 'any'}, 'features': ['isolation']}
@@ -279,7 +281,7 @@ class TestGoIsolation(TestApplicationGo):
isolation['namespaces'] = {
'mount': True,
'credential': True,
'pid': True
'pid': True,
}
self.load('ns_inspect', isolation=isolation)
@@ -337,12 +339,10 @@ class TestGoIsolation(TestApplicationGo):
isolation['namespaces'] = {
'mount': True,
'credential': True,
'pid': True
'pid': True,
}
isolation['automount'] = {
'tmpfs': False
}
isolation['automount'] = {'tmpfs': False}
self.load('ns_inspect', isolation=isolation)
@@ -352,9 +352,7 @@ class TestGoIsolation(TestApplicationGo):
"/ /tmp" not in obj['Mounts'] and "tmpfs" not in obj['Mounts']
), 'app has no /tmp mounted'
isolation['automount'] = {
'tmpfs': True
}
isolation['automount'] = {'tmpfs': True}
self.load('ns_inspect', isolation=isolation)

1
test/test_go_isolation_rootfs.py
View File

@@ -1,6 +1,7 @@
import os
import pytest
from unit.applications.lang.go import TestApplicationGo

1
test/test_http_header.py
View File

@@ -1,4 +1,5 @@
import pytest
from unit.applications.lang.python import TestApplicationPython

11
test/test_java_application.py
View File

@@ -7,6 +7,7 @@ from unit.applications.lang.java import TestApplicationJava
from unit.option import option
from unit.utils import public_dir
class TestJavaApplication(TestApplicationJava):
prerequisites = {'modules': {'java': 'all'}}
@@ -712,9 +713,9 @@ class TestJavaApplication(TestApplicationJava):
assert (
'javax.servlet.include.request_uri: /data/test' in body
) == True, 'include request uri'
#assert (
# assert (
# 'javax.servlet.include.context_path: ' in body
#) == True, 'include request context path'
# ) == True, 'include request context path'
assert (
'javax.servlet.include.servlet_path: /data' in body
) == True, 'include request servlet path'
@@ -754,9 +755,9 @@ class TestJavaApplication(TestApplicationJava):
assert (
'javax.servlet.include.request_uri: null' in body
) == True, 'include request uri'
#assert (
# assert (
# 'javax.servlet.include.context_path: null' in body
#) == True, 'include request context path'
# ) == True, 'include request context path'
assert (
'javax.servlet.include.servlet_path: null' in body
) == True, 'include request servlet path'
@@ -1034,7 +1035,7 @@ class TestJavaApplication(TestApplicationJava):
socks.append(sock)
time.sleep(0.25) # required to avoid greedy request reading
time.sleep(0.25) # required to avoid greedy request reading
threads = set()

1
test/test_java_isolation_rootfs.py
View File

@@ -2,6 +2,7 @@ import os
import subprocess
import pytest
from unit.applications.lang.java import TestApplicationJava
from unit.option import option

191
test/test_java_websockets.py
View File

@@ -2,6 +2,7 @@ import struct
import time
import pytest
from unit.applications.lang.java import TestApplicationJava
from unit.applications.websockets import TestApplicationWebsocket
from unit.option import option
@@ -163,7 +164,7 @@ class TestJavaWebsockets(TestApplicationJava):
@pytest.mark.skip('not yet')
def test_java_websockets_handshake_upgrade_absent(
self
self,
): # FAIL https://tools.ietf.org/html/rfc6455#section-4.2.1
self.load('websockets_mirror')
@@ -256,7 +257,9 @@ class TestJavaWebsockets(TestApplicationJava):
},
)
assert resp['status'] == 400, 'key double' # FAIL https://tools.ietf.org/html/rfc6455#section-11.3.1
assert (
resp['status'] == 400
), 'key double' # FAIL https://tools.ietf.org/html/rfc6455#section-11.3.1
def test_java_websockets_handshake_method_invalid(self):
self.load('websockets_mirror')
@@ -351,14 +354,14 @@ class TestJavaWebsockets(TestApplicationJava):
frame = self.ws.message_read(sock)
self.check_frame(frame, True, opcode, payload)
check_length(0) # 1_1_1
check_length(125) # 1_1_2
check_length(126) # 1_1_3
check_length(127) # 1_1_4
check_length(128) # 1_1_5
check_length(65535) # 1_1_6
check_length(65536) # 1_1_7
check_length(65536, chopsize = 997) # 1_1_8
check_length(0) # 1_1_1
check_length(125) # 1_1_2
check_length(126) # 1_1_3
check_length(127) # 1_1_4
check_length(128) # 1_1_5
check_length(65535) # 1_1_6
check_length(65536) # 1_1_7
check_length(65536, chopsize=997) # 1_1_8
self.close_connection(sock)
@@ -377,14 +380,14 @@ class TestJavaWebsockets(TestApplicationJava):
frame = self.ws.message_read(sock)
self.check_frame(frame, True, opcode, payload)
check_length(0) # 1_2_1
check_length(125) # 1_2_2
check_length(126) # 1_2_3
check_length(127) # 1_2_4
check_length(128) # 1_2_5
check_length(65535) # 1_2_6
check_length(65536) # 1_2_7
check_length(65536, chopsize = 997) # 1_2_8
check_length(0) # 1_2_1
check_length(125) # 1_2_2
check_length(126) # 1_2_3
check_length(127) # 1_2_4
check_length(128) # 1_2_5
check_length(65535) # 1_2_6
check_length(65536) # 1_2_7
check_length(65536, chopsize=997) # 1_2_8
self.close_connection(sock)
@@ -402,11 +405,11 @@ class TestJavaWebsockets(TestApplicationJava):
self.check_frame(frame, True, op_pong, payload, decode=decode)
check_ping('') # 2_1
check_ping('Hello, world!') # 2_2
check_ping('') # 2_1
check_ping('Hello, world!') # 2_2
check_ping(b'\x00\xff\xfe\xfd\xfc\xfb\x00\xff', decode=False) # 2_3
check_ping(b'\xfe' * 125, decode=False) # 2_4
check_ping(b'\xfe' * 125, chopsize=1, decode=False) # 2_6
check_ping(b'\xfe' * 125, decode=False) # 2_4
check_ping(b'\xfe' * 125, chopsize=1, decode=False) # 2_6
self.close_connection(sock)
@@ -867,7 +870,9 @@ class TestJavaWebsockets(TestApplicationJava):
frame = self.ws.frame_read(sock)
if frame['opcode'] == self.ws.OP_TEXT:
self.check_frame(frame, True, self.ws.OP_TEXT, 'fragment1fragment2')
self.check_frame(
frame, True, self.ws.OP_TEXT, 'fragment1fragment2'
)
frame = None
self.check_close(sock, 1002, frame=frame)
@@ -1024,27 +1029,27 @@ class TestJavaWebsockets(TestApplicationJava):
self.close_connection(sock)
# Unit does not support UTF-8 validation
#
# # 6_3_1 FAIL
#
# payload_1 = '\xce\xba\xe1\xbd\xb9\xcf\x83\xce\xbc\xce\xb5'
# payload_2 = '\xed\xa0\x80'
# payload_3 = '\x65\x64\x69\x74\x65\x64'
#
# payload = payload_1 + payload_2 + payload_3
#
# self.ws.message(sock, self.ws.OP_TEXT, payload)
# self.check_close(sock, 1007)
#
# # 6_3_2 FAIL
#
# _, sock, _ = self.ws.upgrade()
#
# self.ws.message(sock, self.ws.OP_TEXT, payload, fragmention_size=1)
# self.check_close(sock, 1007)
#
# # 6_4_1 ... 6_4_4 FAIL
# Unit does not support UTF-8 validation
#
# # 6_3_1 FAIL
#
# payload_1 = '\xce\xba\xe1\xbd\xb9\xcf\x83\xce\xbc\xce\xb5'
# payload_2 = '\xed\xa0\x80'
# payload_3 = '\x65\x64\x69\x74\x65\x64'
#
# payload = payload_1 + payload_2 + payload_3
#
# self.ws.message(sock, self.ws.OP_TEXT, payload)
# self.check_close(sock, 1007)
#
# # 6_3_2 FAIL
#
# _, sock, _ = self.ws.upgrade()
#
# self.ws.message(sock, self.ws.OP_TEXT, payload, fragmention_size=1)
# self.check_close(sock, 1007)
#
# # 6_4_1 ... 6_4_4 FAIL
def test_java_websockets_7_1_1__7_5_1(self):
self.load('websockets_mirror')
@@ -1171,15 +1176,15 @@ class TestJavaWebsockets(TestApplicationJava):
self.ws.frame_write(sock, self.ws.OP_CLOSE, payload)
self.check_close(sock, 1002)
# # 7_5_1 FAIL Unit does not support UTF-8 validation
#
# _, sock, _ = self.ws.upgrade()
#
# payload = self.ws.serialize_close(reason = '\xce\xba\xe1\xbd\xb9\xcf' \
# '\x83\xce\xbc\xce\xb5\xed\xa0\x80\x65\x64\x69\x74\x65\x64')
#
# self.ws.frame_write(sock, self.ws.OP_CLOSE, payload)
# self.check_close(sock, 1007)
# # 7_5_1 FAIL Unit does not support UTF-8 validation
#
# _, sock, _ = self.ws.upgrade()
#
# payload = self.ws.serialize_close(reason = '\xce\xba\xe1\xbd\xb9\xcf' \
# '\x83\xce\xbc\xce\xb5\xed\xa0\x80\x65\x64\x69\x74\x65\x64')
#
# self.ws.frame_write(sock, self.ws.OP_CLOSE, payload)
# self.check_close(sock, 1007)
def test_java_websockets_7_7_X__7_9_X(self):
self.load('websockets_mirror')
@@ -1282,52 +1287,52 @@ class TestJavaWebsockets(TestApplicationJava):
frame = self.ws.frame_read(sock, read_timeout=5)
self.check_frame(frame, True, opcode, payload)
check_payload(op_text, 64 * 2 ** 10) # 9_1_1
check_payload(op_text, 256 * 2 ** 10) # 9_1_2
check_payload(op_text, 2 ** 20) # 9_1_3
check_payload(op_text, 4 * 2 ** 20) # 9_1_4
check_payload(op_text, 8 * 2 ** 20) # 9_1_5
check_payload(op_text, 16 * 2 ** 20) # 9_1_6
check_payload(op_text, 64 * 2 ** 10) # 9_1_1
check_payload(op_text, 256 * 2 ** 10) # 9_1_2
check_payload(op_text, 2 ** 20) # 9_1_3
check_payload(op_text, 4 * 2 ** 20) # 9_1_4
check_payload(op_text, 8 * 2 ** 20) # 9_1_5
check_payload(op_text, 16 * 2 ** 20) # 9_1_6
check_payload(op_binary, 64 * 2 ** 10) # 9_2_1
check_payload(op_binary, 256 * 2 ** 10) # 9_2_2
check_payload(op_binary, 2 ** 20) # 9_2_3
check_payload(op_binary, 4 * 2 ** 20) # 9_2_4
check_payload(op_binary, 8 * 2 ** 20) # 9_2_5
check_payload(op_binary, 16 * 2 ** 20) # 9_2_6
check_payload(op_binary, 64 * 2 ** 10) # 9_2_1
check_payload(op_binary, 256 * 2 ** 10) # 9_2_2
check_payload(op_binary, 2 ** 20) # 9_2_3
check_payload(op_binary, 4 * 2 ** 20) # 9_2_4
check_payload(op_binary, 8 * 2 ** 20) # 9_2_5
check_payload(op_binary, 16 * 2 ** 20) # 9_2_6
if option.system != 'Darwin' and option.system != 'FreeBSD':
check_message(op_text, 64) # 9_3_1
check_message(op_text, 256) # 9_3_2
check_message(op_text, 2 ** 10) # 9_3_3
check_message(op_text, 4 * 2 ** 10) # 9_3_4
check_message(op_text, 16 * 2 ** 10) # 9_3_5
check_message(op_text, 64 * 2 ** 10) # 9_3_6
check_message(op_text, 256 * 2 ** 10) # 9_3_7
check_message(op_text, 2 ** 20) # 9_3_8
check_message(op_text, 4 * 2 ** 20) # 9_3_9
check_message(op_text, 64) # 9_3_1
check_message(op_text, 256) # 9_3_2
check_message(op_text, 2 ** 10) # 9_3_3
check_message(op_text, 4 * 2 ** 10) # 9_3_4
check_message(op_text, 16 * 2 ** 10) # 9_3_5
check_message(op_text, 64 * 2 ** 10) # 9_3_6
check_message(op_text, 256 * 2 ** 10) # 9_3_7
check_message(op_text, 2 ** 20) # 9_3_8
check_message(op_text, 4 * 2 ** 20) # 9_3_9
check_message(op_binary, 64) # 9_4_1
check_message(op_binary, 256) # 9_4_2
check_message(op_binary, 2 ** 10) # 9_4_3
check_message(op_binary, 4 * 2 ** 10) # 9_4_4
check_message(op_binary, 16 * 2 ** 10) # 9_4_5
check_message(op_binary, 64 * 2 ** 10) # 9_4_6
check_message(op_binary, 256 * 2 ** 10) # 9_4_7
check_message(op_binary, 2 ** 20) # 9_4_8
check_message(op_binary, 4 * 2 ** 20) # 9_4_9
check_message(op_binary, 64) # 9_4_1
check_message(op_binary, 256) # 9_4_2
check_message(op_binary, 2 ** 10) # 9_4_3
check_message(op_binary, 4 * 2 ** 10) # 9_4_4
check_message(op_binary, 16 * 2 ** 10) # 9_4_5
check_message(op_binary, 64 * 2 ** 10) # 9_4_6
check_message(op_binary, 256 * 2 ** 10) # 9_4_7
check_message(op_binary, 2 ** 20) # 9_4_8
check_message(op_binary, 4 * 2 ** 20) # 9_4_9
check_payload(op_text, 2 ** 20, chopsize=64) # 9_5_1
check_payload(op_text, 2 ** 20, chopsize=128) # 9_5_2
check_payload(op_text, 2 ** 20, chopsize=256) # 9_5_3
check_payload(op_text, 2 ** 20, chopsize=512) # 9_5_4
check_payload(op_text, 2 ** 20, chopsize=1024) # 9_5_5
check_payload(op_text, 2 ** 20, chopsize=2048) # 9_5_6
check_payload(op_text, 2 ** 20, chopsize=64) # 9_5_1
check_payload(op_text, 2 ** 20, chopsize=128) # 9_5_2
check_payload(op_text, 2 ** 20, chopsize=256) # 9_5_3
check_payload(op_text, 2 ** 20, chopsize=512) # 9_5_4
check_payload(op_text, 2 ** 20, chopsize=1024) # 9_5_5
check_payload(op_text, 2 ** 20, chopsize=2048) # 9_5_6
check_payload(op_binary, 2 ** 20, chopsize=64) # 9_6_1
check_payload(op_binary, 2 ** 20, chopsize=128) # 9_6_2
check_payload(op_binary, 2 ** 20, chopsize=256) # 9_6_3
check_payload(op_binary, 2 ** 20, chopsize=512) # 9_6_4
check_payload(op_binary, 2 ** 20, chopsize=64) # 9_6_1
check_payload(op_binary, 2 ** 20, chopsize=128) # 9_6_2
check_payload(op_binary, 2 ** 20, chopsize=256) # 9_6_3
check_payload(op_binary, 2 ** 20, chopsize=512) # 9_6_4
check_payload(op_binary, 2 ** 20, chopsize=1024) # 9_6_5
check_payload(op_binary, 2 ** 20, chopsize=2048) # 9_6_6

5
test/test_node_application.py
View File

@@ -1,6 +1,7 @@
import re
import pytest
from unit.applications.lang.node import TestApplicationNode
from unit.utils import waitforfiles
@@ -205,7 +206,9 @@ class TestNodeApplication(TestApplicationNode):
def test_node_application_status_message(self):
self.load('status_message')
assert re.search(r'200 blah', self.get(raw_resp=True)), 'status message'
assert re.search(
r'200 blah', self.get(raw_resp=True)
), 'status message'
def test_node_application_get_header_type(self):
self.load('get_header_type')

191
test/test_node_websockets.py
View File

@@ -2,6 +2,7 @@ import struct
import time
import pytest
from unit.applications.lang.node import TestApplicationNode
from unit.applications.websockets import TestApplicationWebsocket
from unit.option import option
@@ -182,7 +183,7 @@ class TestNodeWebsockets(TestApplicationNode):
@pytest.mark.skip('not yet')
def test_node_websockets_handshake_upgrade_absent(
self
self,
): # FAIL https://tools.ietf.org/html/rfc6455#section-4.2.1
self.load('websockets/mirror')
@@ -275,7 +276,9 @@ class TestNodeWebsockets(TestApplicationNode):
},
)
assert resp['status'] == 400, 'key double' # FAIL https://tools.ietf.org/html/rfc6455#section-11.3.1
assert (
resp['status'] == 400
), 'key double' # FAIL https://tools.ietf.org/html/rfc6455#section-11.3.1
def test_node_websockets_handshake_method_invalid(self):
self.load('websockets/mirror')
@@ -370,14 +373,14 @@ class TestNodeWebsockets(TestApplicationNode):
frame = self.ws.frame_read(sock)
self.check_frame(frame, True, opcode, payload)
check_length(0) # 1_1_1
check_length(125) # 1_1_2
check_length(126) # 1_1_3
check_length(127) # 1_1_4
check_length(128) # 1_1_5
check_length(65535) # 1_1_6
check_length(65536) # 1_1_7
check_length(65536, chopsize = 997) # 1_1_8
check_length(0) # 1_1_1
check_length(125) # 1_1_2
check_length(126) # 1_1_3
check_length(127) # 1_1_4
check_length(128) # 1_1_5
check_length(65535) # 1_1_6
check_length(65536) # 1_1_7
check_length(65536, chopsize=997) # 1_1_8
self.close_connection(sock)
@@ -396,14 +399,14 @@ class TestNodeWebsockets(TestApplicationNode):
self.check_frame(frame, True, opcode, payload)
check_length(0) # 1_2_1
check_length(125) # 1_2_2
check_length(126) # 1_2_3
check_length(127) # 1_2_4
check_length(128) # 1_2_5
check_length(65535) # 1_2_6
check_length(65536) # 1_2_7
check_length(65536, chopsize = 997) # 1_2_8
check_length(0) # 1_2_1
check_length(125) # 1_2_2
check_length(126) # 1_2_3
check_length(127) # 1_2_4
check_length(128) # 1_2_5
check_length(65535) # 1_2_6
check_length(65536) # 1_2_7
check_length(65536, chopsize=997) # 1_2_8
self.close_connection(sock)
@@ -421,11 +424,11 @@ class TestNodeWebsockets(TestApplicationNode):
self.check_frame(frame, True, op_pong, payload, decode=decode)
check_ping('') # 2_1
check_ping('Hello, world!') # 2_2
check_ping('') # 2_1
check_ping('Hello, world!') # 2_2
check_ping(b'\x00\xff\xfe\xfd\xfc\xfb\x00\xff', decode=False) # 2_3
check_ping(b'\xfe' * 125, decode=False) # 2_4
check_ping(b'\xfe' * 125, chopsize=1, decode=False) # 2_6
check_ping(b'\xfe' * 125, decode=False) # 2_4
check_ping(b'\xfe' * 125, chopsize=1, decode=False) # 2_6
self.close_connection(sock)
@@ -886,7 +889,9 @@ class TestNodeWebsockets(TestApplicationNode):
frame = self.ws.frame_read(sock)
if frame['opcode'] == self.ws.OP_TEXT:
self.check_frame(frame, True, self.ws.OP_TEXT, 'fragment1fragment2')
self.check_frame(
frame, True, self.ws.OP_TEXT, 'fragment1fragment2'
)
frame = None
self.check_close(sock, 1002, frame=frame)
@@ -1043,27 +1048,27 @@ class TestNodeWebsockets(TestApplicationNode):
self.close_connection(sock)
# Unit does not support UTF-8 validation
#
# # 6_3_1 FAIL
#
# payload_1 = '\xce\xba\xe1\xbd\xb9\xcf\x83\xce\xbc\xce\xb5'
# payload_2 = '\xed\xa0\x80'
# payload_3 = '\x65\x64\x69\x74\x65\x64'
#
# payload = payload_1 + payload_2 + payload_3
#
# self.ws.message(sock, self.ws.OP_TEXT, payload)
# self.check_close(sock, 1007)
#
# # 6_3_2 FAIL
#
# _, sock, _ = self.ws.upgrade()
#
# self.ws.message(sock, self.ws.OP_TEXT, payload, fragmention_size=1)
# self.check_close(sock, 1007)
#
# # 6_4_1 ... 6_4_4 FAIL
# Unit does not support UTF-8 validation
#
# # 6_3_1 FAIL
#
# payload_1 = '\xce\xba\xe1\xbd\xb9\xcf\x83\xce\xbc\xce\xb5'
# payload_2 = '\xed\xa0\x80'
# payload_3 = '\x65\x64\x69\x74\x65\x64'
#
# payload = payload_1 + payload_2 + payload_3
#
# self.ws.message(sock, self.ws.OP_TEXT, payload)
# self.check_close(sock, 1007)
#
# # 6_3_2 FAIL
#
# _, sock, _ = self.ws.upgrade()
#
# self.ws.message(sock, self.ws.OP_TEXT, payload, fragmention_size=1)
# self.check_close(sock, 1007)
#
# # 6_4_1 ... 6_4_4 FAIL
def test_node_websockets_7_1_1__7_5_1(self):
self.load('websockets/mirror')
@@ -1190,15 +1195,15 @@ class TestNodeWebsockets(TestApplicationNode):
self.ws.frame_write(sock, self.ws.OP_CLOSE, payload)
self.check_close(sock, 1002)
# # 7_5_1 FAIL Unit does not support UTF-8 validation
#
# _, sock, _ = self.ws.upgrade()
#
# payload = self.ws.serialize_close(reason = '\xce\xba\xe1\xbd\xb9\xcf' \
# '\x83\xce\xbc\xce\xb5\xed\xa0\x80\x65\x64\x69\x74\x65\x64')
#
# self.ws.frame_write(sock, self.ws.OP_CLOSE, payload)
# self.check_close(sock, 1007)
# # 7_5_1 FAIL Unit does not support UTF-8 validation
#
# _, sock, _ = self.ws.upgrade()
#
# payload = self.ws.serialize_close(reason = '\xce\xba\xe1\xbd\xb9\xcf' \
# '\x83\xce\xbc\xce\xb5\xed\xa0\x80\x65\x64\x69\x74\x65\x64')
#
# self.ws.frame_write(sock, self.ws.OP_CLOSE, payload)
# self.check_close(sock, 1007)
def test_node_websockets_7_7_X__7_9_X(self):
self.load('websockets/mirror')
@@ -1301,52 +1306,52 @@ class TestNodeWebsockets(TestApplicationNode):
frame = self.ws.frame_read(sock, read_timeout=5)
self.check_frame(frame, True, opcode, payload)
check_payload(op_text, 64 * 2 ** 10) # 9_1_1
check_payload(op_text, 256 * 2 ** 10) # 9_1_2
check_payload(op_text, 2 ** 20) # 9_1_3
check_payload(op_text, 4 * 2 ** 20) # 9_1_4
check_payload(op_text, 8 * 2 ** 20) # 9_1_5
check_payload(op_text, 16 * 2 ** 20) # 9_1_6
check_payload(op_text, 64 * 2 ** 10) # 9_1_1
check_payload(op_text, 256 * 2 ** 10) # 9_1_2
check_payload(op_text, 2 ** 20) # 9_1_3
check_payload(op_text, 4 * 2 ** 20) # 9_1_4
check_payload(op_text, 8 * 2 ** 20) # 9_1_5
check_payload(op_text, 16 * 2 ** 20) # 9_1_6
check_payload(op_binary, 64 * 2 ** 10) # 9_2_1
check_payload(op_binary, 256 * 2 ** 10) # 9_2_2
check_payload(op_binary, 2 ** 20) # 9_2_3
check_payload(op_binary, 4 * 2 ** 20) # 9_2_4
check_payload(op_binary, 8 * 2 ** 20) # 9_2_5
check_payload(op_binary, 16 * 2 ** 20) # 9_2_6
check_payload(op_binary, 64 * 2 ** 10) # 9_2_1
check_payload(op_binary, 256 * 2 ** 10) # 9_2_2
check_payload(op_binary, 2 ** 20) # 9_2_3
check_payload(op_binary, 4 * 2 ** 20) # 9_2_4
check_payload(op_binary, 8 * 2 ** 20) # 9_2_5
check_payload(op_binary, 16 * 2 ** 20) # 9_2_6
if option.system != 'Darwin' and option.system != 'FreeBSD':
check_message(op_text, 64) # 9_3_1
check_message(op_text, 256) # 9_3_2
check_message(op_text, 2 ** 10) # 9_3_3
check_message(op_text, 4 * 2 ** 10) # 9_3_4
check_message(op_text, 16 * 2 ** 10) # 9_3_5
check_message(op_text, 64 * 2 ** 10) # 9_3_6
check_message(op_text, 256 * 2 ** 10) # 9_3_7
check_message(op_text, 2 ** 20) # 9_3_8
check_message(op_text, 4 * 2 ** 20) # 9_3_9
check_message(op_text, 64) # 9_3_1
check_message(op_text, 256) # 9_3_2
check_message(op_text, 2 ** 10) # 9_3_3
check_message(op_text, 4 * 2 ** 10) # 9_3_4
check_message(op_text, 16 * 2 ** 10) # 9_3_5
check_message(op_text, 64 * 2 ** 10) # 9_3_6
check_message(op_text, 256 * 2 ** 10) # 9_3_7
check_message(op_text, 2 ** 20) # 9_3_8
check_message(op_text, 4 * 2 ** 20) # 9_3_9
check_message(op_binary, 64) # 9_4_1
check_message(op_binary, 256) # 9_4_2
check_message(op_binary, 2 ** 10) # 9_4_3
check_message(op_binary, 4 * 2 ** 10) # 9_4_4
check_message(op_binary, 16 * 2 ** 10) # 9_4_5
check_message(op_binary, 64 * 2 ** 10) # 9_4_6
check_message(op_binary, 256 * 2 ** 10) # 9_4_7
check_message(op_binary, 2 ** 20) # 9_4_8
check_message(op_binary, 4 * 2 ** 20) # 9_4_9
check_message(op_binary, 64) # 9_4_1
check_message(op_binary, 256) # 9_4_2
check_message(op_binary, 2 ** 10) # 9_4_3
check_message(op_binary, 4 * 2 ** 10) # 9_4_4
check_message(op_binary, 16 * 2 ** 10) # 9_4_5
check_message(op_binary, 64 * 2 ** 10) # 9_4_6
check_message(op_binary, 256 * 2 ** 10) # 9_4_7
check_message(op_binary, 2 ** 20) # 9_4_8
check_message(op_binary, 4 * 2 ** 20) # 9_4_9
check_payload(op_text, 2 ** 20, chopsize=64) # 9_5_1
check_payload(op_text, 2 ** 20, chopsize=128) # 9_5_2
check_payload(op_text, 2 ** 20, chopsize=256) # 9_5_3
check_payload(op_text, 2 ** 20, chopsize=512) # 9_5_4
check_payload(op_text, 2 ** 20, chopsize=1024) # 9_5_5
check_payload(op_text, 2 ** 20, chopsize=2048) # 9_5_6
check_payload(op_text, 2 ** 20, chopsize=64) # 9_5_1
check_payload(op_text, 2 ** 20, chopsize=128) # 9_5_2
check_payload(op_text, 2 ** 20, chopsize=256) # 9_5_3
check_payload(op_text, 2 ** 20, chopsize=512) # 9_5_4
check_payload(op_text, 2 ** 20, chopsize=1024) # 9_5_5
check_payload(op_text, 2 ** 20, chopsize=2048) # 9_5_6
check_payload(op_binary, 2 ** 20, chopsize=64) # 9_6_1
check_payload(op_binary, 2 ** 20, chopsize=128) # 9_6_2
check_payload(op_binary, 2 ** 20, chopsize=256) # 9_6_3
check_payload(op_binary, 2 ** 20, chopsize=512) # 9_6_4
check_payload(op_binary, 2 ** 20, chopsize=64) # 9_6_1
check_payload(op_binary, 2 ** 20, chopsize=128) # 9_6_2
check_payload(op_binary, 2 ** 20, chopsize=256) # 9_6_3
check_payload(op_binary, 2 ** 20, chopsize=512) # 9_6_4
check_payload(op_binary, 2 ** 20, chopsize=1024) # 9_6_5
check_payload(op_binary, 2 ** 20, chopsize=2048) # 9_6_6

1
test/test_perl_application.py
View File

@@ -1,6 +1,7 @@
import re
import pytest
from unit.applications.lang.perl import TestApplicationPerl

14
test/test_php_application.py
View File

@@ -5,9 +5,11 @@ import time
from subprocess import call
import pytest
from unit.applications.lang.php import TestApplicationPHP
from unit.option import option
class TestPHPApplication(TestApplicationPHP):
prerequisites = {'modules': {'php': 'all'}}
@@ -428,11 +430,13 @@ class TestPHPApplication(TestApplicationPHP):
self.load('auth')
def check_auth(auth):
resp = self.get(headers={
'Host': 'localhost',
'Authorization': auth,
'Connection': 'close',
})
resp = self.get(
headers={
'Host': 'localhost',
'Authorization': auth,
'Connection': 'close',
}
)
assert resp['status'] == 200, 'status'
assert resp['headers']['X-Digest'] == 'not set', 'Digest'

5
test/test_php_isolation.py
View File

@@ -1,4 +1,5 @@
import pytest
from unit.applications.lang.php import TestApplicationPHP
from unit.option import option
@@ -28,7 +29,7 @@ class TestPHPIsolation(TestApplicationPHP):
isolation['namespaces'] = {
'mount': True,
'credential': True,
'pid': True
'pid': True,
}
self.load('phpinfo', isolation=isolation)
@@ -64,7 +65,7 @@ class TestPHPIsolation(TestApplicationPHP):
isolation['namespaces'] = {
'mount': True,
'credential': True,
'pid': True
'pid': True,
}
self.load('list-extensions', isolation=isolation)

10
test/test_proxy.py
View File

@@ -3,6 +3,7 @@ import socket
import time
import pytest
from conftest import run_process
from unit.applications.lang.python import TestApplicationPython
from unit.option import option
@@ -464,9 +465,9 @@ Content-Length: 10
def test_proxy_invalid(self):
def check_proxy(proxy):
assert 'error' in \
self.conf([{"action": {"proxy": proxy}}], 'routes'), \
'proxy invalid'
assert 'error' in self.conf(
[{"action": {"proxy": proxy}}], 'routes'
), 'proxy invalid'
check_proxy('blah')
check_proxy('/blah')
@@ -502,7 +503,8 @@ Content-Length: 10
"type": "python",
"processes": {"spare": 0},
"path": option.test_dir + "/python/mirror",
"working_directory": option.test_dir + "/python/mirror",
"working_directory": option.test_dir
+ "/python/mirror",
"module": "wsgi",
},
},

10
test/test_python_application.py
View File

@@ -5,6 +5,7 @@ import re
import time
import pytest
from unit.applications.lang.python import TestApplicationPython
from unit.option import option
@@ -31,7 +32,8 @@ Content-Type: text/html
Connection: close
custom-header: BLAH
%s""" % (len(body), body.encode()),
%s"""
% (len(body), body.encode()),
raw=True,
)
@@ -816,7 +818,11 @@ last line: 987654321
assert ['/new', *sys_path] == get_path(), 'check path update'
set_path('["/blah1", "/blah2"]')
assert ['/blah1', '/blah2', *sys_path] == get_path(), 'check path array'
assert [
'/blah1',
'/blah2',
*sys_path,
] == get_path(), 'check path array'
def test_python_application_path_invalid(self):
self.load('path')

18
test/test_python_isolation.py
View File

@@ -1,5 +1,5 @@
import pytest
from unit.applications.lang.python import TestApplicationPython
from unit.option import option
from unit.utils import findmnt
@@ -32,7 +32,7 @@ class TestPythonIsolation(TestApplicationPython):
isolation['namespaces'] = {
'mount': True,
'credential': True,
'pid': True
'pid': True,
}
self.load('ns_inspect', isolation=isolation)
@@ -43,8 +43,7 @@ class TestPythonIsolation(TestApplicationPython):
), 'temp_dir does not exists in rootfs'
assert (
self.getjson(url='/?path=/proc/self')['body']['FileExists']
== True
self.getjson(url='/?path=/proc/self')['body']['FileExists'] == True
), 'no /proc/self'
assert (
@@ -66,15 +65,12 @@ class TestPythonIsolation(TestApplicationPython):
if not is_su:
pytest.skip('requires root')
isolation = {
'rootfs': temp_dir,
'automount': {'language_deps': False}
}
isolation = {'rootfs': temp_dir, 'automount': {'language_deps': False}}
self.load('empty', isolation=isolation)
assert findmnt().find(temp_dir) == -1
assert (self.get()['status'] != 200), 'disabled language_deps'
assert self.get()['status'] != 200, 'disabled language_deps'
assert findmnt().find(temp_dir) == -1
isolation['automount']['language_deps'] = True
@@ -82,7 +78,7 @@ class TestPythonIsolation(TestApplicationPython):
self.load('empty', isolation=isolation)
assert findmnt().find(temp_dir) == -1
assert (self.get()['status'] == 200), 'enabled language_deps'
assert self.get()['status'] == 200, 'enabled language_deps'
assert waitformount(temp_dir), 'language_deps mount'
self.conf({"listeners": {}, "applications": {}})
@@ -90,8 +86,6 @@ class TestPythonIsolation(TestApplicationPython):
assert waitforunmount(temp_dir), 'language_deps unmount'
def test_python_isolation_procfs(self, is_su, temp_dir):
isolation_features = option.available['features']['isolation'].keys()
if not is_su:
pytest.skip('requires root')

4
test/test_python_isolation_chroot.py
View File

@@ -1,4 +1,5 @@
import pytest
from unit.applications.lang.python import TestApplicationPython
@@ -21,8 +22,7 @@ class TestPythonIsolation(TestApplicationPython):
), 'temp_dir does not exists in rootfs'
assert (
self.getjson(url='/?path=/proc/self')['body']['FileExists']
== True
self.getjson(url='/?path=/proc/self')['body']['FileExists'] == True
), 'no /proc/self'
assert (

1
test/test_python_procman.py
View File

@@ -3,6 +3,7 @@ import subprocess
import time
import pytest
from unit.applications.lang.python import TestApplicationPython
from unit.option import option

7
test/test_respawn.py
View File

@@ -76,9 +76,10 @@ class TestRespawn(TestApplicationPython):
self.kill_pids(pid)
skip_alert(r'process %s exited on signal 9' % pid)
assert self.wait_for_process(
self.PATTERN_CONTROLLER, unit_pid
) is not None
assert (
self.wait_for_process(self.PATTERN_CONTROLLER, unit_pid)
is not None
)
assert self.get()['status'] == 200

11
test/test_routing.py
View File

@@ -1,5 +1,6 @@
# -*- coding: utf-8 -*-
import pytest
from unit.applications.proto import TestApplicationProto
from unit.option import option
@@ -232,11 +233,11 @@ class TestRouting(TestApplicationProto):
if not option.available['modules']['regex']:
pytest.skip('requires regex')
self.route_match({"uri":"~"})
self.route_match({"uri": "~"})
assert self.get(url='/')['status'] == 200, 'empty regexp'
assert self.get(url='/anything')['status'] == 200, '/anything'
self.route_match({"uri":"!~"})
self.route_match({"uri": "!~"})
assert self.get(url='/')['status'] == 404, 'empty regexp 2'
assert self.get(url='/nothing')['status'] == 404, '/nothing'
@@ -336,8 +337,7 @@ class TestRouting(TestApplicationProto):
"type": "python",
"processes": {"spare": 0},
"path": option.test_dir + '/python/empty',
"working_directory": option.test_dir
+ '/python/empty',
"working_directory": option.test_dir + '/python/empty',
"module": "wsgi",
}
},
@@ -495,8 +495,7 @@ class TestRouting(TestApplicationProto):
'routes/0/action',
), 'proxy pass'
assert 'error' in self.conf(
{"share": temp_dir, "pass": "applications/app"},
'routes/0/action',
{"share": temp_dir, "pass": "applications/app"}, 'routes/0/action',
), 'share pass'
def test_routes_rules_two(self):

12
test/test_ruby_application.py
View File

@@ -2,6 +2,7 @@ import re
import subprocess
import pytest
from unit.applications.lang.ruby import TestApplicationRuby
@@ -208,7 +209,6 @@ class TestRubyApplication(TestApplicationRuby):
self.get()
assert (
self.wait_for_record(r'\[error\].+1234567890') is not None
), 'errors write int'
@@ -228,9 +228,13 @@ class TestRubyApplication(TestApplicationRuby):
self.load('encoding')
try:
locales = subprocess.check_output(
['locale', '-a'], stderr=subprocess.STDOUT,
).decode().split('\n')
locales = (
subprocess.check_output(
['locale', '-a'], stderr=subprocess.STDOUT,
)
.decode()
.split('\n')
)
except (FileNotFoundError, subprocess.CalledProcessError):
pytest.skip('require locale')

1
test/test_ruby_isolation.py
View File

@@ -2,6 +2,7 @@ import os
import shutil
import pytest
from unit.applications.lang.ruby import TestApplicationRuby
from unit.option import option

1
test/test_settings.py
View File

@@ -3,6 +3,7 @@ import socket
import time
import pytest
from unit.applications.lang.python import TestApplicationPython

6
test/test_share_fallback.py
View File

@@ -1,6 +1,7 @@
import os
import pytest
from unit.applications.proto import TestApplicationProto
from unit.option import option
@@ -81,10 +82,7 @@ class TestStatic(TestApplicationProto):
def test_fallback_share(self, temp_dir):
self.action_update(
{
"share": "/blah",
"fallback": {"share": temp_dir + "/assets"},
}
{"share": "/blah", "fallback": {"share": temp_dir + "/assets"},}
)
resp = self.get()

11
test/test_static.py
View File

@@ -2,6 +2,7 @@ import os
import socket
import pytest
from unit.applications.proto import TestApplicationProto
from unit.option import option
from unit.utils import waitforfiles
@@ -85,8 +86,7 @@ class TestStatic(TestApplicationProto):
def test_static_space_in_name(self, temp_dir):
os.rename(
temp_dir + '/assets/dir/file',
temp_dir + '/assets/dir/fi le',
temp_dir + '/assets/dir/file', temp_dir + '/assets/dir/fi le',
)
assert waitforfiles(temp_dir + '/assets/dir/fi le')
assert self.get(url='/dir/fi le')['body'] == 'blah', 'file name'
@@ -95,9 +95,7 @@ class TestStatic(TestApplicationProto):
assert waitforfiles(temp_dir + '/assets/di r/fi le')
assert self.get(url='/di r/fi le')['body'] == 'blah', 'dir name'
os.rename(
temp_dir + '/assets/di r', temp_dir + '/assets/ di r '
)
os.rename(temp_dir + '/assets/di r', temp_dir + '/assets/ di r ')
assert waitforfiles(temp_dir + '/assets/ di r /fi le')
assert (
self.get(url='/ di r /fi le')['body'] == 'blah'
@@ -150,8 +148,7 @@ class TestStatic(TestApplicationProto):
), 'file name 2'
os.rename(
temp_dir + '/assets/ di r ',
temp_dir + '/assets/ди ректория',
temp_dir + '/assets/ di r ', temp_dir + '/assets/ди ректория',
)
assert waitforfiles(temp_dir + '/assets/ди ректория/фа йл')
assert (

3
test/test_tls.py
View File

@@ -4,6 +4,7 @@ import ssl
import subprocess
import pytest
from unit.applications.tls import TestApplicationTLS
from unit.option import option
@@ -22,7 +23,7 @@ class TestTLS(TestApplicationTLS):
assert 'success' in self.conf(
{
"pass": "applications/" + application,
"tls": {"certificate": cert}
"tls": {"certificate": cert},
},
'listeners/*:' + str(port),
)

28
test/test_tls_sni.py
View File

@@ -1,7 +1,8 @@
import subprocess
import ssl
import subprocess
import pytest
from unit.applications.tls import TestApplicationTLS
from unit.option import option
@@ -23,10 +24,7 @@ class TestTLSSNI(TestApplicationTLS):
def add_tls(self, cert='default'):
assert 'success' in self.conf(
{
"pass": "routes",
"tls": {"certificate": cert}
},
{"pass": "routes", "tls": {"certificate": cert}},
'listeners/*:7080',
)
@@ -153,10 +151,7 @@ basicConstraints = critical,CA:TRUE"""
def test_tls_sni(self):
bundles = {
"default": {
"subj": "default",
"alt_names": ["default"],
},
"default": {"subj": "default", "alt_names": ["default"]},
"localhost.com": {
"subj": "localhost.com",
"alt_names": ["alt1.localhost.com"],
@@ -205,10 +200,7 @@ basicConstraints = critical,CA:TRUE"""
def test_tls_sni_wildcard(self):
bundles = {
"localhost.com": {
"subj": "localhost.com",
"alt_names": [],
},
"localhost.com": {"subj": "localhost.com", "alt_names": []},
"example.com": {
"subj": "example.com",
"alt_names": ["*.example.com", "*.alt.example.com"],
@@ -248,11 +240,7 @@ basicConstraints = critical,CA:TRUE"""
self.check_cert('example', bundles['localhost']['subj'])
def test_tls_sni_empty_cn(self):
bundles = {
"localhost": {
"alt_names": ["alt.localhost.com"],
}
}
bundles = {"localhost": {"alt_names": ["alt.localhost.com"]}}
self.config_bundles(bundles)
self.add_tls(["localhost"])
@@ -266,7 +254,9 @@ basicConstraints = critical,CA:TRUE"""
)
assert resp['status'] == 200
assert sock.getpeercert()['subjectAltName'][0][1] == 'alt.localhost.com'
assert (
sock.getpeercert()['subjectAltName'][0][1] == 'alt.localhost.com'
)
def test_tls_sni_invalid(self):
self.config_bundles({"localhost": {"subj": "subj1", "alt_names": ''}})

12
test/test_variables.py
View File

@@ -92,16 +92,8 @@ class TestVariables(TestApplicationProto):
"*:7080": {"pass": "upstreams$uri"},
"*:7081": {"pass": "routes/one"},
},
"upstreams": {
"1": {
"servers": {
"127.0.0.1:7081": {},
},
},
},
"routes": {
"one": [{"action": {"return": 200}}],
},
"upstreams": {"1": {"servers": {"127.0.0.1:7081": {}}}},
"routes": {"one": [{"action": {"return": 200}}]},
},
), 'upstreams initial configuration'

22
test/unit/applications/tls.py
View File

@@ -21,10 +21,14 @@ class TestApplicationTLS(TestApplicationProto):
'req',
'-x509',
'-new',
'-subj', '/CN=' + name + '/',
'-config', option.temp_dir + '/openssl.conf',
'-out', option.temp_dir + '/' + name + '.crt',
'-keyout', option.temp_dir + '/' + name + '.key',
'-subj',
'/CN=' + name + '/',
'-config',
option.temp_dir + '/openssl.conf',
'-out',
option.temp_dir + '/' + name + '.crt',
'-keyout',
option.temp_dir + '/' + name + '.key',
],
stderr=subprocess.STDOUT,
)
@@ -75,12 +79,14 @@ class TestApplicationTLS(TestApplicationProto):
a_names += "DNS.%d = %s\n" % (i, k)
# Generates section for sign request extension
a_sec = """req_extensions = myca_req_extensions
a_sec = """req_extensions = myca_req_extensions
[ myca_req_extensions ]
subjectAltName = @alt_names
{a_names}""".format(a_names=a_names)
{a_names}""".format(
a_names=a_names
)
with open(conf_path, 'w') as f:
f.write(
@@ -90,7 +96,9 @@ encrypt_key = no
distinguished_name = req_distinguished_name
{a_sec}
[ req_distinguished_name ]""".format(a_sec=a_sec if alt_names else "")
[ req_distinguished_name ]""".format(
a_sec=a_sec if alt_names else ""
)
)
def load(self, script, name=None):

18
test/unit/applications/websockets.py
View File

@@ -43,11 +43,7 @@ class TestApplicationWebsocket(TestApplicationProto):
'Sec-WebSocket-Version': 13,
}
_, sock = self.get(
headers=headers,
no_recv=True,
start=True,
)
_, sock = self.get(headers=headers, no_recv=True, start=True,)
resp = ''
while True:
@@ -57,7 +53,7 @@ class TestApplicationWebsocket(TestApplicationProto):
resp += sock.recv(4096).decode()
if (resp.startswith('HTTP/') and '\r\n\r\n' in resp):
if resp.startswith('HTTP/') and '\r\n\r\n' in resp:
resp = self._resp_to_dict(resp)
break
@@ -90,8 +86,8 @@ class TestApplicationWebsocket(TestApplicationProto):
frame = {}
head1, = struct.unpack('!B', recv_bytes(sock, 1))
head2, = struct.unpack('!B', recv_bytes(sock, 1))
(head1,) = struct.unpack('!B', recv_bytes(sock, 1))
(head2,) = struct.unpack('!B', recv_bytes(sock, 1))
frame['fin'] = bool(head1 & 0b10000000)
frame['rsv1'] = bool(head1 & 0b01000000)
@@ -103,10 +99,10 @@ class TestApplicationWebsocket(TestApplicationProto):
length = head2 & 0b01111111
if length == 126:
data = recv_bytes(sock, 2)
length, = struct.unpack('!H', data)
(length,) = struct.unpack('!H', data)
elif length == 127:
data = recv_bytes(sock, 8)
length, = struct.unpack('!Q', data)
(length,) = struct.unpack('!Q', data)
if frame['mask']:
mask_bits = recv_bytes(sock, 4)
@@ -121,7 +117,7 @@ class TestApplicationWebsocket(TestApplicationProto):
if frame['opcode'] == self.OP_CLOSE:
if length >= 2:
code, = struct.unpack('!H', data[:2])
(code,) = struct.unpack('!H', data[:2])
reason = data[2:].decode('utf-8')
if not (code in self.CLOSE_CODES or 3000 <= code < 5000):
pytest.fail('Invalid status code')

4
test/unit/check/isolation.py
View File

@@ -12,6 +12,7 @@ from unit.utils import getns
allns = ['pid', 'mnt', 'ipc', 'uts', 'cgroup', 'net']
http = TestHTTP()
def check_isolation():
test_conf = {"namespaces": {"credential": True}}
available = option.available
@@ -117,8 +118,7 @@ def check_isolation():
"body_empty": {
"type": "perl",
"processes": {"spare": 0},
"working_directory": option.test_dir
+ "/perl/body_empty",
"working_directory": option.test_dir + "/perl/body_empty",
"script": option.test_dir + "/perl/body_empty/psgi.pl",
"isolation": {"namespaces": {"credential": True}},
}

23
test/unit/http.py
View File

@@ -10,15 +10,16 @@ import pytest
from unit.option import option
class TestHTTP():
class TestHTTP:
def http(self, start_str, **kwargs):
sock_type = kwargs.get('sock_type', 'ipv4')
port = kwargs.get('port', 7080)
url = kwargs.get('url', '/')
http = 'HTTP/1.0' if 'http_10' in kwargs else 'HTTP/1.1'
headers = kwargs.get('headers',
{'Host': 'localhost', 'Connection': 'close'})
headers = kwargs.get(
'headers', {'Host': 'localhost', 'Connection': 'close'}
)
body = kwargs.get('body', b'')
crlf = '\r\n'
@@ -305,8 +306,9 @@ class TestHTTP():
return body, content_type
def form_url_encode(self, fields):
data = "&".join("%s=%s" % (name, value)
for name, value in fields.items()).encode()
data = "&".join(
"%s=%s" % (name, value) for name, value in fields.items()
).encode()
return data, 'application/x-www-form-urlencoded'
def multipart_encode(self, fields):
@@ -326,7 +328,9 @@ class TestHTTP():
datatype = value['type']
if not isinstance(value['data'], io.IOBase):
pytest.fail('multipart encoding of file requires a stream.')
pytest.fail(
'multipart encoding of file requires a stream.'
)
data = value['data'].read()
@@ -336,9 +340,10 @@ class TestHTTP():
else:
pytest.fail('multipart requires a string or stream data')
body += (
"--%s\r\nContent-Disposition: form-data; name=\"%s\""
) % (boundary, field)
body += ("--%s\r\nContent-Disposition: form-data; name=\"%s\"") % (
boundary,
field,
)
if filename != '':
body += "; filename=\"%s\"" % filename

3
test/unit/option.py
View File

@@ -1,4 +1,4 @@
class Options():
class Options:
_options = {
'skip_alerts': [],
'skip_sanitizer': False,
@@ -13,4 +13,5 @@ class Options():
raise AttributeError
option = Options()