Tests: skip_alert() converted to the fixture.

This commit is contained in:
Andrei Zeliankou
2020-12-08 14:37:33 +00:00
parent 8f91628563
commit b2e767819f
15 changed files with 32 additions and 39 deletions

View File

@@ -74,10 +74,6 @@ def pytest_configure(config):
fcntl.fcntl(sys.stdout.fileno(), fcntl.F_SETFL, 0) fcntl.fcntl(sys.stdout.fileno(), fcntl.F_SETFL, 0)
def skip_alert(*alerts):
option.skip_alerts.extend(alerts)
def pytest_generate_tests(metafunc): def pytest_generate_tests(metafunc):
cls = metafunc.cls cls = metafunc.cls
if (not hasattr(cls, 'application_type') if (not hasattr(cls, 'application_type')
@@ -314,7 +310,7 @@ def _check_alerts(path=None):
alerts = re.findall(r'.+\[alert\].+', log) alerts = re.findall(r'.+\[alert\].+', log)
if alerts: if alerts:
print('All alerts/sanitizer errors found in log:') print('\nAll alerts/sanitizer errors found in log:')
[print(alert) for alert in alerts] [print(alert) for alert in alerts]
found = True found = True
@@ -378,6 +374,14 @@ def stop_processes():
return 'Fail to stop process(es)' return 'Fail to stop process(es)'
@pytest.fixture()
def skip_alert():
def _skip(*alerts):
option.skip_alerts.extend(alerts)
return _skip
@pytest.fixture @pytest.fixture
def temp_dir(request): def temp_dir(request):
return unit_instance['temp_dir'] return unit_instance['temp_dir']

View File

@@ -4,7 +4,6 @@ from distutils.version import LooseVersion
import pytest import pytest
from conftest import skip_alert
from unit.applications.lang.python import TestApplicationPython from unit.applications.lang.python import TestApplicationPython
from unit.option import option from unit.option import option
@@ -361,7 +360,7 @@ Connection: close
self.get(headers=headers_delay_1) self.get(headers=headers_delay_1)
def test_asgi_application_loading_error(self): def test_asgi_application_loading_error(self, skip_alert):
skip_alert(r'Python failed to import module "blah"') skip_alert(r'Python failed to import module "blah"')
self.load('empty', module="blah") self.load('empty', module="blah")

View File

@@ -4,7 +4,6 @@ from distutils.version import LooseVersion
import pytest import pytest
from conftest import skip_alert
from unit.applications.lang.python import TestApplicationPython from unit.applications.lang.python import TestApplicationPython
from unit.applications.websockets import TestApplicationWebsocket from unit.applications.websockets import TestApplicationWebsocket
from unit.option import option from unit.option import option
@@ -17,7 +16,8 @@ class TestASGIWebsockets(TestApplicationPython):
ws = TestApplicationWebsocket() ws = TestApplicationWebsocket()
def setup_method(self): @pytest.fixture(autouse=True)
def setup_method_fixture(self, request, 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'

View File

@@ -1,6 +1,5 @@
import pytest import pytest
from conftest import skip_alert
from unit.control import TestControl from unit.control import TestControl
@@ -337,7 +336,7 @@ class TestConfiguration(TestControl):
assert 'success' in self.conf(conf) assert 'success' in self.conf(conf)
def test_unprivileged_user_error(self, is_su): def test_unprivileged_user_error(self, is_su, skip_alert):
skip_alert(r'cannot set user "root"', r'failed to apply new conf') skip_alert(r'cannot set user "root"', r'failed to apply new conf')
if is_su: if is_su:
pytest.skip('unprivileged tests') pytest.skip('unprivileged tests')

View File

@@ -3,7 +3,6 @@ import os
import re import re
import time import time
from conftest import skip_alert
from unit.applications.lang.java import TestApplicationJava from unit.applications.lang.java import TestApplicationJava
from unit.option import option from unit.option import option
from unit.utils import public_dir from unit.utils import public_dir
@@ -11,7 +10,7 @@ from unit.utils import public_dir
class TestJavaApplication(TestApplicationJava): class TestJavaApplication(TestApplicationJava):
prerequisites = {'modules': {'java': 'all'}} prerequisites = {'modules': {'java': 'all'}}
def test_java_conf_error(self, temp_dir): def test_java_conf_error(self, temp_dir, skip_alert):
skip_alert( skip_alert(
r'realpath.*failed', r'realpath.*failed',
r'failed to apply new conf', r'failed to apply new conf',

View File

@@ -3,7 +3,6 @@ import time
import pytest import pytest
from conftest import skip_alert
from unit.applications.lang.java import TestApplicationJava from unit.applications.lang.java import TestApplicationJava
from unit.applications.websockets import TestApplicationWebsocket from unit.applications.websockets import TestApplicationWebsocket
from unit.option import option from unit.option import option
@@ -14,7 +13,8 @@ class TestJavaWebsockets(TestApplicationJava):
ws = TestApplicationWebsocket() ws = TestApplicationWebsocket()
def setup_method(self): @pytest.fixture(autouse=True)
def setup_method_fixture(self, request, 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'

View File

@@ -3,7 +3,6 @@ import time
import pytest import pytest
from conftest import skip_alert
from unit.applications.lang.node import TestApplicationNode from unit.applications.lang.node import TestApplicationNode
from unit.applications.websockets import TestApplicationWebsocket from unit.applications.websockets import TestApplicationWebsocket
from unit.option import option from unit.option import option
@@ -14,7 +13,8 @@ class TestNodeWebsockets(TestApplicationNode):
ws = TestApplicationWebsocket() ws = TestApplicationWebsocket()
def setup_method(self): @pytest.fixture(autouse=True)
def setup_method_fixture(self, request, 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'

View File

@@ -2,7 +2,6 @@ import re
import pytest import pytest
from conftest import skip_alert
from conftest import unit_stop from conftest import unit_stop
from unit.applications.lang.perl import TestApplicationPerl from unit.applications.lang.perl import TestApplicationPerl
@@ -170,7 +169,7 @@ class TestPerlApplication(TestApplicationPerl):
assert self.get()['body'] == 'body\n', 'body io file' assert self.get()['body'] == 'body\n', 'body io file'
@pytest.mark.skip('not yet') @pytest.mark.skip('not yet')
def test_perl_application_syntax_error(self): def test_perl_application_syntax_error(self, skip_alert):
skip_alert(r'PSGI: Failed to parse script') skip_alert(r'PSGI: Failed to parse script')
self.load('syntax_error') self.load('syntax_error')

View File

@@ -5,7 +5,6 @@ import time
import pytest import pytest
from conftest import run_process from conftest import run_process
from conftest import skip_alert
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.utils import waitforsocket from unit.utils import waitforsocket
@@ -482,7 +481,7 @@ Content-Length: 10
check_proxy('http://[:]:7080') check_proxy('http://[:]:7080')
check_proxy('http://[::7080') check_proxy('http://[::7080')
def test_proxy_loop(self): def test_proxy_loop(self, skip_alert):
skip_alert( skip_alert(
r'socket.*failed', r'socket.*failed',
r'accept.*failed', r'accept.*failed',

View File

@@ -5,7 +5,6 @@ import time
import pytest import pytest
from conftest import skip_alert
from conftest import unit_stop from conftest import unit_stop
from unit.applications.lang.python import TestApplicationPython from unit.applications.lang.python import TestApplicationPython
from unit.option import option from unit.option import option
@@ -520,13 +519,13 @@ last line: 987654321
assert self.get()['body'] == 'body\n', 'body io file' assert self.get()['body'] == 'body\n', 'body io file'
@pytest.mark.skip('not yet') @pytest.mark.skip('not yet')
def test_python_application_syntax_error(self): def test_python_application_syntax_error(self, skip_alert):
skip_alert(r'Python failed to import module "wsgi"') skip_alert(r'Python failed to import module "wsgi"')
self.load('syntax_error') self.load('syntax_error')
assert self.get()['status'] == 500, 'syntax error' assert self.get()['status'] == 500, 'syntax error'
def test_python_application_loading_error(self): def test_python_application_loading_error(self, skip_alert):
skip_alert(r'Python failed to import module "blah"') skip_alert(r'Python failed to import module "blah"')
self.load('empty', module="blah") self.load('empty', module="blah")
@@ -791,7 +790,7 @@ last line: 987654321
assert obj['UID'] == nobody_uid, 'root uid group=root' assert obj['UID'] == nobody_uid, 'root uid group=root'
assert obj['GID'] == 0, 'root gid group=root' assert obj['GID'] == 0, 'root gid group=root'
def test_python_application_callable(self): def test_python_application_callable(self, skip_alert):
skip_alert(r'Python failed to get "blah" from module') skip_alert(r'Python failed to get "blah" from module')
self.load('callable') self.load('callable')

View File

@@ -2,7 +2,6 @@ import re
import subprocess import subprocess
import time import time
from conftest import skip_alert
from unit.applications.lang.python import TestApplicationPython from unit.applications.lang.python import TestApplicationPython
from unit.option import option from unit.option import option
@@ -56,7 +55,7 @@ class TestRespawn(TestApplicationPython):
assert len(re.findall(self.PATTERN_CONTROLLER, output)) == 1 assert len(re.findall(self.PATTERN_CONTROLLER, output)) == 1
assert len(re.findall(self.app_name, output)) == 1 assert len(re.findall(self.app_name, output)) == 1
def test_respawn_router(self): def test_respawn_router(self, skip_alert):
pid = self.pid_by_name(self.PATTERN_ROUTER) pid = self.pid_by_name(self.PATTERN_ROUTER)
self.kill_pids(pid) self.kill_pids(pid)
@@ -66,7 +65,7 @@ class TestRespawn(TestApplicationPython):
self.smoke_test() self.smoke_test()
def test_respawn_controller(self): def test_respawn_controller(self, skip_alert):
pid = self.pid_by_name(self.PATTERN_CONTROLLER) pid = self.pid_by_name(self.PATTERN_CONTROLLER)
self.kill_pids(pid) self.kill_pids(pid)
@@ -78,7 +77,7 @@ class TestRespawn(TestApplicationPython):
self.smoke_test() self.smoke_test()
def test_respawn_application(self): def test_respawn_application(self, skip_alert):
pid = self.pid_by_name(self.app_name) pid = self.pid_by_name(self.app_name)
self.kill_pids(pid) self.kill_pids(pid)

View File

@@ -1,7 +1,6 @@
# -*- coding: utf-8 -*- # -*- coding: utf-8 -*-
import pytest import pytest
from conftest import skip_alert
from unit.applications.proto import TestApplicationProto from unit.applications.proto import TestApplicationProto
from unit.option import option from unit.option import option
@@ -366,7 +365,7 @@ class TestRouting(TestApplicationProto):
assert self.get()['status'] == 200, 'route match absent' assert self.get()['status'] == 200, 'route match absent'
def test_routes_route_action_absent(self): def test_routes_route_action_absent(self, skip_alert):
skip_alert(r'failed to apply new conf') skip_alert(r'failed to apply new conf')
assert 'error' in self.conf( assert 'error' in self.conf(
@@ -755,7 +754,7 @@ class TestRouting(TestApplicationProto):
'routes/main' 'routes/main'
), 'route edit configure 9' ), 'route edit configure 9'
def test_match_edit(self): def test_match_edit(self, skip_alert):
skip_alert(r'failed to apply new conf') skip_alert(r'failed to apply new conf')
self.route_match({"method": ["GET", "POST"]}) self.route_match({"method": ["GET", "POST"]})
@@ -1352,7 +1351,7 @@ class TestRouting(TestApplicationProto):
assert self.get(url='/?var2=val2')['status'] == 404, 'arr 7' assert self.get(url='/?var2=val2')['status'] == 404, 'arr 7'
assert self.get(url='/?var3=foo')['status'] == 200, 'arr 8' assert self.get(url='/?var3=foo')['status'] == 200, 'arr 8'
def test_routes_match_arguments_invalid(self): def test_routes_match_arguments_invalid(self, skip_alert):
# TODO remove it after controller fixed # TODO remove it after controller fixed
skip_alert(r'failed to apply new conf') skip_alert(r'failed to apply new conf')

View File

@@ -2,7 +2,6 @@ import re
import pytest import pytest
from conftest import skip_alert
from conftest import unit_stop from conftest import unit_stop
from unit.applications.lang.ruby import TestApplicationRuby from unit.applications.lang.ruby import TestApplicationRuby
@@ -160,7 +159,7 @@ class TestRubyApplication(TestApplicationRuby):
assert self.post(body=body)['body'] == body, 'input rewind' assert self.post(body=body)['body'] == body, 'input rewind'
@pytest.mark.skip('not yet') @pytest.mark.skip('not yet')
def test_ruby_application_syntax_error(self): def test_ruby_application_syntax_error(self, skip_alert):
skip_alert( skip_alert(
r'Failed to parse rack script', r'Failed to parse rack script',
r'syntax error', r'syntax error',

View File

@@ -1,6 +1,5 @@
import os import os
from conftest import skip_alert
from unit.applications.proto import TestApplicationProto from unit.applications.proto import TestApplicationProto
from unit.option import option from unit.option import option
@@ -117,7 +116,7 @@ class TestStatic(TestApplicationProto):
assert resp['status'] == 200, 'fallback proxy status' assert resp['status'] == 200, 'fallback proxy status'
assert resp['body'] == '', 'fallback proxy' assert resp['body'] == '', 'fallback proxy'
def test_fallback_proxy_loop(self): def test_fallback_proxy_loop(self, skip_alert):
skip_alert( skip_alert(
r'open.*/blah/index.html.*failed', r'open.*/blah/index.html.*failed',
r'accept.*failed', r'accept.*failed',

View File

@@ -5,7 +5,6 @@ import subprocess
import pytest import pytest
from conftest import skip_alert
from unit.applications.tls import TestApplicationTLS from unit.applications.tls import TestApplicationTLS
from unit.option import option from unit.option import option
@@ -505,7 +504,7 @@ basicConstraints = critical,CA:TRUE"""
'/certificates' '/certificates'
), 'remove all certificates' ), 'remove all certificates'
def test_tls_application_respawn(self): def test_tls_application_respawn(self, skip_alert):
self.load('mirror') self.load('mirror')
self.certificate() self.certificate()