phpged/unit
1
0
Fork 0
Code Issues 1 Pull requests Projects 1 Releases Packages Wiki Activity

Tests: style.

Browse source
This commit is contained in:
Andrei Zeliankou 2022-04-11 21:05:14 +01:00
parent aeed86c682
commit 0f72534660
36 changed files with 285 additions and 337 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

24
test/conftest.py
View file

@ -159,9 +159,7 @@ def pytest_generate_tests(metafunc):
type + ' ' + available_versions[0]
)
elif callable(prereq_version):
generate_tests(
list(filter(prereq_version, available_versions))
)
generate_tests(list(filter(prereq_version, available_versions)))
else:
raise ValueError(
@ -320,9 +318,7 @@ 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:
for attempt in range(10):
@ -520,7 +516,7 @@ def _clear_conf(sock, *, log=None):
try:
certs = json.loads(
http.get(url='/certificates', sock_type='unix', addr=sock,)['body']
http.get(url='/certificates', sock_type='unix', addr=sock)['body']
).keys()
except json.JSONDecodeError:
@ -528,7 +524,9 @@ def _clear_conf(sock, *, log=None):
for cert in certs:
resp = http.delete(
url='/certificates/' + cert, sock_type='unix', addr=sock,
url='/certificates/' + cert,
sock_type='unix',
addr=sock,
)['body']
assert 'success' in resp, 'remove certificate'
@ -554,9 +552,7 @@ def _check_fds(*, log=None):
)
ps['fds'] += fds_diff
assert (
fds_diff <= option.fds_threshold
), 'descriptors leak main process'
assert fds_diff <= option.fds_threshold, 'descriptors leak main process'
else:
ps['fds'] = _count_fds(unit_instance['pid'])
@ -588,7 +584,8 @@ def _count_fds(pid):
try:
out = subprocess.check_output(
['procstat', '-f', pid], stderr=subprocess.STDOUT,
['procstat', '-f', pid],
stderr=subprocess.STDOUT,
).decode()
return len(out.splitlines())
@ -597,7 +594,8 @@ def _count_fds(pid):
try:
out = subprocess.check_output(
['lsof', '-n', '-p', pid], stderr=subprocess.STDOUT,
['lsof', '-n', '-p', pid],
stderr=subprocess.STDOUT,
).decode()
return len(out.splitlines())

12
test/python/204_no_content/asgi.py
View file

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

88
test/test_asgi_websockets.py
View file

@ -164,7 +164,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
@ -942,9 +942,7 @@ 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)
@ -1189,7 +1187,7 @@ class TestASGIWebsockets(TestApplicationPython):
_, sock, _ = self.ws.upgrade()
self.ws.frame_write(sock, self.ws.OP_TEXT, 'BAsd7&jh23' * 26 * 2 ** 10)
self.ws.frame_write(sock, self.ws.OP_TEXT, 'BAsd7&jh23' * 26 * 2**10)
self.ws.frame_write(sock, self.ws.OP_TEXT, payload)
self.ws.frame_write(sock, self.ws.OP_CLOSE, self.ws.serialize_close())
@ -1351,62 +1349,62 @@ class TestASGIWebsockets(TestApplicationPython):
def check_message(opcode, f_size):
if opcode == self.ws.OP_TEXT:
payload = '*' * 4 * 2 ** 20
payload = '*' * 4 * 2**20
else:
payload = b'*' * 4 * 2 ** 20
payload = b'*' * 4 * 2**20
self.ws.message(sock, opcode, payload, fragmention_size=f_size)
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, 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, 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=1024) # 9_6_5
check_payload(op_binary, 2 ** 20, chopsize=2048) # 9_6_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=1024) # 9_6_5
check_payload(op_binary, 2**20, chopsize=2048) # 9_6_6
self.close_connection(sock)

8
test/test_go_application.py
View file

@ -157,9 +157,7 @@ class TestGoApplication(TestApplicationGo):
'applications/command_line_arguments/arguments',
)
assert (
self.get()['body'] == arg1 + ',' + arg2 + ',' + arg3
), 'arguments'
assert self.get()['body'] == arg1 + ',' + arg2 + ',' + arg3, 'arguments'
def test_go_application_command_line_arguments_change(self):
self.load('command_line_arguments')
@ -176,6 +174,4 @@ class TestGoApplication(TestApplicationGo):
assert 'success' in self.conf('[]', args_path)
assert (
self.get()['headers']['Content-Length'] == '0'
), 'arguments empty'
assert self.get()['headers']['Content-Length'] == '0', 'arguments empty'

4
test/test_go_isolation.py
View file

