import os
import re
import pytest
from unit.applications.lang.python import ApplicationPython
from unit.option import option
prerequisites = {'modules': {'python': 'any'}}
client = ApplicationPython()
@pytest.fixture(autouse=True)
def setup_method_fixture():
assert 'success' in client.conf(
{
"listeners": {
"*:7080": {"pass": "upstreams/one"},
"*:7090": {"pass": "upstreams/two"},
"*:7081": {"pass": "routes/one"},
"*:7082": {"pass": "routes/two"},
"*:7083": {"pass": "routes/three"},
},
"upstreams": {
"one": {
"servers": {
"127.0.0.1:7081": {},
"127.0.0.1:7082": {},
},
},
"two": {
"servers": {
"127.0.0.1:7081": {},
"127.0.0.1:7082": {},
},
},
},
"routes": {
"one": [{"action": {"return": 200}}],
"two": [{"action": {"return": 201}}],
"three": [{"action": {"return": 202}}],
},
"applications": {},
},
), 'upstreams initial configuration'
client.cpu_count = os.cpu_count()
def get_resps(req=100, port=7080):
resps = [0]
for _ in range(req):
status = client.get(port=port)['status']
if 200 > status or status > 209:
continue
ups = status % 10
if ups > len(resps) - 1:
resps.extend([0] * (ups - len(resps) + 1))
resps[ups] += 1
return resps
def get_resps_sc(req=100, port=7080):
to_send = b"""GET / HTTP/1.1
Host: localhost
""" * (
req - 1
)
to_send += b"""GET / HTTP/1.1
Host: localhost
Connection: close
"""
resp = client.http(to_send, raw_resp=True, raw=True, port=port)
status = re.findall(r'HTTP\/\d\.\d\s(\d\d\d)', resp)
status = list(filter(lambda x: x[:2] == '20', status))
ups = list(map(lambda x: int(x[-1]), status))
resps = [0] * (max(ups) + 1)
for _, up in enumerate(ups):
resps[up] += 1
return resps
def test_upstreams_rr_no_weight():
resps = get_resps()
assert sum(resps) == 100, 'no weight sum'
assert abs(resps[0] - resps[1]) <= client.cpu_count, 'no weight'
assert 'success' in client.conf_delete(
'upstreams/one/servers/127.0.0.1:7081'
), 'no weight server remove'
resps = get_resps(req=50)
assert resps[1] == 50, 'no weight 2'
assert 'success' in client.conf(
{}, 'upstreams/one/servers/127.0.0.1:7081'
), 'no weight server revert'
resps = get_resps()
assert sum(resps) == 100, 'no weight 3 sum'
assert abs(resps[0] - resps[1]) <= client.cpu_count, 'no weight 3'
assert 'success' in client.conf(
{}, 'upstreams/one/servers/127.0.0.1:7083'
), 'no weight server new'
resps = get_resps()
assert sum(resps) == 100, 'no weight 4 sum'
assert max(resps) - min(resps) <= client.cpu_count, 'no weight 4'
resps = get_resps_sc(req=30)
assert resps[0] == 10, 'no weight 4 0'
assert resps[1] == 10, 'no weight 4 1'
assert resps[2] == 10, 'no weight 4 2'
def test_upstreams_rr_weight():
assert 'success' in client.conf(
{"weight": 3}, 'upstreams/one/servers/127.0.0.1:7081'
), 'configure weight'
resps = get_resps_sc()
assert resps[0] == 75, 'weight 3 0'
assert resps[1] == 25, 'weight 3 1'
assert 'success' in client.conf_delete(
'upstreams/one/servers/127.0.0.1:7081/weight'
), 'configure weight remove'
resps = get_resps_sc(req=10)
assert resps[0] == 5, 'weight 0 0'
assert resps[1] == 5, 'weight 0 1'
assert 'success' in client.conf(
'1', 'upstreams/one/servers/127.0.0.1:7081/weight'
), 'configure weight 1'
resps = get_resps_sc()
assert resps[0] == 50, 'weight 1 0'
assert resps[1] == 50, 'weight 1 1'
assert 'success' in client.conf(
{
"127.0.0.1:7081": {"weight": 3},
"127.0.0.1:7083": {"weight": 2},
},
'upstreams/one/servers',
), 'configure weight 2'
resps = get_resps_sc()
assert resps[0] == 60, 'weight 2 0'
assert resps[2] == 40, 'weight 2 1'
def test_upstreams_rr_weight_rational():
def set_weights(w1, w2):
assert 'success' in client.conf(
{
"127.0.0.1:7081": {"weight": w1},
"127.0.0.1:7082": {"weight": w2},
},
'upstreams/one/servers',
), 'configure weights'
def check_reqs(w1, w2, reqs=10):
resps = get_resps_sc(req=reqs)
assert resps[0] == reqs * w1 / (w1 + w2), 'weight 1'
assert resps[1] == reqs * w2 / (w1 + w2), 'weight 2'
def check_weights(w1, w2):
set_weights(w1, w2)
check_reqs(w1, w2)
check_weights(0, 1)
check_weights(0, 999999.0123456)
check_weights(1, 9)
check_weights(100000, 900000)
check_weights(1, 0.25)
check_weights(1, 0.25)
check_weights(0.2, 0.8)
check_weights(1, 1.5)
check_weights(1e-3, 1e-3)
check_weights(1e-20, 1e-20)
check_weights(1e4, 1e4)
check_weights(1000000, 1000000)
set_weights(0.25, 0.25)
assert 'success' in client.conf_delete(
'upstreams/one/servers/127.0.0.1:7081/weight'
), 'delete weight'
check_reqs(1, 0.25)
assert 'success' in client.conf(
{
"127.0.0.1:7081": {"weight": 0.1},
"127.0.0.1:7082": {"weight": 1},
"127.0.0.1:7083": {"weight": 0.9},
},
'upstreams/one/servers',
), 'configure weights'
resps = get_resps_sc(req=20)
assert resps[0] == 1, 'weight 3 1'
assert resps[1] == 10, 'weight 3 2'
assert resps[2] == 9, 'weight 3 3'
def test_upstreams_rr_independent():
def sum_resps(*args):
sum_r = [0] * len(args[0])
for arg in args:
sum_r = [x + y for x, y in zip(sum_r, arg)]
return sum_r
resps = get_resps_sc(req=30, port=7090)
assert resps[0] == 15, 'dep two before 0'
assert resps[1] == 15, 'dep two before 1'
resps = get_resps_sc(req=30)
assert resps[0] == 15, 'dep one before 0'
assert resps[1] == 15, 'dep one before 1'
assert 'success' in client.conf(
'2', 'upstreams/two/servers/127.0.0.1:7081/weight'
), 'configure dep weight'
resps = get_resps_sc(req=30, port=7090)
assert resps[0] == 20, 'dep two 0'
assert resps[1] == 10, 'dep two 1'
resps = get_resps_sc(req=30)
assert resps[0] == 15, 'dep one 0'
assert resps[1] == 15, 'dep one 1'
assert 'success' in client.conf(
'1', 'upstreams/two/servers/127.0.0.1:7081/weight'
), 'configure dep weight 1'
r_one, r_two = [0, 0], [0, 0]
for _ in range(10):
r_one = sum_resps(r_one, get_resps(req=10))
r_two = sum_resps(r_two, get_resps(req=10, port=7090))
assert sum(r_one) == 100, 'dep one mix sum'
assert abs(r_one[0] - r_one[1]) <= client.cpu_count, 'dep one mix'
assert sum(r_two) == 100, 'dep two mix sum'
assert abs(r_two[0] - r_two[1]) <= client.cpu_count, 'dep two mix'
def test_upstreams_rr_delay():
delayed_dir = f'{option.test_dir}/python/delayed'
assert 'success' in client.conf(
{
"listeners": {
"*:7080": {"pass": "upstreams/one"},
"*:7081": {"pass": "routes"},
"*:7082": {"pass": "routes"},
},
"upstreams": {
"one": {
"servers": {
"127.0.0.1:7081": {},
"127.0.0.1:7082": {},
},
},
},
"routes": [
{
"match": {"destination": "*:7081"},
"action": {"pass": "applications/delayed"},
},
{
"match": {"destination": "*:7082"},
"action": {"return": 201},
},
],
"applications": {
"delayed": {
"type": client.get_application_type(),
"processes": {"spare": 0},
"path": delayed_dir,
"working_directory": delayed_dir,
"module": "wsgi",
}
},
},
), 'upstreams initial configuration'
req = 50
socks = []
for i in range(req):
delay = 1 if i % 5 == 0 else 0
sock = client.get(
headers={
'Host': 'localhost',
'Content-Length': '0',
'X-Delay': str(delay),
'Connection': 'close',
},
no_recv=True,
)
socks.append(sock)
resps = [0, 0]
for i in range(req):
resp = client.recvall(socks[i]).decode()
socks[i].close()
m = re.search(r'HTTP/1.1 20(\d)', resp)
assert m is not None, 'status'
resps[int(m.group(1))] += 1
assert sum(resps) == req, 'delay sum'
assert abs(resps[0] - resps[1]) <= client.cpu_count, 'delay'
def test_upstreams_rr_active_req():
conns = 5
socks = []
socks2 = []
for _ in range(conns):
sock = client.get(no_recv=True)
socks.append(sock)
sock2 = client.http(
b"""POST / HTTP/1.1
Host: localhost
Content-Length: 10
Connection: close
""",
no_recv=True,
raw=True,
)
socks2.append(sock2)
# Send one more request and read response to make sure that previous
# requests had enough time to reach server.
assert client.get()['body'] == ''
assert 'success' in client.conf(
{"127.0.0.1:7083": {"weight": 2}},
'upstreams/one/servers',
), 'active req new server'
assert 'success' in client.conf_delete(
'upstreams/one/servers/127.0.0.1:7083'
), 'active req server remove'
assert 'success' in client.conf_delete(
'listeners/*:7080'
), 'delete listener'
assert 'success' in client.conf_delete(
'upstreams/one'
), 'active req upstream remove'
for i in range(conns):
assert (
client.http(b'', sock=socks[i], raw=True)['body'] == ''
), 'active req GET'
assert (
client.http(b"""0123456789""", sock=socks2[i], raw=True)['body']
== ''
), 'active req POST'
def test_upstreams_rr_bad_server():
assert 'success' in client.conf(
{"weight": 1}, 'upstreams/one/servers/127.0.0.1:7084'
), 'configure bad server'
resps = get_resps_sc(req=30)
assert resps[0] == 10, 'bad server 0'
assert resps[1] == 10, 'bad server 1'
assert sum(resps) == 20, 'bad server sum'
def test_upstreams_rr_pipeline():
resps = get_resps_sc()
assert resps[0] == 50, 'pipeline 0'
assert resps[1] == 50, 'pipeline 1'
def test_upstreams_rr_post():
resps = [0, 0]
for _ in range(50):
resps[client.get()['status'] % 10] += 1
resps[client.post(body='0123456789')['status'] % 10] += 1
assert sum(resps) == 100, 'post sum'
assert abs(resps[0] - resps[1]) <= client.cpu_count, 'post'
def test_upstreams_rr_unix(temp_dir):
addr_0 = f'{temp_dir}/sock_0'
addr_1 = f'{temp_dir}/sock_1'
assert 'success' in client.conf(
{
"*:7080": {"pass": "upstreams/one"},
f"unix:{addr_0}": {"pass": "routes/one"},
f"unix:{addr_1}": {"pass": "routes/two"},
},
'listeners',
), 'configure listeners unix'
assert 'success' in client.conf(
{f"unix:{addr_0}": {}, f"unix:{addr_1}": {}},
'upstreams/one/servers',
), 'configure servers unix'
resps = get_resps_sc()
assert resps[0] == 50, 'unix 0'
assert resps[1] == 50, 'unix 1'
def test_upstreams_rr_ipv6():
assert 'success' in client.conf(
{
"*:7080": {"pass": "upstreams/one"},
"[::1]:7081": {"pass": "routes/one"},
"[::1]:7082": {"pass": "routes/two"},
},
'listeners',
), 'configure listeners ipv6'
assert 'success' in client.conf(
{"[::1]:7081": {}, "[::1]:7082": {}}, 'upstreams/one/servers'
), 'configure servers ipv6'
resps = get_resps_sc()
assert resps[0] == 50, 'ipv6 0'
assert resps[1] == 50, 'ipv6 1'
def test_upstreams_rr_servers_empty():
assert 'success' in client.conf(
{}, 'upstreams/one/servers'
), 'configure servers empty'
assert client.get()['status'] == 502, 'servers empty'
assert 'success' in client.conf(
{"127.0.0.1:7081": {"weight": 0}}, 'upstreams/one/servers'
), 'configure servers empty one'
assert client.get()['status'] == 502, 'servers empty one'
assert 'success' in client.conf(
{
"127.0.0.1:7081": {"weight": 0},
"127.0.0.1:7082": {"weight": 0},
},
'upstreams/one/servers',
), 'configure servers empty two'
assert client.get()['status'] == 502, 'servers empty two'
def test_upstreams_rr_invalid():
assert 'error' in client.conf({}, 'upstreams'), 'upstreams empty'
assert 'error' in client.conf({}, 'upstreams/one'), 'named upstreams empty'
assert 'error' in client.conf(
{}, 'upstreams/one/servers/127.0.0.1'
), 'invalid address'
assert 'error' in client.conf(
{}, 'upstreams/one/servers/127.0.0.1:7081/blah'
), 'invalid server option'
def check_weight(w):
assert 'error' in client.conf(
w, 'upstreams/one/servers/127.0.0.1:7081/weight'
), 'invalid weight option'
check_weight({})
check_weight('-1')
check_weight('1.')
check_weight('1.1.')
check_weight('.')
check_weight('.01234567890123')
check_weight('1000001')
check_weight('2e6')
