Tests: removed unused variables.

This commit is contained in:
Andrei Zeliankou
2023-05-25 14:26:12 +01:00
parent 47cdfb6f30
commit 3e4fa1e202
34 changed files with 87 additions and 91 deletions

View File

@@ -175,7 +175,7 @@ Unexpected prerequisite version "{prereq_version}" for module "{module}" in
) )
def pytest_sessionstart(session): def pytest_sessionstart():
option.available = {'modules': {}, 'features': {}} option.available = {'modules': {}, 'features': {}}
unit = unit_run() unit = unit_run()
@@ -185,7 +185,7 @@ def pytest_sessionstart(session):
# read unit.log # read unit.log
for i in range(50): for _ in range(50):
with open(Log.get_path(), 'r') as f: with open(Log.get_path(), 'r') as f:
log = f.read() log = f.read()
m = re.search('controller started', log) m = re.search('controller started', log)
@@ -237,7 +237,7 @@ def pytest_sessionstart(session):
@pytest.hookimpl(tryfirst=True, hookwrapper=True) @pytest.hookimpl(tryfirst=True, hookwrapper=True)
def pytest_runtest_makereport(item, call): def pytest_runtest_makereport(item):
# execute all other hooks to obtain the report object # execute all other hooks to obtain the report object
outcome = yield outcome = yield
rep = outcome.get_result() rep = outcome.get_result()
@@ -567,7 +567,7 @@ def _clear_temp_dir():
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) os.remove(path)
else: else:
for attempt in range(10): for _ in range(10):
try: try:
shutil.rmtree(path) shutil.rmtree(path)
break break
@@ -582,7 +582,7 @@ def _check_processes():
controller_pid = _fds_info['controller']['pid'] controller_pid = _fds_info['controller']['pid']
unit_pid = unit_instance['pid'] unit_pid = unit_instance['pid']
for i in range(600): for _ in range(600):
out = ( out = (
subprocess.check_output( subprocess.check_output(
['ps', '-ax', '-o', 'pid', '-o', 'ppid', '-o', 'command'] ['ps', '-ax', '-o', 'pid', '-o', 'ppid', '-o', 'command']
@@ -625,7 +625,7 @@ def _check_processes():
@print_log_on_assert @print_log_on_assert
def _check_fds(*, log=None): def _check_fds(*, log=None):
def waitforfds(diff): def waitforfds(diff):
for i in range(600): for _ in range(600):
fds_diff = diff() fds_diff = diff()
if fds_diff <= option.fds_threshold: if fds_diff <= option.fds_threshold:
@@ -748,7 +748,7 @@ def skip_fds_check():
@pytest.fixture @pytest.fixture
def temp_dir(request): def temp_dir():
return unit_instance['temp_dir'] return unit_instance['temp_dir']
@@ -758,16 +758,16 @@ def is_unsafe(request):
@pytest.fixture @pytest.fixture
def is_su(request): def is_su():
return os.geteuid() == 0 return os.geteuid() == 0
@pytest.fixture @pytest.fixture
def unit_pid(request): def unit_pid():
return unit_instance['process'].pid return unit_instance['process'].pid
def pytest_sessionfinish(session): def pytest_sessionfinish():
if not option.restart and option.save_log: if not option.restart and option.save_log:
print(f'Path to unit.log:\n{Log.get_path()}\n') print(f'Path to unit.log:\n{Log.get_path()}\n')

View File

@@ -32,7 +32,7 @@ class TestAccessLog(TestApplicationPython):
assert self.get()['status'] == 200, 'init' assert self.get()['status'] == 200, 'init'
(resp, sock) = self.post( (_, sock) = self.post(
headers={ headers={
'Host': 'localhost', 'Host': 'localhost',
'Connection': 'keep-alive', 'Connection': 'keep-alive',
@@ -46,7 +46,7 @@ class TestAccessLog(TestApplicationPython):
self.wait_for_record(r'"POST / HTTP/1.1" 200 5') is not None self.wait_for_record(r'"POST / HTTP/1.1" 200 5') is not None
), 'keepalive 1' ), 'keepalive 1'
resp = self.post(sock=sock, body='0123456789') _ = self.post(sock=sock, body='0123456789')
assert ( assert (
self.wait_for_record(r'"POST / HTTP/1.1" 200 10') is not None self.wait_for_record(r'"POST / HTTP/1.1" 200 10') is not None
@@ -169,7 +169,7 @@ Connection: close
assert self.post()['status'] == 200, 'init' assert self.post()['status'] == 200, 'init'
resp = self.http(b"""GE""", raw=True, read_timeout=1) _ = self.http(b"""GE""", raw=True, read_timeout=1)
time.sleep(1) time.sleep(1)
@@ -193,7 +193,7 @@ Connection: close
assert self.post()['status'] == 200, 'init' assert self.post()['status'] == 200, 'init'
resp = self.http(b"""GET / HTTP/1.1""", raw=True, read_timeout=1) _ = self.http(b"""GET / HTTP/1.1""", raw=True, read_timeout=1)
time.sleep(1) time.sleep(1)
@@ -206,7 +206,7 @@ Connection: close
assert self.post()['status'] == 200, 'init' assert self.post()['status'] == 200, 'init'
resp = self.http(b"""GET / HTTP/1.1\n""", raw=True, read_timeout=1) _ = self.http(b"""GET / HTTP/1.1\n""", raw=True, read_timeout=1)
time.sleep(1) time.sleep(1)

View File

@@ -59,7 +59,7 @@ custom-header: BLAH
}, 'headers' }, 'headers'
assert resp['body'] == body, 'body' assert resp['body'] == body, 'body'
def test_asgi_application_ipv6(self, temp_dir): def test_asgi_application_ipv6(self):
self.load('empty') self.load('empty')
assert 'success' in self.conf( assert 'success' in self.conf(
@@ -401,7 +401,7 @@ Connection: close
socks = [] socks = []
for i in range(2): for _ in range(2):
sock = self.get( sock = self.get(
headers={ headers={
'Host': 'localhost', 'Host': 'localhost',

View File

@@ -19,7 +19,7 @@ class TestASGIWebsockets(TestApplicationPython):
ws = TestApplicationWebsocket() ws = TestApplicationWebsocket()
@pytest.fixture(autouse=True) @pytest.fixture(autouse=True)
def setup_method_fixture(self, request, skip_alert): def setup_method_fixture(self, skip_alert):
assert 'success' in self.conf( assert 'success' in self.conf(
{'http': {'websocket': {'keepalive_interval': 0}}}, 'settings' {'http': {'websocket': {'keepalive_interval': 0}}}, 'settings'
), 'clear keepalive_interval' ), 'clear keepalive_interval'
@@ -73,7 +73,7 @@ class TestASGIWebsockets(TestApplicationPython):
def test_asgi_websockets_subprotocol(self): def test_asgi_websockets_subprotocol(self):
self.load('websockets/subprotocol') self.load('websockets/subprotocol')
resp, sock, key = self.ws.upgrade() resp, sock, _ = self.ws.upgrade()
sock.close() sock.close()
assert resp['status'] == 101, 'status' assert resp['status'] == 101, 'status'
@@ -951,7 +951,7 @@ class TestASGIWebsockets(TestApplicationPython):
_, sock, _ = self.ws.upgrade() _, sock, _ = self.ws.upgrade()
for i in range(0, 2): for _ in range(0, 2):
self.ws.frame_write(sock, self.ws.OP_CONT, 'fragment1', fin=False) self.ws.frame_write(sock, self.ws.OP_CONT, 'fragment1', fin=False)
self.ws.frame_write(sock, self.ws.OP_TEXT, 'fragment2', fin=False) self.ws.frame_write(sock, self.ws.OP_TEXT, 'fragment2', fin=False)
self.ws.frame_write(sock, self.ws.OP_CONT, 'fragment3', fin=True) self.ws.frame_write(sock, self.ws.OP_CONT, 'fragment3', fin=True)
@@ -961,7 +961,7 @@ class TestASGIWebsockets(TestApplicationPython):
_, sock, _ = self.ws.upgrade() _, sock, _ = self.ws.upgrade()
for i in range(0, 2): for _ in range(0, 2):
self.ws.frame_write(sock, self.ws.OP_CONT, 'fragment1', fin=True) self.ws.frame_write(sock, self.ws.OP_CONT, 'fragment1', fin=True)
self.ws.frame_write(sock, self.ws.OP_TEXT, 'fragment2', fin=False) self.ws.frame_write(sock, self.ws.OP_TEXT, 'fragment2', fin=False)
self.ws.frame_write(sock, self.ws.OP_CONT, 'fragment3', fin=True) self.ws.frame_write(sock, self.ws.OP_CONT, 'fragment3', fin=True)

View File

@@ -96,7 +96,7 @@ class TestClientIP(TestApplicationPython):
]: ]:
assert self.get_xff(ip, 'ipv6') == ip, 'replace' assert self.get_xff(ip, 'ipv6') == ip, 'replace'
def test_client_ip_unix(self, temp_dir): def test_client_ip_unix(self):
self.client_ip({'header': 'X-Forwarded-For', 'source': 'unix'}) self.client_ip({'header': 'X-Forwarded-For', 'source': 'unix'})
assert self.get_xff('1.1.1.1') == '127.0.0.1', 'bad source ipv4' assert self.get_xff('1.1.1.1') == '127.0.0.1', 'bad source ipv4'

View File

@@ -25,7 +25,7 @@ class TestConfiguration(TestControl):
def test_json_unicode(self): def test_json_unicode(self):
assert 'success' in self.conf( assert 'success' in self.conf(
u""" """
{ {
"ap\u0070": { "ap\u0070": {
"type": "\u0070ython", "type": "\u0070ython",
@@ -64,7 +64,7 @@ class TestConfiguration(TestControl):
def test_json_unicode_number(self): def test_json_unicode_number(self):
assert 'success' in self.conf( assert 'success' in self.conf(
u""" """
{ {
"app": { "app": {
"type": "python", "type": "python",
@@ -252,7 +252,7 @@ class TestConfiguration(TestControl):
) )
def test_listeners_port_release(self): def test_listeners_port_release(self):
for i in range(10): for _ in range(10):
fail = False fail = False
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s: with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)

View File

@@ -8,7 +8,7 @@ class TestGoApplication(TestApplicationGo):
prerequisites = {'modules': {'go': 'all'}} prerequisites = {'modules': {'go': 'all'}}
@pytest.fixture(autouse=True) @pytest.fixture(autouse=True)
def setup_method_fixture(self, request, skip_alert): def setup_method_fixture(self, skip_alert):
skip_alert(r'\[unit\] close\(\d+\) failed: Bad file descriptor') skip_alert(r'\[unit\] close\(\d+\) failed: Bad file descriptor')
def test_go_application_variables(self): def test_go_application_variables(self):

View File

@@ -12,7 +12,7 @@ class TestGoIsolation(TestApplicationGo):
prerequisites = {'modules': {'go': 'any'}, 'features': ['isolation']} prerequisites = {'modules': {'go': 'any'}, 'features': ['isolation']}
@pytest.fixture(autouse=True) @pytest.fixture(autouse=True)
def setup_method_fixture(self, request, skip_alert): def setup_method_fixture(self, skip_alert):
skip_alert(r'\[unit\] close\(\d+\) failed: Bad file descriptor') skip_alert(r'\[unit\] close\(\d+\) failed: Bad file descriptor')
def unpriv_creds(self): def unpriv_creds(self):

View File

@@ -8,7 +8,7 @@ class TestGoIsolationRootfs(TestApplicationGo):
prerequisites = {'modules': {'go': 'all'}} prerequisites = {'modules': {'go': 'all'}}
@pytest.fixture(autouse=True) @pytest.fixture(autouse=True)
def setup_method_fixture(self, request, skip_alert): def setup_method_fixture(self, skip_alert):
skip_alert(r'\[unit\] close\(\d+\) failed: Bad file descriptor') skip_alert(r'\[unit\] close\(\d+\) failed: Bad file descriptor')
def test_go_isolation_rootfs_chroot(self, is_su, temp_dir): def test_go_isolation_rootfs_chroot(self, is_su, temp_dir):

View File

@@ -1003,7 +1003,7 @@ class TestJavaApplication(TestApplicationJava):
socks = [] socks = []
for i in range(4): for _ in range(4):
sock = self.get( sock = self.get(
headers={ headers={
'Host': 'localhost', 'Host': 'localhost',

View File

@@ -13,7 +13,7 @@ class TestJavaWebsockets(TestApplicationJava):
ws = TestApplicationWebsocket() ws = TestApplicationWebsocket()
@pytest.fixture(autouse=True) @pytest.fixture(autouse=True)
def setup_method_fixture(self, request, skip_alert): def setup_method_fixture(self, skip_alert):
assert 'success' in self.conf( assert 'success' in self.conf(
{'http': {'websocket': {'keepalive_interval': 0}}}, 'settings' {'http': {'websocket': {'keepalive_interval': 0}}}, 'settings'
), 'clear keepalive_interval' ), 'clear keepalive_interval'
@@ -878,7 +878,7 @@ class TestJavaWebsockets(TestApplicationJava):
_, sock, _ = self.ws.upgrade() _, sock, _ = self.ws.upgrade()
for i in range(0, 2): for _ in range(0, 2):
self.ws.frame_write(sock, self.ws.OP_CONT, 'fragment1', fin=False) self.ws.frame_write(sock, self.ws.OP_CONT, 'fragment1', fin=False)
self.ws.frame_write(sock, self.ws.OP_TEXT, 'fragment2', fin=False) self.ws.frame_write(sock, self.ws.OP_TEXT, 'fragment2', fin=False)
self.ws.frame_write(sock, self.ws.OP_CONT, 'fragment3', fin=True) self.ws.frame_write(sock, self.ws.OP_CONT, 'fragment3', fin=True)
@@ -888,7 +888,7 @@ class TestJavaWebsockets(TestApplicationJava):
_, sock, _ = self.ws.upgrade() _, sock, _ = self.ws.upgrade()
for i in range(0, 2): for _ in range(0, 2):
self.ws.frame_write(sock, self.ws.OP_CONT, 'fragment1', fin=True) self.ws.frame_write(sock, self.ws.OP_CONT, 'fragment1', fin=True)
self.ws.frame_write(sock, self.ws.OP_TEXT, 'fragment2', fin=False) self.ws.frame_write(sock, self.ws.OP_TEXT, 'fragment2', fin=False)
self.ws.frame_write(sock, self.ws.OP_CONT, 'fragment3', fin=True) self.ws.frame_write(sock, self.ws.OP_CONT, 'fragment3', fin=True)

View File

@@ -42,7 +42,7 @@ class TestNJS(TestApplicationProto):
self.set_share(f'"{temp_dir}/assets/`string`"') self.set_share(f'"{temp_dir}/assets/`string`"')
assert self.get()['status'] == 200 assert self.get()['status'] == 200
def test_njs_template_expression(self, temp_dir): def test_njs_template_expression(self):
self.create_files('str', 'localhost') self.create_files('str', 'localhost')
self.check_expression('${uri}', '/str') self.check_expression('${uri}', '/str')
@@ -50,7 +50,7 @@ class TestNJS(TestApplicationProto):
self.check_expression('${uri + host}') self.check_expression('${uri + host}')
self.check_expression('${uri + `${host}`}') self.check_expression('${uri + `${host}`}')
def test_njs_iteration(self, temp_dir): def test_njs_iteration(self):
self.create_files('Connection,Host', 'close,localhost') self.create_files('Connection,Host', 'close,localhost')
self.check_expression('/${Object.keys(headers).sort().join()}') self.check_expression('/${Object.keys(headers).sort().join()}')
@@ -74,7 +74,7 @@ class TestNJS(TestApplicationProto):
self.set_share(f'"`{temp_dir}/assets/${{args.foo}}`"') self.set_share(f'"`{temp_dir}/assets/${{args.foo}}`"')
assert self.get(url='/?foo=str')['status'] == 200, 'args' assert self.get(url='/?foo=str')['status'] == 200, 'args'
def test_njs_invalid(self, temp_dir, skip_alert): def test_njs_invalid(self, skip_alert):
skip_alert(r'js exception:') skip_alert(r'js exception:')
def check_invalid(template): def check_invalid(template):

View File

@@ -13,7 +13,7 @@ class TestNodeWebsockets(TestApplicationNode):
ws = TestApplicationWebsocket() ws = TestApplicationWebsocket()
@pytest.fixture(autouse=True) @pytest.fixture(autouse=True)
def setup_method_fixture(self, request, skip_alert): def setup_method_fixture(self, skip_alert):
assert 'success' in self.conf( assert 'success' in self.conf(
{'http': {'websocket': {'keepalive_interval': 0}}}, 'settings' {'http': {'websocket': {'keepalive_interval': 0}}}, 'settings'
), 'clear keepalive_interval' ), 'clear keepalive_interval'
@@ -897,7 +897,7 @@ class TestNodeWebsockets(TestApplicationNode):
_, sock, _ = self.ws.upgrade() _, sock, _ = self.ws.upgrade()
for i in range(0, 2): for _ in range(0, 2):
self.ws.frame_write(sock, self.ws.OP_CONT, 'fragment1', fin=False) self.ws.frame_write(sock, self.ws.OP_CONT, 'fragment1', fin=False)
self.ws.frame_write(sock, self.ws.OP_TEXT, 'fragment2', fin=False) self.ws.frame_write(sock, self.ws.OP_TEXT, 'fragment2', fin=False)
self.ws.frame_write(sock, self.ws.OP_CONT, 'fragment3', fin=True) self.ws.frame_write(sock, self.ws.OP_CONT, 'fragment3', fin=True)
@@ -907,7 +907,7 @@ class TestNodeWebsockets(TestApplicationNode):
_, sock, _ = self.ws.upgrade() _, sock, _ = self.ws.upgrade()
for i in range(0, 2): for _ in range(0, 2):
self.ws.frame_write(sock, self.ws.OP_CONT, 'fragment1', fin=True) self.ws.frame_write(sock, self.ws.OP_CONT, 'fragment1', fin=True)
self.ws.frame_write(sock, self.ws.OP_TEXT, 'fragment2', fin=False) self.ws.frame_write(sock, self.ws.OP_TEXT, 'fragment2', fin=False)
self.ws.frame_write(sock, self.ws.OP_CONT, 'fragment3', fin=True) self.ws.frame_write(sock, self.ws.OP_CONT, 'fragment3', fin=True)

View File

@@ -263,7 +263,7 @@ class TestPerlApplication(TestApplicationPerl):
socks = [] socks = []
for i in range(4): for _ in range(4):
sock = self.get( sock = self.get(
headers={ headers={
'Host': 'localhost', 'Host': 'localhost',

View File

@@ -807,7 +807,7 @@ opcache.preload_user = {option.user or getpass.getuser()}
assert r['headers']['X-Pid'] != pid, 'new instance' assert r['headers']['X-Pid'] != pid, 'new instance'
assert r['headers']['X-Cached'] == '1', 'cached' assert r['headers']['X-Cached'] == '1', 'cached'
def test_php_application_opcache_preload_chdir(self, temp_dir): def test_php_application_opcache_preload_chdir(self):
self.load('opcache') self.load('opcache')
self.check_opcache() self.check_opcache()
@@ -817,7 +817,7 @@ opcache.preload_user = {option.user or getpass.getuser()}
assert self.get()['headers']['X-Cached'] == '0', 'not cached' assert self.get()['headers']['X-Cached'] == '0', 'not cached'
assert self.get()['headers']['X-Cached'] == '1', 'cached' assert self.get()['headers']['X-Cached'] == '1', 'cached'
def test_php_application_opcache_preload_ffr(self, temp_dir): def test_php_application_opcache_preload_ffr(self):
self.load('opcache') self.load('opcache')
self.check_opcache() self.check_opcache()

View File

@@ -39,7 +39,7 @@ Content-Length: 10
""" """
while True: while True:
connection, client_address = sock.accept() connection, _ = sock.accept()
data = recvall(connection).decode() data = recvall(connection).decode()

View File

@@ -5,7 +5,6 @@ import time
from conftest import run_process from conftest import run_process
from unit.applications.lang.python import TestApplicationPython from unit.applications.lang.python import TestApplicationPython
from unit.option import option
from unit.utils import waitforsocket from unit.utils import waitforsocket
@@ -15,7 +14,7 @@ class TestProxyChunked(TestApplicationPython):
SERVER_PORT = 7999 SERVER_PORT = 7999
@staticmethod @staticmethod
def run_server(server_port, temp_dir): def run_server(server_port):
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1) sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
@@ -42,7 +41,7 @@ class TestProxyChunked(TestApplicationPython):
return data return data
while True: while True:
connection, client_address = sock.accept() connection, _ = sock.accept()
req = """HTTP/1.1 200 OK\r\nTransfer-Encoding: chunked""" req = """HTTP/1.1 200 OK\r\nTransfer-Encoding: chunked"""
@@ -85,7 +84,7 @@ class TestProxyChunked(TestApplicationPython):
return self.get(*args, http_10=True, **kwargs) return self.get(*args, http_10=True, **kwargs)
def setup_method(self): def setup_method(self):
run_process(self.run_server, self.SERVER_PORT, option.temp_dir) run_process(self.run_server, self.SERVER_PORT)
waitforsocket(self.SERVER_PORT) waitforsocket(self.SERVER_PORT)
assert 'success' in self.conf( assert 'success' in self.conf(

View File

@@ -875,7 +875,7 @@ last line: 987654321
socks = [] socks = []
for i in range(4): for _ in range(4):
sock = self.get( sock = self.get(
headers={ headers={
'Host': 'localhost', 'Host': 'localhost',

View File

@@ -125,7 +125,7 @@ class TestPythonIsolation(TestApplicationPython):
self.getjson(url='/?path=/proc/self')['body']['FileExists'] == True self.getjson(url='/?path=/proc/self')['body']['FileExists'] == True
), '/proc/self' ), '/proc/self'
def test_python_isolation_cgroup(self, is_su, temp_dir): def test_python_isolation_cgroup(self, is_su):
if not is_su: if not is_su:
pytest.skip('requires root') pytest.skip('requires root')
@@ -148,7 +148,7 @@ class TestPythonIsolation(TestApplicationPython):
assert len(cgroup_rel.parts) >= len(cgroup_abs.parts) assert len(cgroup_rel.parts) >= len(cgroup_abs.parts)
def test_python_isolation_cgroup_two(self, is_su, temp_dir): def test_python_isolation_cgroup_two(self, is_su):
if not is_su: if not is_su:
pytest.skip('requires root') pytest.skip('requires root')

View File

@@ -160,7 +160,7 @@ class TestPythonProcman(TestApplicationPython):
def test_python_processes_connection_keepalive(self): def test_python_processes_connection_keepalive(self):
self.conf_proc({"spare": 0, "max": 6, "idle_timeout": 2}) self.conf_proc({"spare": 0, "max": 6, "idle_timeout": 2})
(resp, sock) = self.get( (_, sock) = self.get(
headers={'Host': 'localhost', 'Connection': 'keep-alive'}, headers={'Host': 'localhost', 'Connection': 'keep-alive'},
start=True, start=True,
read_timeout=1, read_timeout=1,

View File

@@ -30,7 +30,7 @@ class TestRespawn(TestApplicationPython):
subprocess.call(['kill', '-9', *pids]) subprocess.call(['kill', '-9', *pids])
def wait_for_process(self, process, unit_pid): def wait_for_process(self, process, unit_pid):
for i in range(50): for _ in range(50):
found = self.pid_by_name(process, unit_pid) found = self.pid_by_name(process, unit_pid)
if found is not None: if found is not None:

View File

@@ -129,8 +129,8 @@ Connection: close
check_location(f'#{unsafe}', f'#{unsafe_enc}') check_location(f'#{unsafe}', f'#{unsafe_enc}')
# %00-%20 and %7F-%FF always encoded. # %00-%20 and %7F-%FF always encoded.
check_location(u"\u0000\u0018\u001F\u0020\u0021", "%00%18%1F%20!") check_location("\u0000\u0018\u001F\u0020\u0021", "%00%18%1F%20!")
check_location(u"\u007F\u0080н\u20BD", "%7F%C2%80%D0%BD%E2%82%BD") check_location("\u007F\u0080н\u20BD", "%7F%C2%80%D0%BD%E2%82%BD")
# Encoded string detection. If at least one char need to be encoded # Encoded string detection. If at least one char need to be encoded
# then whole string will be encoded. # then whole string will be encoded.

View File

@@ -1490,28 +1490,28 @@ class TestRouting(TestApplicationPython):
return (sock, port) return (sock, port)
sock, port = sock_port() sock, port = sock_port()
sock2, port2 = sock_port() sock2, _ = sock_port()
self.route_match({"source": f'127.0.0.1:{port}'}) self.route_match({"source": f'127.0.0.1:{port}'})
assert self.get(sock=sock)['status'] == 200, 'exact' assert self.get(sock=sock)['status'] == 200, 'exact'
assert self.get(sock=sock2)['status'] == 404, 'exact 2' assert self.get(sock=sock2)['status'] == 404, 'exact 2'
sock, port = sock_port() sock, port = sock_port()
sock2, port2 = sock_port() sock2, _ = sock_port()
self.route_match({"source": f'!127.0.0.1:{port}'}) self.route_match({"source": f'!127.0.0.1:{port}'})
assert self.get(sock=sock)['status'] == 404, 'negative' assert self.get(sock=sock)['status'] == 404, 'negative'
assert self.get(sock=sock2)['status'] == 200, 'negative 2' assert self.get(sock=sock2)['status'] == 200, 'negative 2'
sock, port = sock_port() sock, port = sock_port()
sock2, port2 = sock_port() sock2, _ = sock_port()
self.route_match({"source": [f'*:{port}', "!127.0.0.1"]}) self.route_match({"source": [f'*:{port}', "!127.0.0.1"]})
assert self.get(sock=sock)['status'] == 404, 'negative 3' assert self.get(sock=sock)['status'] == 404, 'negative 3'
assert self.get(sock=sock2)['status'] == 404, 'negative 4' assert self.get(sock=sock2)['status'] == 404, 'negative 4'
sock, port = sock_port() sock, port = sock_port()
sock2, port2 = sock_port() sock2, _ = sock_port()
self.route_match({"source": f'127.0.0.1:{port}-{port}'}) self.route_match({"source": f'127.0.0.1:{port}-{port}'})
assert self.get(sock=sock)['status'] == 200, 'range single' assert self.get(sock=sock)['status'] == 200, 'range single'

View File

@@ -387,7 +387,7 @@ class TestRubyApplication(TestApplicationRuby):
socks = [] socks = []
for i in range(4): for _ in range(4):
sock = self.get( sock = self.get(
headers={ headers={
'Host': 'localhost', 'Host': 'localhost',

View File

@@ -89,7 +89,7 @@ class TestSettings(TestApplicationPython):
self.load('empty') self.load('empty')
def req(): def req():
(resp, sock) = self.http( (_, sock) = self.http(
b"""GET / HTTP/1.1 b"""GET / HTTP/1.1
""", """,
start=True, start=True,
@@ -173,7 +173,7 @@ Connection: close
self.load('empty') self.load('empty')
def req(): def req():
(resp, sock) = self.http( (_, sock) = self.http(
b"""POST / HTTP/1.1 b"""POST / HTTP/1.1
Host: localhost Host: localhost
Content-Length: 10 Content-Length: 10
@@ -293,7 +293,7 @@ Connection: close
self.load('empty') self.load('empty')
def req(): def req():
(resp, sock) = self.get( (_, sock) = self.get(
headers={'Host': 'localhost', 'Connection': 'keep-alive'}, headers={'Host': 'localhost', 'Connection': 'keep-alive'},
start=True, start=True,
read_timeout=1, read_timeout=1,

View File

@@ -74,14 +74,14 @@ class TestStaticChroot(TestApplicationProto):
assert self.get(url='/dir/file')['status'] == 200, 'chroot' assert self.get(url='/dir/file')['status'] == 200, 'chroot'
def test_static_chroot_empty(self, temp_dir): def test_static_chroot_empty(self):
assert 'success' in self.update_action('') assert 'success' in self.update_action('')
assert self.get(url='/dir/file')['status'] == 200, 'empty absolute' assert self.get(url='/dir/file')['status'] == 200, 'empty absolute'
assert 'success' in self.update_action("", ".$uri") assert 'success' in self.update_action("", ".$uri")
assert self.get(url=self.test_path)['status'] == 200, 'empty relative' assert self.get(url=self.test_path)['status'] == 200, 'empty relative'
def test_static_chroot_relative(self, is_su, temp_dir): def test_static_chroot_relative(self, is_su):
if is_su: if is_su:
pytest.skip("Does't work under root.") pytest.skip("Does't work under root.")

View File

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

View File

@@ -28,7 +28,7 @@ class TestTLS(TestApplicationTLS):
{"pass": f"applications/{application}"}, f'listeners/*:{port}' {"pass": f"applications/{application}"}, f'listeners/*:{port}'
) )
def req(self, name='localhost', subject=None, x509=False): def req(self, name='localhost', subject=None):
subj = subject if subject is not None else f'/CN={name}/' subj = subject if subject is not None else f'/CN={name}/'
subprocess.check_output( subprocess.check_output(
@@ -426,7 +426,7 @@ basicConstraints = critical,CA:TRUE"""
assert self.get_ssl()['status'] == 200, 'certificate chain long' assert self.get_ssl()['status'] == 200, 'certificate chain long'
def test_tls_certificate_empty_cn(self, temp_dir): def test_tls_certificate_empty_cn(self):
self.certificate('root', False) self.certificate('root', False)
self.req(subject='/') self.req(subject='/')
@@ -442,7 +442,7 @@ basicConstraints = critical,CA:TRUE"""
assert cert['chain'][0]['subject'] == {}, 'empty subject' assert cert['chain'][0]['subject'] == {}, 'empty subject'
assert cert['chain'][0]['issuer']['common_name'] == 'root', 'issuer' assert cert['chain'][0]['issuer']['common_name'] == 'root', 'issuer'
def test_tls_certificate_empty_cn_san(self, temp_dir): def test_tls_certificate_empty_cn_san(self):
self.certificate('root', False) self.certificate('root', False)
self.openssl_conf( self.openssl_conf(
@@ -535,7 +535,7 @@ basicConstraints = critical,CA:TRUE"""
} }
), 'load application configuration' ), 'load application configuration'
(resp, sock) = self.get_ssl(start=True) (_, sock) = self.get_ssl(start=True)
time.sleep(5) time.sleep(5)

View File

@@ -8,7 +8,7 @@ class TestTLSConfCommand(TestApplicationTLS):
prerequisites = {'modules': {'openssl': 'any'}} prerequisites = {'modules': {'openssl': 'any'}}
@pytest.fixture(autouse=True) @pytest.fixture(autouse=True)
def setup_method_fixture(self, request): def setup_method_fixture(self):
self.certificate() self.certificate()
assert 'success' in self.conf( assert 'success' in self.conf(
@@ -35,7 +35,7 @@ class TestTLSConfCommand(TestApplicationTLS):
# Set one conf_commands (disable protocol). # Set one conf_commands (disable protocol).
(resp, sock) = self.get_ssl(start=True) (_, sock) = self.get_ssl(start=True)
shared_ciphers = sock.shared_ciphers() shared_ciphers = sock.shared_ciphers()
protocols = list(set(c[1] for c in shared_ciphers)) protocols = list(set(c[1] for c in shared_ciphers))
@@ -55,7 +55,7 @@ class TestTLSConfCommand(TestApplicationTLS):
sock.close() sock.close()
if len(protocols) > 1: if len(protocols) > 1:
(resp, sock) = self.get_ssl(start=True) (_, sock) = self.get_ssl(start=True)
cipher = sock.cipher() cipher = sock.cipher()
assert cipher[1] != protocol, 'new protocol used' assert cipher[1] != protocol, 'new protocol used'
@@ -82,7 +82,7 @@ class TestTLSConfCommand(TestApplicationTLS):
), 'cipher disabled' ), 'cipher disabled'
if len(ciphers) > 1: if len(ciphers) > 1:
(resp, sock) = self.get_ssl(start=True) (_, sock) = self.get_ssl(start=True)
cipher_new = sock.cipher() cipher_new = sock.cipher()
assert cipher_new[1] == cipher[1], 'previous protocol used' assert cipher_new[1] == cipher[1], 'previous protocol used'

View File

@@ -19,7 +19,7 @@ class TestTLSSession(TestApplicationTLS):
prerequisites = {'modules': {'openssl': 'any'}} prerequisites = {'modules': {'openssl': 'any'}}
@pytest.fixture(autouse=True) @pytest.fixture(autouse=True)
def setup_method_fixture(self, request): def setup_method_fixture(self):
self.certificate() self.certificate()
assert 'success' in self.conf( assert 'success' in self.conf(
@@ -70,21 +70,21 @@ class TestTLSSession(TestApplicationTLS):
) )
def test_tls_session(self): def test_tls_session(self):
client, sess, ctx, reused = self.connect() _, sess, ctx, reused = self.connect()
assert not reused, 'new connection' assert not reused, 'new connection'
client, _, _, reused = self.connect(ctx, sess) _, _, _, reused = self.connect(ctx, sess)
assert not reused, 'no cache' assert not reused, 'no cache'
assert 'success' in self.add_session(cache_size=2) assert 'success' in self.add_session(cache_size=2)
client, sess, ctx, reused = self.connect() _, sess, ctx, reused = self.connect()
assert not reused, 'new connection cache' assert not reused, 'new connection cache'
client, _, _, reused = self.connect(ctx, sess) _, _, _, reused = self.connect(ctx, sess)
assert reused, 'cache' assert reused, 'cache'
client, _, _, reused = self.connect(ctx, sess) _, _, _, reused = self.connect(ctx, sess)
assert reused, 'cache 2' assert reused, 'cache 2'
# check that at least one session of four is not reused # check that at least one session of four is not reused
@@ -108,15 +108,15 @@ class TestTLSSession(TestApplicationTLS):
def test_tls_session_timeout(self): def test_tls_session_timeout(self):
assert 'success' in self.add_session(cache_size=5, timeout=1) assert 'success' in self.add_session(cache_size=5, timeout=1)
client, sess, ctx, reused = self.connect() _, sess, ctx, reused = self.connect()
assert not reused, 'new connection' assert not reused, 'new connection'
client, _, _, reused = self.connect(ctx, sess) _, _, _, reused = self.connect(ctx, sess)
assert reused, 'no timeout' assert reused, 'no timeout'
time.sleep(3) time.sleep(3)
client, _, _, reused = self.connect(ctx, sess) _, _, _, reused = self.connect(ctx, sess)
assert not reused, 'timeout' assert not reused, 'timeout'
def test_tls_session_invalid(self): def test_tls_session_invalid(self):

View File

@@ -7,7 +7,6 @@ from OpenSSL.SSL import (
TLSv1_2_METHOD, TLSv1_2_METHOD,
Context, Context,
Connection, Connection,
Session,
_lib, _lib,
) )
from unit.applications.tls import TestApplicationTLS from unit.applications.tls import TestApplicationTLS
@@ -22,7 +21,7 @@ class TestTLSTicket(TestApplicationTLS):
49TZXi/Y4/8RSIO7QPsU51/HLR1gWIMhVM2m9yh93Bw=' 49TZXi/Y4/8RSIO7QPsU51/HLR1gWIMhVM2m9yh93Bw='
@pytest.fixture(autouse=True) @pytest.fixture(autouse=True)
def setup_method_fixture(self, request): def setup_method_fixture(self):
self.certificate() self.certificate()
listener_conf = { listener_conf = {

View File

@@ -26,7 +26,7 @@ class TestApplicationProto(TestControl):
def wait_for_record(self, pattern, name='unit.log', wait=150, flags=re.M): def wait_for_record(self, pattern, name='unit.log', wait=150, flags=re.M):
with Log.open(name) as f: with Log.open(name) as f:
for i in range(wait): for _ in range(wait):
found = re.search(pattern, f.read(), flags) found = re.search(pattern, f.read(), flags)
if found is not None: if found is not None:

View File

@@ -14,7 +14,6 @@ http = TestHTTP()
def check_isolation(): def check_isolation():
test_conf = {"namespaces": {"credential": True}}
available = option.available available = option.available
conf = '' conf = ''

View File

@@ -24,7 +24,7 @@ def public_dir(path):
def waitforfiles(*files, timeout=50): def waitforfiles(*files, timeout=50):
for i in range(timeout): for _ in range(timeout):
wait = False wait = False
for f in files: for f in files:
@@ -41,10 +41,10 @@ def waitforfiles(*files, timeout=50):
def waitforglob(pattern, count=1, timeout=50): def waitforglob(pattern, count=1, timeout=50):
for i in range(timeout): for _ in range(timeout):
n = 0 n = 0
for f in glob.glob(pattern): for _ in glob.glob(pattern):
n += 1 n += 1
if n == count: if n == count:
@@ -56,7 +56,7 @@ def waitforglob(pattern, count=1, timeout=50):
def waitforsocket(port): def waitforsocket(port):
for i in range(50): for _ in range(50):
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock: with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
try: try:
sock.settimeout(5) sock.settimeout(5)
@@ -102,7 +102,7 @@ def sysctl():
def waitformount(template, timeout=50): def waitformount(template, timeout=50):
for i in range(timeout): for _ in range(timeout):
if findmnt().find(template) != -1: if findmnt().find(template) != -1:
return True return True
@@ -112,7 +112,7 @@ def waitformount(template, timeout=50):
def waitforunmount(template, timeout=50): def waitforunmount(template, timeout=50):
for i in range(timeout): for _ in range(timeout):
if findmnt().find(template) == -1: if findmnt().find(template) == -1:
return True return True