@ -167,9 +167,7 @@ class TestGoIsolation(TestApplicationGo):
user='nobody',
isolation={
'namespaces': {'credential': True},
'uidmap': [
{'container': 0, 'host': 0, 'size': nobody_uid + 1}
],
'uidmap': [{'container': 0, 'host': 0, 'size': nobody_uid + 1}],
},
)

16
test/test_http_header.py
View file

@ -213,7 +213,7 @@ Connection: close
self.post(
headers={
'Host': 'localhost',
'Content-Length': str(2 ** 64),
'Content-Length': str(2**64),
'Connection': 'close',
},
body='X' * 1000,
@ -325,9 +325,7 @@ Connection: close
def test_http_header_host_port_empty(self):
self.load('host')
resp = self.get(
headers={'Host': 'exmaple.com:', 'Connection': 'close'}
)
resp = self.get(headers={'Host': 'exmaple.com:', 'Connection': 'close'})
assert resp['status'] == 200, 'Host port empty status'
assert (
@ -376,9 +374,7 @@ Connection: close
def test_http_header_host_trailing_period_2(self):
self.load('host')
resp = self.get(
headers={'Host': 'EXAMPLE.COM.', 'Connection': 'close'}
)
resp = self.get(headers={'Host': 'EXAMPLE.COM.', 'Connection': 'close'})
assert resp['status'] == 200, 'Host trailing period 2 status'
assert (
@ -453,14 +449,16 @@ Connection: close
assert 'CUSTOM' not in resp['headers']['All-Headers']
assert 'success' in self.conf(
{'http': {'discard_unsafe_fields': False}}, 'settings',
{'http': {'discard_unsafe_fields': False}},
'settings',
)
resp = check_status("!#$%&'*+.^`|~Custom_Header")
assert 'CUSTOM' in resp['headers']['All-Headers']
assert 'success' in self.conf(
{'http': {'discard_unsafe_fields': True}}, 'settings',
{'http': {'discard_unsafe_fields': True}},
'settings',
)
resp = check_status("!Custom-Header")

44
test/test_java_application.py
View file

@ -173,9 +173,7 @@ class TestJavaApplication(TestApplicationJava):
}
)
assert (
resp['headers']['X-Session-Id'] == session_id
), 'session active 2'
assert resp['headers']['X-Session-Id'] == session_id, 'session active 2'
time.sleep(2)
@ -187,9 +185,7 @@ class TestJavaApplication(TestApplicationJava):
}
)
assert (
resp['headers']['X-Session-Id'] == session_id
), 'session active 3'
assert resp['headers']['X-Session-Id'] == session_id, 'session active 3'
def test_java_application_session_inactive(self):
self.load('session_inactive')
@ -213,9 +209,7 @@ class TestJavaApplication(TestApplicationJava):
}
)
assert (
resp['headers']['X-Session-Id'] != session_id
), 'session inactive'
assert resp['headers']['X-Session-Id'] != session_id, 'session inactive'
def test_java_application_session_invalidate(self):
self.load('session_invalidate')
@ -391,9 +385,7 @@ class TestJavaApplication(TestApplicationJava):
assert (
headers['X-Content-Type'] == 'text/plain;charset=utf-8'
), '#1 response Content-Type'
assert (
headers['X-Character-Encoding'] == 'utf-8'
), '#1 response charset'
assert headers['X-Character-Encoding'] == 'utf-8', '#1 response charset'
headers = self.get(url='/2')['headers']
@ -445,15 +437,11 @@ class TestJavaApplication(TestApplicationJava):
headers = self.get(url='/6')['headers']
assert (
'Content-Type' in headers
) == False, '#6 no Content-Type header'
assert ('Content-Type' in headers) == False, '#6 no Content-Type header'
assert (
'X-Content-Type' in headers
) == False, '#6 no response Content-Type'
assert (
headers['X-Character-Encoding'] == 'utf-8'
), '#6 response charset'
assert headers['X-Character-Encoding'] == 'utf-8', '#6 response charset'
headers = self.get(url='/7')['headers']
@ -463,9 +451,7 @@ class TestJavaApplication(TestApplicationJava):
assert (
headers['X-Content-Type'] == 'text/plain;charset=utf-8'
), '#7 response Content-Type'
assert (
headers['X-Character-Encoding'] == 'utf-8'
), '#7 response charset'
assert headers['X-Character-Encoding'] == 'utf-8', '#7 response charset'
headers = self.get(url='/8')['headers']
@ -475,9 +461,7 @@ class TestJavaApplication(TestApplicationJava):
assert (
headers['X-Content-Type'] == 'text/html;charset=utf-8'
), '#8 response Content-Type'
assert (
headers['X-Character-Encoding'] == 'utf-8'
), '#8 response charset'
assert headers['X-Character-Encoding'] == 'utf-8', '#8 response charset'
def test_java_application_welcome_files(self):
self.load('welcome_files')
@ -490,9 +474,7 @@ class TestJavaApplication(TestApplicationJava):
resp = self.get(url='/dir1/')
assert (
'This is index.txt.' in resp['body']
) == True, 'dir1 index body'
assert ('This is index.txt.' in resp['body']) == True, 'dir1 index body'
assert resp['headers']['X-TXT-Filter'] == '1', 'TXT Filter header'
headers = self.get(url='/dir2/')['headers']
@ -655,9 +637,7 @@ class TestJavaApplication(TestApplicationJava):
assert (
headers['X-FORWARD-Id'] == 'data'
), 'forward request servlet mapping'
assert (
headers['X-FORWARD-Request-URI'] == '/fwd'
), 'forward request uri'
assert headers['X-FORWARD-Request-URI'] == '/fwd', 'forward request uri'
assert (
headers['X-FORWARD-Servlet-Path'] == '/fwd'
), 'forward request servlet path'
@ -1003,9 +983,7 @@ class TestJavaApplication(TestApplicationJava):
)
assert resp['status'] == 200, 'multipart status'
assert re.search(
r'sample\.txt created', resp['body']
), 'multipart body'
assert re.search(r'sample\.txt created', resp['body']), 'multipart body'
assert (
self.search_in_log(
r'^Data from sample file$', name=reldst + '/sample.txt'

5
test/test_java_isolation_rootfs.py
View file

@ -11,7 +11,7 @@ class TestJavaIsolationRootfs(TestApplicationJava):
def setup_method(self, is_su):
if not is_su:
return
pytest.skip('require root')
os.makedirs(option.temp_dir + '/jars')
os.makedirs(option.temp_dir + '/tmp')
@ -61,7 +61,8 @@ class TestJavaIsolationRootfs(TestApplicationJava):
self.load('empty_war', isolation=isolation)
assert 'success' in self.conf(
'"/"', '/config/applications/empty_war/working_directory',
'"/"',
'/config/applications/empty_war/working_directory',
)
assert 'success' in self.conf(

86
test/test_java_websockets.py
View file

@ -869,9 +869,7 @@ 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)
@ -1116,7 +1114,7 @@ class TestJavaWebsockets(TestApplicationJava):
_, sock, _ = self.ws.upgrade()
self.ws.frame_write(sock, self.ws.OP_TEXT, 'BAsd7&jh23' * 26 * 2 ** 10)
self.ws.frame_write(sock, self.ws.OP_TEXT, 'BAsd7&jh23' * 26 * 2**10)
self.ws.frame_write(sock, self.ws.OP_TEXT, payload)
self.ws.frame_write(sock, self.ws.OP_CLOSE, self.ws.serialize_close())
@ -1278,62 +1276,62 @@ class TestJavaWebsockets(TestApplicationJava):
def check_message(opcode, f_size):
if opcode == self.ws.OP_TEXT:
payload = '*' * 4 * 2 ** 20
payload = '*' * 4 * 2**20
else:
payload = b'*' * 4 * 2 ** 20
payload = b'*' * 4 * 2**20
self.ws.message(sock, opcode, payload, fragmention_size=f_size)
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, 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, 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=1024) # 9_6_5
check_payload(op_binary, 2 ** 20, chopsize=2048) # 9_6_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=1024) # 9_6_5
check_payload(op_binary, 2**20, chopsize=2048) # 9_6_6
self.close_connection(sock)

4
test/test_node_application.py
View file

@ -218,9 +218,7 @@ 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')

86
test/test_node_websockets.py
View file

@ -888,9 +888,7 @@ 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)
@ -1135,7 +1133,7 @@ class TestNodeWebsockets(TestApplicationNode):
_, sock, _ = self.ws.upgrade()
self.ws.frame_write(sock, self.ws.OP_TEXT, 'BAsd7&jh23' * 26 * 2 ** 10)
self.ws.frame_write(sock, self.ws.OP_TEXT, 'BAsd7&jh23' * 26 * 2**10)
self.ws.frame_write(sock, self.ws.OP_TEXT, payload)
self.ws.frame_write(sock, self.ws.OP_CLOSE, self.ws.serialize_close())
@ -1297,62 +1295,62 @@ class TestNodeWebsockets(TestApplicationNode):
def check_message(opcode, f_size):
if opcode == self.ws.OP_TEXT:
payload = '*' * 4 * 2 ** 20
payload = '*' * 4 * 2**20
else:
payload = b'*' * 4 * 2 ** 20
payload = b'*' * 4 * 2**20
self.ws.message(sock, opcode, payload, fragmention_size=f_size)
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, 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, 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=1024) # 9_6_5
check_payload(op_binary, 2 ** 20, chopsize=2048) # 9_6_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=1024) # 9_6_5
check_payload(op_binary, 2**20, chopsize=2048) # 9_6_6
self.close_connection(sock)

