1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
|
import os
import sys
from contextlib import contextmanager
from pathlib import Path
from subprocess import PIPE
from subprocess import Popen
from subprocess import check_output
from tempfile import TemporaryDirectory
from test.utils import allpythons
from test.utils import austin
from test.utils import compress
from test.utils import has_pattern
from test.utils import requires_sudo
from test.utils import run_python
from test.utils import threads
from threading import Thread
from time import sleep
import psutil
import pytest
from requests import get
pytestmark = pytest.mark.skipif(
sys.platform == "win32", reason="Not supported on Windows"
)
UWSGI = Path(__file__).parent / "uwsgi"
@contextmanager
def uwsgi(app="app.py", port=9090, args=[], env=None):
with Popen(
[
"uwsgi",
"--http",
f":{port}",
"--wsgi-file",
UWSGI / app,
*args,
],
stdout=PIPE,
stderr=PIPE,
env=env or os.environ,
) as uw:
sleep(0.5)
assert uw.poll() is None, uw.stderr.read().decode()
pid = uw.pid
assert pid is not None, uw.stderr.read().decode()
try:
yield uw
finally:
uw.kill()
# Sledgehammer
for proc in psutil.process_iter():
if "uwsgi" in proc.name():
proc.terminate()
proc.wait()
@contextmanager
def venv(py, reqs):
with TemporaryDirectory() as tmp_path:
venv_path = Path(tmp_path) / ".venv"
p = run_python(py, "-m", "venv", str(venv_path))
p.wait(120)
assert 0 == p.returncode, "Virtual environment was created successfully"
env = os.environ.copy()
env["LD_LIBRARY_PATH"] = str(venv_path / "lib")
env["PATH"] = str(venv_path / "bin") + os.pathsep + env["PATH"]
check_output(
["python", "-m", "pip", "install", "-r", reqs, "--use-pep517"], env=env
)
yield env
@requires_sudo
@allpythons()
def test_uwsgi(py):
request_thread = Thread(target=get, args=("http://localhost:9090",))
with venv(py, reqs=Path(__file__).parent / "requirements-uwsgi.txt") as env:
with uwsgi(env=env) as uw:
request_thread.start()
result = austin("-x", "2", "-Cp", str(uw.pid))
assert has_pattern(result.stdout, "app.py:application:5"), compress(
result.stdout
)
request_thread.join()
@requires_sudo
@allpythons()
def test_uwsgi_multiprocess(py):
request_thread = Thread(target=get, args=("http://localhost:9091",))
with venv(py, reqs=Path(__file__).parent / "requirements-uwsgi.txt") as env:
with uwsgi(
port=9091, args=["--processes", "2", "--threads", "2"], env=env
) as uw:
request_thread.start()
result = austin("-x", "2", "-Cp", str(uw.pid))
assert has_pattern(result.stdout, "app.py:application:5"), compress(
result.stdout
)
ts = threads(result.stdout)
assert len(ts) >= 4, ts
assert len({p for p, _ in ts}) >= 2, ts
request_thread.join()
|
