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

Tests: unified setup method usage.

Browse source 
To make fixtures accessible inside of setup methods in tests all these methods
are renamed to the "setup_method_fixture" and decorated by autouse flag.

Also all setup methods moved to the top of the files.
This commit is contained in:
Andrei Zeliankou 2023-05-25 16:56:14 +01:00
parent 3e4fa1e202
commit 18fcc07c77
15 changed files with 113 additions and 93 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

8
test/test_client_ip.py
View file

@ -1,3 +1,4 @@
import pytest
from unit.applications.lang.python import TestApplicationPython
from unit.option import option
@ -5,6 +6,10 @@ from unit.option import option
class TestClientIP(TestApplicationPython):
prerequisites = {'modules': {'python': 'any'}}
@pytest.fixture(autouse=True)
def setup_method_fixture(self):
self.load('client_ip')
def client_ip(self, options):
assert 'success' in self.conf(
{
@ -39,9 +44,6 @@ class TestClientIP(TestApplicationPython):
headers={'Connection': 'close', 'X-Forwarded-For': xff},
)['body']
def setup_method(self):
self.load('client_ip')
def test_client_ip_single_ip(self):
self.client_ip(
{'header': 'X-Forwarded-For', 'source': '123.123.123.123'}

8
test/test_forwarded_header.py
View file

@ -1,9 +1,14 @@
import pytest
from unit.applications.lang.python import TestApplicationPython
class TestForwardedHeader(TestApplicationPython):
prerequisites = {'modules': {'python': 'any'}}
@pytest.fixture(autouse=True)
def setup_method_fixture(self):
self.load('forwarded_header')
def forwarded_header(self, forwarded):
assert 'success' in self.conf(
{
@ -40,9 +45,6 @@ class TestForwardedHeader(TestApplicationPython):
def get_scheme(self, *args, **kwargs):
return self.get_fwd(*args, **kwargs)['Url-Scheme']
def setup_method(self):
self.load('forwarded_header')
def test_forwarded_header_single_ip(self):
self.forwarded_header(
{

11
test/test_java_isolation_rootfs.py
View file

@ -9,13 +9,14 @@ from unit.option import option
class TestJavaIsolationRootfs(TestApplicationJava):
prerequisites = {'modules': {'java': 'all'}}
def setup_method(self, is_su):
@pytest.fixture(autouse=True)
def setup_method_fixture(self, is_su, temp_dir):
if not is_su:
pytest.skip('require root')
os.makedirs(f'{option.temp_dir}/jars')
os.makedirs(f'{option.temp_dir}/tmp')
os.chmod(f'{option.temp_dir}/tmp', 0o777)
os.makedirs(f'{temp_dir}/jars')
os.makedirs(f'{temp_dir}/tmp')
os.chmod(f'{temp_dir}/tmp', 0o777)
try:
subprocess.run(
@ -23,7 +24,7 @@ class TestJavaIsolationRootfs(TestApplicationJava):
"mount",
"--bind",
f'{option.current_dir}/build',
f'{option.temp_dir}/jars',
f'{temp_dir}/jars',
],
stderr=subprocess.STDOUT,
)

6
test/test_njs.py
View file

@ -1,5 +1,6 @@
import os
import pytest
from unit.applications.proto import TestApplicationProto
from unit.option import option
from unit.utils import waitforfiles
@ -8,12 +9,13 @@ from unit.utils import waitforfiles
class TestNJS(TestApplicationProto):
prerequisites = {'modules': {'njs': 'any'}}
def setup_method(self):
@pytest.fixture(autouse=True)
def setup_method_fixture(self, temp_dir):
assert 'success' in self.conf(
{
"listeners": {"*:7080": {"pass": "routes"}},
"routes": [
{"action": {"share": f"{option.temp_dir}/assets$uri"}}
{"action": {"share": f"{temp_dir}/assets$uri"}}
],
}
)

77
test/test_proxy.py
View file

@ -14,6 +14,45 @@ class TestProxy(TestApplicationPython):
SERVER_PORT = 7999
@pytest.fixture(autouse=True)
def setup_method_fixture(self):
run_process(self.run_server, self.SERVER_PORT)
waitforsocket(self.SERVER_PORT)
python_dir = f'{option.test_dir}/python'
assert 'success' in self.conf(
{
"listeners": {
"*:7080": {"pass": "routes"},
"*:7081": {"pass": "applications/mirror"},
},
"routes": [{"action": {"proxy": "http://127.0.0.1:7081"}}],
"applications": {
"mirror": {
"type": self.get_application_type(),
"processes": {"spare": 0},
"path": f'{python_dir}/mirror',
"working_directory": f'{python_dir}/mirror',
"module": "wsgi",
},
"custom_header": {
"type": self.get_application_type(),
"processes": {"spare": 0},
"path": f'{python_dir}/custom_header',
"working_directory": f'{python_dir}/custom_header',
"module": "wsgi",
},
"delayed": {
"type": self.get_application_type(),
"processes": {"spare": 0},
"path": f'{python_dir}/delayed',
"working_directory": f'{python_dir}/delayed',
"module": "wsgi",
},
},
}
), 'proxy initial configuration'
@staticmethod
def run_server(server_port):
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
@ -59,44 +98,6 @@ Content-Length: 10
def post_http10(self, *args, **kwargs):
return self.post(*args, http_10=True, **kwargs)
def setup_method(self):
run_process(self.run_server, self.SERVER_PORT)
waitforsocket(self.SERVER_PORT)
python_dir = f'{option.test_dir}/python'
assert 'success' in self.conf(
{
"listeners": {
"*:7080": {"pass": "routes"},
"*:7081": {"pass": "applications/mirror"},
},
"routes": [{"action": {"proxy": "http://127.0.0.1:7081"}}],
"applications": {
"mirror": {
"type": self.get_application_type(),
"processes": {"spare": 0},
"path": f'{python_dir}/mirror',
"working_directory": f'{python_dir}/mirror',
"module": "wsgi",
},
"custom_header": {
"type": self.get_application_type(),
"processes": {"spare": 0},
"path": f'{python_dir}/custom_header',
"working_directory": f'{python_dir}/custom_header',
"module": "wsgi",
},
"delayed": {
"type": self.get_application_type(),
"processes": {"spare": 0},
"path": f'{python_dir}/delayed',
"working_directory": f'{python_dir}/delayed',
"module": "wsgi",
},
},
}
), 'proxy initial configuration'
def test_proxy_http10(self):
for _ in range(10):
assert self.get_http10()['status'] == 200, 'status'

40
test/test_proxy_chunked.py
View file

@ -3,6 +3,7 @@ import select
import socket
import time
import pytest
from conftest import run_process
from unit.applications.lang.python import TestApplicationPython
from unit.utils import waitforsocket
@ -13,6 +14,26 @@ class TestProxyChunked(TestApplicationPython):
SERVER_PORT = 7999
@pytest.fixture(autouse=True)
def setup_method_fixture(self):
run_process(self.run_server, self.SERVER_PORT)
waitforsocket(self.SERVER_PORT)
assert 'success' in self.conf(
{
"listeners": {
"*:7080": {"pass": "routes"},
},
"routes": [
{
"action": {
"proxy": f'http://127.0.0.1:{self.SERVER_PORT}'
}
}
],
}
), 'proxy initial configuration'
@staticmethod
def run_server(server_port):
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
@ -83,25 +104,6 @@ class TestProxyChunked(TestApplicationPython):
def get_http10(self, *args, **kwargs):
return self.get(*args, http_10=True, **kwargs)
def setup_method(self):
run_process(self.run_server, self.SERVER_PORT)
waitforsocket(self.SERVER_PORT)
assert 'success' in self.conf(
{
"listeners": {
"*:7080": {"pass": "routes"},
},
"routes": [
{
"action": {
"proxy": f'http://127.0.0.1:{self.SERVER_PORT}'
}
}
],
}
), 'proxy initial configuration'
def test_proxy_chunked(self):
for _ in range(10):
assert self.get_http10(body='\r\n\r\n0\r\n\r\n')['status'] == 200

5
test/test_python_procman.py
View file

@ -11,8 +11,9 @@ from unit.option import option
class TestPythonProcman(TestApplicationPython):
prerequisites = {'modules': {'python': 'any'}}
def setup_method(self):
self.app_name = f'app-{option.temp_dir.split("/")[-1]}'
@pytest.fixture(autouse=True)
def setup_method_fixture(self, temp_dir):
self.app_name = f'app-{temp_dir.split("/")[-1]}'
self.app_proc = f'applications/{self.app_name}/processes'
self.load('empty', self.app_name)

7
test/test_respawn.py
View file

@ -2,8 +2,8 @@ import re
import subprocess
import time
import pytest
from unit.applications.lang.python import TestApplicationPython
from unit.option import option
class TestRespawn(TestApplicationPython):
@ -12,8 +12,9 @@ class TestRespawn(TestApplicationPython):
PATTERN_ROUTER = 'unit: router'
PATTERN_CONTROLLER = 'unit: controller'
def setup_method(self):
self.app_name = f'app-{option.temp_dir.split("/")[-1]}'
@pytest.fixture(autouse=True)
def setup_method_fixture(self, temp_dir):
self.app_name = f'app-{temp_dir.split("/")[-1]}'
self.load('empty', self.app_name)

4
test/test_return.py
View file

@ -1,12 +1,14 @@
import re
import pytest
from unit.applications.proto import TestApplicationProto
class TestReturn(TestApplicationProto):
prerequisites = {}
def setup_method(self):
@pytest.fixture(autouse=True)
def setup_method_fixture(self):
self._load_conf(
{
"listeners": {"*:7080": {"pass": "routes"}},

3
test/test_rewrite.py
View file

@ -8,7 +8,8 @@ from unit.option import option
class TestRewrite(TestApplicationProto):
prerequisites = {}
def setup_method(self):
@pytest.fixture(autouse=True)
def setup_method_fixture(self):
assert 'success' in self.conf(
{
"listeners": {"*:7080": {"pass": "routes"}},

3
test/test_routing.py
View file

@ -7,7 +7,8 @@ from unit.option import option
class TestRouting(TestApplicationPython):
prerequisites = {'modules': {'python': 'any'}}
def setup_method(self):
@pytest.fixture(autouse=True)
def setup_method_fixture(self):
assert 'success' in self.conf(
{
"listeners": {"*:7080": {"pass": "routes"}},

22
test/test_static.py
View file

@ -3,21 +3,21 @@ import socket
import pytest
from unit.applications.proto import TestApplicationProto
from unit.option import option
from unit.utils import waitforfiles
class TestStatic(TestApplicationProto):
prerequisites = {}
def setup_method(self):
os.makedirs(f'{option.temp_dir}/assets/dir')
with open(f'{option.temp_dir}/assets/index.html', 'w') as index, open(
f'{option.temp_dir}/assets/README', 'w'
) as readme, open(
f'{option.temp_dir}/assets/log.log', 'w'
) as log, open(
f'{option.temp_dir}/assets/dir/file', 'w'
@pytest.fixture(autouse=True)
def setup_method_fixture(self, temp_dir):
os.makedirs(f'{temp_dir}/assets/dir')
assets_dir = f'{temp_dir}/assets'
with open(f'{assets_dir}/index.html', 'w') as index, open(
f'{assets_dir}/README', 'w'
) as readme, open(f'{assets_dir}/log.log', 'w') as log, open(
f'{assets_dir}/dir/file', 'w'
) as file:
index.write('0123456789')
readme.write('readme')
@ -27,9 +27,7 @@ class TestStatic(TestApplicationProto):
self._load_conf(
{
"listeners": {"*:7080": {"pass": "routes"}},
"routes": [
{"action": {"share": f'{option.temp_dir}/assets$uri'}}
],
"routes": [{"action": {"share": f'{assets_dir}$uri'}}],
"settings": {
"http": {
"static": {

4
test/test_tls_sni.py
View file

@ -1,6 +1,7 @@
import ssl
import subprocess
import pytest
from unit.applications.tls import TestApplicationTLS
from unit.option import option
@ -8,7 +9,8 @@ from unit.option import option
class TestTLSSNI(TestApplicationTLS):
prerequisites = {'modules': {'openssl': 'any'}}
def setup_method(self):
@pytest.fixture(autouse=True)
def setup_method_fixture(self):
self._load_conf(
{
"listeners": {"*:7080": {"pass": "routes"}},

4
test/test_upstreams_rr.py
View file

@ -1,6 +1,7 @@
import os
import re
import pytest
from unit.applications.lang.python import TestApplicationPython
from unit.option import option
@ -8,7 +9,8 @@ from unit.option import option
class TestUpstreamsRR(TestApplicationPython):
prerequisites = {'modules': {'python': 'any'}}
def setup_method(self):
@pytest.fixture(autouse=True)
def setup_method_fixture(self):
assert 'success' in self.conf(
{
"listeners": {

4
test/test_variables.py
View file

@ -1,6 +1,7 @@
import re
import time
import pytest
from unit.applications.proto import TestApplicationProto
from unit.option import option
@ -8,7 +9,8 @@ from unit.option import option
class TestVariables(TestApplicationProto):
prerequisites = {}
def setup_method(self):
@pytest.fixture(autouse=True)
def setup_method_fixture(self):
assert 'success' in self.conf(
{
"listeners": {"*:7080": {"pass": "routes"}},