3
test/test_perl_application.py
View file

@ -134,8 +134,7 @@ class TestPerlApplication(TestApplicationPerl):
assert self.get()['body'] == '1', 'errors result'
assert (
self.wait_for_record(r'\[error\].+Error in application')
is not None
self.wait_for_record(r'\[error\].+Error in application') is not None
), 'errors print'
def test_perl_application_header_equal_names(self):

11
test/test_php_application.py
View file

@ -419,9 +419,7 @@ opcache.preload_user = %(user)s
body = self.get()['body']
assert not re.search(
r'time: \d+', body
), 'disable_functions comma time'
assert not re.search(r'time: \d+', body), 'disable_functions comma time'
assert not re.search(
r'exec: \/\w+', body
), 'disable_functions comma exec'
@ -498,9 +496,7 @@ opcache.preload_user = %(user)s
body = self.get()['body']
assert not re.search(
r'time: \d+', body
), 'disable_functions space time'
assert not re.search(r'time: \d+', body), 'disable_functions space time'
assert not re.search(
r'exec: \/\w+', body
), 'disable_functions space exec'
@ -671,7 +667,8 @@ opcache.preload_user = %(user)s
assert resp['body'] == script_cwd, 'default cwd'
assert 'success' in self.conf(
'"' + option.test_dir + '"', 'applications/cwd/working_directory',
'"' + option.test_dir + '"',
'applications/cwd/working_directory',
)
resp = self.get()

9
test/test_proxy.py
View file

@ -75,8 +75,7 @@ Content-Length: 10
"type": self.get_application_type(),
"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",
},
"custom_header": {
@ -126,8 +125,7 @@ Content-Length: 10
"type": self.get_application_type(),
"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",
}
},
@ -502,8 +500,7 @@ Content-Length: 10
"type": self.get_application_type(),
"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",
},
},

14
test/test_proxy_chunked.py
View file

@ -90,12 +90,13 @@ class TestProxyChunked(TestApplicationPython):
assert 'success' in self.conf(
{
"listeners": {"*:7080": {"pass": "routes"},},
"listeners": {
"*:7080": {"pass": "routes"},
},
"routes": [
{
"action": {
"proxy": "http://127.0.0.1:"
+ str(self.SERVER_PORT)
"proxy": "http://127.0.0.1:" + str(self.SERVER_PORT)
}
}
],
@ -166,9 +167,7 @@ class TestProxyChunked(TestApplicationPython):
assert (
self.get_http10(
body=self.chunks(
[('1', hex(i % 16)[2:]) for i in range(4096)]
),
body=self.chunks([('1', hex(i % 16)[2:]) for i in range(4096)]),
)['body']
== part * 256
)
@ -210,8 +209,7 @@ class TestProxyChunked(TestApplicationPython):
assert resp['body'][-5:] != '0\r\n\r\n', 'no zero chunk'
assert (
self.get_http10(body='\r\n\r\n80000000\r\nA X 100')['status']
== 200
self.get_http10(body='\r\n\r\n80000000\r\nA X 100')['status'] == 200
)
assert (
self.get_http10(body='\r\n\r\n10000000000000000\r\nA X 100')[

4
test/test_python_application.py
View file

@ -735,9 +735,7 @@ last line: 987654321
'nobody uid user=nobody group=%s' % group
)
assert obj['GID'] == group_id, (
'nobody gid user=nobody group=%s' % group
)
assert obj['GID'] == group_id, 'nobody gid user=nobody group=%s' % group
self.load('user_group', group=group)

7
test/test_python_isolation.py
View file

@ -56,9 +56,7 @@ class TestPythonIsolation(TestApplicationPython):
ret = self.getjson(url='/?path=/app/python/ns_inspect')
assert (
ret['body']['FileExists'] == True
), 'application exists in rootfs'
assert ret['body']['FileExists'] == True, 'application exists in rootfs'
def test_python_isolation_rootfs_no_language_deps(self, is_su, temp_dir):
if not is_su:
@ -93,8 +91,7 @@ class TestPythonIsolation(TestApplicationPython):
self.load('ns_inspect', isolation=isolation)
assert (
self.getjson(url='/?path=/proc/self')['body']['FileExists']
== False
self.getjson(url='/?path=/proc/self')['body']['FileExists'] == False
), 'no /proc/self'
isolation['automount']['procfs'] = True

4
test/test_python_isolation_chroot.py
View file

@ -35,6 +35,4 @@ class TestPythonIsolation(TestApplicationPython):
ret = self.getjson(url='/?path=/app/python/ns_inspect')
assert (
ret['body']['FileExists'] == True
), 'application exists in rootfs'
assert ret['body']['FileExists'] == True, 'application exists in rootfs'

3
test/test_respawn.py
View file

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

41
test/test_routing.py
View file

@ -12,7 +12,10 @@ class TestRouting(TestApplicationPython):
{
"listeners": {"*:7080": {"pass": "routes"}},
"routes": [
{"match": {"method": "GET"}, "action": {"return": 200},}
{
"match": {"method": "GET"},
"action": {"return": 200},
}
],
"applications": {},
}
@ -490,11 +493,15 @@ class TestRouting(TestApplicationPython):
'routes/0/action',
), 'proxy share'
assert 'error' in self.conf(
{"proxy": "http://127.0.0.1:7081", "pass": "applications/app",},
{
"proxy": "http://127.0.0.1:7081",
"pass": "applications/app",
},
'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):
@ -693,7 +700,8 @@ class TestRouting(TestApplicationPython):
assert self.post()['status'] == 404, 'routes edit POST'
assert 'success' in self.conf_post(
{"match": {"method": "POST"}, "action": {"return": 200}}, 'routes',
{"match": {"method": "POST"}, "action": {"return": 200}},
'routes',
), 'routes edit configure 2'
assert 'GET' == self.conf_get(
'routes/0/match/method'
@ -733,7 +741,8 @@ class TestRouting(TestApplicationPython):
assert self.post()['status'] == 404, 'routes edit POST 5'
assert 'success' in self.conf_post(
{"match": {"method": "POST"}, "action": {"return": 200}}, 'routes',
{"match": {"method": "POST"}, "action": {"return": 200}},
'routes',
), 'routes edit configure 6'
assert self.get()['status'] == 404, 'routes edit GET 6'
@ -1042,9 +1051,7 @@ class TestRouting(TestApplicationPython):
def check_headers(hds):
hds = dict({"Host": "localhost", "Connection": "close"}, **hds)
assert (
self.get(headers=hds)['status'] == 200
), 'headers array match'
assert self.get(headers=hds)['status'] == 200, 'headers array match'
def check_headers_404(hds):
hds = dict({"Host": "localhost", "Connection": "close"}, **hds)
@ -1262,9 +1269,7 @@ class TestRouting(TestApplicationPython):
self.get(url='/?foo=bar&blah=test')['status'] == 200
), 'multiple 2'
assert self.get(url='/?foo=bar&blah')['status'] == 404, 'multiple 3'
assert (
self.get(url='/?foo=bar&blah=tes')['status'] == 404
), 'multiple 4'
assert self.get(url='/?foo=bar&blah=tes')['status'] == 404, 'multiple 4'
assert (
self.get(url='/?foo=b%61r&bl%61h=t%65st')['status'] == 200
), 'multiple 5'
@ -1494,9 +1499,7 @@ class TestRouting(TestApplicationPython):
sock, port = sock_port()
sock2, port2 = sock_port()
self.route_match(
{"source": "127.0.0.1:" + str(port) + "-" + str(port)}
)
self.route_match({"source": "127.0.0.1:" + str(port) + "-" + str(port)})
assert self.get(sock=sock)['status'] == 200, 'range single'
assert self.get(sock=sock2)['status'] == 404, 'range single 2'
@ -1544,7 +1547,10 @@ class TestRouting(TestApplicationPython):
def test_routes_source_addr(self):
assert 'success' in self.conf(
{"*:7080": {"pass": "routes"}, "[::1]:7081": {"pass": "routes"},},
{
"*:7080": {"pass": "routes"},
"[::1]:7081": {"pass": "routes"},
},
'listeners',
), 'source listeners configure'
@ -1650,7 +1656,10 @@ class TestRouting(TestApplicationPython):
def test_routes_source_cidr(self):
assert 'success' in self.conf(
{"*:7080": {"pass": "routes"}, "[::1]:7081": {"pass": "routes"},},
{
"*:7080": {"pass": "routes"},
"[::1]:7081": {"pass": "routes"},
},
'listeners',
), 'source listeners configure'

9
test/test_ruby_application.py
View file

@ -175,8 +175,7 @@ class TestRubyApplication(TestApplicationRuby):
self.get()
assert (
self.wait_for_record(r'\[error\].+Error in application')
is not None
self.wait_for_record(r'\[error\].+Error in application') is not None
), 'errors puts'
def test_ruby_application_errors_puts_int(self):
@ -194,8 +193,7 @@ class TestRubyApplication(TestApplicationRuby):
self.get()
assert (
self.wait_for_record(r'\[error\].+Error in application')
is not None
self.wait_for_record(r'\[error\].+Error in application') is not None
), 'errors write'
def test_ruby_application_errors_write_to_s_custom(self):
@ -229,7 +227,8 @@ class TestRubyApplication(TestApplicationRuby):
try:
locales = (
subprocess.check_output(
['locale', '-a'], stderr=subprocess.STDOUT,
['locale', '-a'],
stderr=subprocess.STDOUT,
)
.decode()
.split('\n')

5
test/test_ruby_hooks.py
View file

@ -81,7 +81,10 @@ class TestRubyHooks(TestApplicationRuby):
threads = 1
self.load(
'hooks', processes=processes, threads=threads, hooks='multiple.rb',
'hooks',
processes=processes,
threads=threads,
hooks='multiple.rb',
)
hooked = self._wait_cookie('worker_boot.*', processes)

6
test/test_ruby_isolation.py
View file

@ -34,11 +34,13 @@ class TestRubyIsolation(TestApplicationRuby):
self.load('status_int', isolation=isolation)
assert 'success' in self.conf(
'"/ruby/status_int/config.ru"', 'applications/status_int/script',
'"/ruby/status_int/config.ru"',
'applications/status_int/script',
)
assert 'success' in self.conf(
'"/ruby/status_int"', 'applications/status_int/working_directory',
'"/ruby/status_int"',
'applications/status_int/working_directory',
)
assert self.get()['status'] == 200, 'status int'

20
test/test_settings.py
View file

@ -207,9 +207,7 @@ Connection: close
{"unix:" + addr: {'application': 'body_generate'}}, 'listeners'
)
assert 'success' in self.conf(
{'http': {'send_timeout': 1}}, 'settings'
)
assert 'success' in self.conf({'http': {'send_timeout': 1}}, 'settings')
data = req(addr, data_len)
assert re.search(r'200 OK', data), 'send timeout status'
@ -237,14 +235,10 @@ Connection: close
assert self.get()['status'] == 200, 'init'
assert 'success' in self.conf(
{'http': {'idle_timeout': 2}}, 'settings'
)
assert 'success' in self.conf({'http': {'idle_timeout': 2}}, 'settings')
assert req()['status'] == 408, 'status idle timeout'
assert 'success' in self.conf(
{'http': {'idle_timeout': 7}}, 'settings'
)
assert 'success' in self.conf({'http': {'idle_timeout': 7}}, 'settings')
assert req()['status'] == 200, 'status idle timeout 2'
def test_settings_idle_timeout_2(self):
@ -259,14 +253,10 @@ Connection: close
assert self.get()['status'] == 200, 'init'
assert 'success' in self.conf(
{'http': {'idle_timeout': 1}}, 'settings'
)
assert 'success' in self.conf({'http': {'idle_timeout': 1}}, 'settings')
assert req()['status'] == 408, 'status idle timeout'
assert 'success' in self.conf(
{'http': {'idle_timeout': 7}}, 'settings'
)
assert 'success' in self.conf({'http': {'idle_timeout': 7}}, 'settings')
assert req()['status'] == 200, 'status idle timeout 2'
def test_settings_max_body_size(self):

28
test/test_static.py
View file

@ -133,7 +133,8 @@ 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'
@ -154,9 +155,7 @@ class TestStatic(TestApplicationProto):
assert (
self.get(url='/ di r %2Ffi le')['body'] == 'blah'
), 'slash encoded'
assert (
self.get(url='/ di r /fi%20le')['body'] == 'blah'
), 'file encoded'
assert self.get(url='/ di r /fi%20le')['body'] == 'blah', 'file encoded'
assert (
self.get(url='/%20di%20r%20%2Ffi%20le')['body'] == 'blah'
), 'encoded'
@ -195,7 +194,8 @@ 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 (
@ -266,13 +266,14 @@ class TestStatic(TestApplicationProto):
self.get(url='/')['headers']['Content-Type'] == 'text/plain'
), 'mime_types index default'
assert (
self.get(url='/dir/file')['headers']['Content-Type']
== 'text/plain'
self.get(url='/dir/file')['headers']['Content-Type'] == 'text/plain'
), 'mime_types file in dir'
def test_static_mime_types_partial_match(self):
assert 'success' in self.conf(
{"text/x-blah": ["ile", "fil", "f", "e", ".file"],},
{
"text/x-blah": ["ile", "fil", "f", "e", ".file"],
},
'settings/http/static/mime_types',
), 'configure mime_types'
assert 'Content-Type' not in self.get(url='/dir/file'), 'partial match'
@ -313,16 +314,14 @@ class TestStatic(TestApplicationProto):
'"file"', 'settings/http/static/mime_types/text%2Fplain'
), 'mime_types add array element'
assert (
self.get(url='/dir/file')['headers']['Content-Type']
== 'text/plain'
self.get(url='/dir/file')['headers']['Content-Type'] == 'text/plain'
), 'mime_types reverted'
assert 'success' in self.conf(
'"file"', 'settings/http/static/mime_types/text%2Fplain'
), 'configure mime_types update'
assert (
self.get(url='/dir/file')['headers']['Content-Type']
== 'text/plain'
self.get(url='/dir/file')['headers']['Content-Type'] == 'text/plain'
), 'mime_types updated'
assert (
'Content-Type' not in self.get(url='/log.log')['headers']
@ -346,7 +345,10 @@ class TestStatic(TestApplicationProto):
'settings/http/static/mime_types',
), 'mime_types same extensions array'
assert 'error' in self.conf(
{"text/x-code": [".h", ".c", "readme"], "text/plain": "README",},
{
"text/x-code": [".h", ".c", "readme"],
"text/plain": "README",
},
'settings/http/static/mime_types',
), 'mime_types same extensions case insensitive'

27
test/test_static_chroot.py
View file

@ -26,13 +26,14 @@ class TestStaticChroot(TestApplicationProto):
def update_action(self, share, chroot):
return self.conf(
{"share": share, "chroot": chroot}, 'routes/0/action',
{"share": share, "chroot": chroot},
'routes/0/action',
)
def get_custom(self, uri, host):
return self.get(
url=uri, headers={'Host': host, 'Connection': 'close'}
)['status']
return self.get(url=uri, headers={'Host': host, 'Connection': 'close'})[
'status'
]
def test_static_chroot(self, temp_dir):
assert self.get(url='/dir/file')['status'] == 200, 'default chroot'
@ -101,7 +102,8 @@ class TestStaticChroot(TestApplicationProto):
), 'chroot empty absolute'
assert 'success' in self.conf(
{"share": ".$uri", "chroot": ""}, 'routes/0/action',
{"share": ".$uri", "chroot": ""},
'routes/0/action',
), 'configure chroot empty relative'
assert (
@ -120,13 +122,15 @@ class TestStaticChroot(TestApplicationProto):
assert self.get(url='/dir/file')['status'] == 403, 'relative chroot'
assert 'success' in self.conf(
{"share": ".$uri"}, 'routes/0/action',
{"share": ".$uri"},
'routes/0/action',
), 'configure relative share'
assert self.get(url=self.test_path)['status'] == 200, 'relative share'
assert 'success' in self.conf(
{"share": ".$uri", "chroot": "."}, 'routes/0/action',
{"share": ".$uri", "chroot": "."},
'routes/0/action',
), 'configure relative'
assert self.get(url=self.test_path)['status'] == 200, 'relative'
@ -208,13 +212,16 @@ class TestStaticChroot(TestApplicationProto):
def test_static_chroot_invalid(self, temp_dir):
assert 'error' in self.conf(
{"share": temp_dir, "chroot": True}, 'routes/0/action',
{"share": temp_dir, "chroot": True},
'routes/0/action',
), 'configure chroot error'
assert 'error' in self.conf(
{"share": temp_dir, "symlinks": "True"}, 'routes/0/action',
{"share": temp_dir, "symlinks": "True"},
'routes/0/action',
), 'configure symlink error'
assert 'error' in self.conf(
{"share": temp_dir, "mount": "True"}, 'routes/0/action',
{"share": temp_dir, "mount": "True"},
'routes/0/action',
), 'configure mount error'
assert 'error' in self.update_action(

4
test/test_static_mount.py
View file

@ -41,9 +41,7 @@ class TestStaticMount(TestApplicationProto):
self._load_conf(
{
"listeners": {"*:7080": {"pass": "routes"}},
"routes": [
{"action": {"share": temp_dir + "/assets/dir$uri"}}
],
"routes": [{"action": {"share": temp_dir + "/assets/dir$uri"}}],
}
)

4
test/test_tls.py
View file

@ -358,9 +358,7 @@ basicConstraints = critical,CA:TRUE"""
self.add_tls(cert='int')
assert (
self.get_ssl()['status'] == 200
), 'certificate chain intermediate'
assert self.get_ssl()['status'] == 200, 'certificate chain intermediate'
# intermediate server

7
test/test_tls_sni.py
View file

@ -178,7 +178,8 @@ basicConstraints = critical,CA:TRUE"""
self.add_tls(["localhost.com", "example.com"])
resp, sock = self.get_ssl(
headers={'Content-Length': '0', 'Connection': 'close'}, start=True,
headers={'Content-Length': '0', 'Connection': 'close'},
start=True,
)
assert resp['status'] == 200
assert (
@ -272,9 +273,7 @@ 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": ''}})

7
test/test_tls_tickets.py
View file

@ -17,9 +17,7 @@ class TestTLSTicket(TestApplicationTLS):
prerequisites = {'modules': {'openssl': 'any'}}
ticket = 'U1oDTh11mMxODuw12gS0EXX1E/PkZG13cJNQ6m5+6BGlfPTjNlIEw7PSVU3X1gTE'
ticket2 = (
'5AV0DSYIYbZWZQB7fCnTHZmMxtotb/aXjam+n2XS79lTvX3Tq9xGqpC8XKNEF2lt'
)
ticket2 = '5AV0DSYIYbZWZQB7fCnTHZmMxtotb/aXjam+n2XS79lTvX3Tq9xGqpC8XKNEF2lt'
ticket80 = '6Pfil8lv/k8zf8MndPpfXaO5EAV6dhME6zs6CfUyq2yziynQwSywtKQMqHGnJ2HR\
49TZXi/Y4/8RSIO7QPsU51/HLR1gWIMhVM2m9yh93Bw='
@ -182,7 +180,8 @@ class TestTLSTicket(TestApplicationTLS):
def test_tls_ticket_invalid(self):
def check_tickets(tickets):
assert 'error' in self.conf(
{"tickets": tickets}, 'listeners/*:7080/tls/session',
{"tickets": tickets},
'listeners/*:7080/tls/session',
)
check_tickets({})

3
test/test_upstreams_rr.py
View file

@ -342,7 +342,8 @@ Connection: close
assert self.get()['body'] == ''
assert 'success' in self.conf(
{"127.0.0.1:7083": {"weight": 2}}, 'upstreams/one/servers',
{"127.0.0.1:7083": {"weight": 2}},
'upstreams/one/servers',
), 'active req new server'
assert 'success' in self.conf_delete(
'upstreams/one/servers/127.0.0.1:7083'

4
test/test_usr1.py
View file

@ -59,9 +59,7 @@ class TestUSR1(TestApplicationPython):
body = 'body_for_a_log_new\n'
assert self.post(body=body)['status'] == 200
assert (
self.wait_for_record(body, log_new) is not None
), 'rename new'
assert self.wait_for_record(body, log_new) is not None, 'rename new'
assert not os.path.isfile(log_path), 'rename old'
os.kill(unit_pid, signal.SIGUSR1)

7
test/unit/applications/lang/go.py
View file

@ -24,8 +24,8 @@ class TestApplicationGo(TestApplicationProto):
env['GOCACHE'] = cache_dir
shutil.copy2(
option.test_dir + '/go/' + script + '/' + name + '.go',
temp_dir)
option.test_dir + '/go/' + script + '/' + name + '.go', temp_dir
)
if static:
args = [
@ -55,7 +55,8 @@ class TestApplicationGo(TestApplicationProto):
f"""module test/app
require unit.nginx.org/go v0.0.0
replace unit.nginx.org/go => {replace_path}
""")
"""
)
if option.detailed:
print("\n$ GOPATH=" + env['GOPATH'] + " " + " ".join(args))

10
test/unit/applications/websockets.py
View file

@ -43,7 +43,11 @@ 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:
@ -218,9 +222,7 @@ class TestApplicationWebsocket(TestApplicationProto):
while pos < message_len:
end = min(pos + fragmention_size, message_len)
fin = end == message_len
self.frame_write(
sock, op_code, message[pos:end], fin=fin, **kwargs
)
self.frame_write(sock, op_code, message[pos:end], fin=fin, **kwargs)
op_code = self.OP_CONT
pos = end

4
test/unit/control.py
View file

@ -30,10 +30,6 @@ def args_handler(conf_func):
class TestControl(TestHTTP):
# TODO socket reuse
# TODO http client
@args_handler
def conf(self, conf, url):
return self.put(**self._get_args(url, conf))['body']

4
test/unit/http.py
View file

@ -328,9 +328,7 @@ 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()