import os
import platform
import resource
import subprocess
import sys
import threading
import pytest
sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent))
from eggfetch.compat.httpx import Client, AsyncClient, MockTransport, Response
from native_fixtures import local_http_server
def _load_resource_thresholds():
try:
import tomllib
except ModuleNotFoundError:
import tomli as tomllib
thresholds_path = os.path.join(
os.path.dirname(__file__), "..", "..", "..", "..",
"compat", "httpx", "0.28.1", "resource-thresholds.toml",
)
thresholds_path = os.path.normpath(thresholds_path)
if not os.path.exists(thresholds_path):
pytest.fail(f"Resource thresholds file not found: {thresholds_path}")
with open(thresholds_path, "rb") as f:
data = tomllib.load(f)
system = platform.system().lower()
if system not in data.get("platform", {}):
pytest.fail(f"No resource thresholds for platform: {system}")
return data["platform"][system]
def _get_fd_count():
if platform.system() == "Linux":
fd_dir = f"/proc/{os.getpid()}/fd"
if not os.path.isdir(fd_dir):
raise RuntimeError(f"FD directory not available: {fd_dir}")
return len(os.listdir(fd_dir))
elif platform.system() == "Darwin":
result = subprocess.run(
["lsof", "-p", str(os.getpid())],
capture_output=True, text=True, timeout=5,
)
return len(result.stdout.strip().split("\n")) - 1
else:
raise RuntimeError(f"FD counting not supported on {platform.system()}")
def _get_thread_count():
return threading.active_count()
def _get_rss_bytes():
usage = resource.getrusage(resource.RUSAGE_SELF)
if platform.system() == "Linux":
return usage.ru_maxrss * 1024
return usage.ru_maxrss
def _handler(request):
return Response(200, content=b"x" * 1024)
class TestResourceStability:
def test_fd_stability_under_repeated_requests(self):
thresholds = _load_resource_thresholds()
max_fd_delta = thresholds.get("max_fd_delta", 10)
try:
fd_before = _get_fd_count()
except RuntimeError as e:
pytest.fail(str(e))
with Client(transport=MockTransport(_handler)) as client:
for _ in range(50):
resp = client.get("http://testserver/")
assert resp.status_code == 200
fd_after = _get_fd_count()
delta = fd_after - fd_before
assert fd_after <= fd_before + max_fd_delta, (
f"FD leak detected: before={fd_before}, after={fd_after}, "
f"delta={delta}, threshold={max_fd_delta}"
)
def test_thread_stability_under_repeated_requests(self):
thresholds = _load_resource_thresholds()
max_thread_delta = thresholds.get("max_thread_delta", 5)
threads_before = _get_thread_count()
with Client(transport=MockTransport(_handler)) as client:
for _ in range(50):
resp = client.get("http://testserver/")
assert resp.status_code == 200
threads_after = _get_thread_count()
delta = threads_after - threads_before
assert threads_after <= threads_before + max_thread_delta, (
f"Thread leak detected: before={threads_before}, after={threads_after}, "
f"delta={delta}, threshold={max_thread_delta}"
)
def test_fd_stability_under_rapid_open_close(self):
thresholds = _load_resource_thresholds()
max_fd_delta = thresholds.get("max_fd_delta", 10)
try:
fd_before = _get_fd_count()
except RuntimeError as e:
pytest.fail(str(e))
for _ in range(20):
with Client(transport=MockTransport(_handler)) as client:
resp = client.get("http://testserver/")
assert resp.status_code == 200
fd_after = _get_fd_count()
delta = fd_after - fd_before
assert fd_after <= fd_before + max_fd_delta, (
f"FD leak on rapid open/close: before={fd_before}, after={fd_after}, "
f"delta={delta}, threshold={max_fd_delta}"
)
def test_thread_stability_under_rapid_open_close(self):
thresholds = _load_resource_thresholds()
max_thread_delta = thresholds.get("max_thread_delta", 5)
threads_before = _get_thread_count()
for _ in range(20):
with Client(transport=MockTransport(_handler)) as client:
resp = client.get("http://testserver/")
assert resp.status_code == 200
threads_after = _get_thread_count()
delta = threads_after - threads_before
assert threads_after <= threads_before + max_thread_delta, (
f"Thread leak on rapid open/close: before={threads_before}, "
f"after={threads_after}, delta={delta}, threshold={max_thread_delta}"
)
def test_memory_stability_under_repeated_requests(self):
thresholds = _load_resource_thresholds()
max_rss_growth = thresholds.get("max_rss_growth_bytes", 10 * 1024 * 1024)
rss_before = _get_rss_bytes()
with Client(transport=MockTransport(_handler)) as client:
for _ in range(100):
resp = client.get("http://testserver/")
_ = resp.content
rss_after = _get_rss_bytes()
growth = rss_after - rss_before
assert rss_after <= rss_before + max_rss_growth, (
f"Memory growth: before={rss_before}, after={rss_after}, "
f"growth={growth}, threshold={max_rss_growth}"
)
class TestEarlyExitResourceCleanup:
def test_exception_in_handler_releases_resources(self):
thresholds = _load_resource_thresholds()
max_fd_delta = thresholds.get("max_fd_delta", 10)
try:
fd_before = _get_fd_count()
except RuntimeError as e:
pytest.fail(str(e))
def error_handler(request):
raise RuntimeError("handler error")
with Client(transport=MockTransport(error_handler)) as client:
for _ in range(20):
with pytest.raises(RuntimeError, match="handler error"):
client.get("http://testserver/")
fd_after = _get_fd_count()
delta = fd_after - fd_before
assert fd_after <= fd_before + max_fd_delta, (
f"FD leak after errors: before={fd_before}, after={fd_after}, "
f"delta={delta}, threshold={max_fd_delta}"
)
def test_mixed_success_failure_stable(self):
thresholds = _load_resource_thresholds()
max_fd_delta = thresholds.get("max_fd_delta", 10)
try:
fd_before = _get_fd_count()
except RuntimeError as e:
pytest.fail(str(e))
call_count = [0]
def mixed_handler(request):
call_count[0] += 1
if call_count[0] % 3 == 0:
raise RuntimeError("intermittent error")
return Response(200)
with Client(transport=MockTransport(mixed_handler)) as client:
for _ in range(30):
try:
resp = client.get("http://testserver/")
assert resp.status_code == 200
except RuntimeError:
pass
fd_after = _get_fd_count()
delta = fd_after - fd_before
assert fd_after <= fd_before + max_fd_delta, (
f"FD leak under mixed load: before={fd_before}, after={fd_after}, "
f"delta={delta}, threshold={max_fd_delta}"
)
class TestConcurrentResourceStability:
def test_concurrent_sync_clients_stable(self):
thresholds = _load_resource_thresholds()
max_fd_delta = thresholds.get("max_fd_delta", 10)
try:
fd_before = _get_fd_count()
except RuntimeError as e:
pytest.fail(str(e))
def make_request():
with Client(transport=MockTransport(_handler)) as client:
for _ in range(10):
resp = client.get("http://testserver/")
assert resp.status_code == 200
threads = [threading.Thread(target=make_request) for _ in range(5)]
for t in threads:
t.start()
for t in threads:
t.join(timeout=10)
fd_after = _get_fd_count()
delta = fd_after - fd_before
assert fd_after <= fd_before + max_fd_delta * 2, (
f"FD leak under concurrency: before={fd_before}, after={fd_after}, "
f"delta={delta}, threshold={max_fd_delta * 2}"
)
class TestRealSocketResourceStability:
def test_fd_stability_real_socket(self):
thresholds = _load_resource_thresholds()
max_fd_delta = thresholds.get("max_fd_delta", 10)
try:
fd_before = _get_fd_count()
except RuntimeError as e:
pytest.fail(str(e))
with local_http_server() as (host, port):
url = f"http://{host}:{port}/json"
with Client(timeout=10) as client:
for _ in range(30):
resp = client.get(url)
assert resp.status_code == 200
fd_after = _get_fd_count()
delta = fd_after - fd_before
assert fd_after <= fd_before + max_fd_delta, (
f"FD leak on real sockets: before={fd_before}, after={fd_after}, "
f"delta={delta}, threshold={max_fd_delta}"
)
def test_thread_stability_real_socket(self):
thresholds = _load_resource_thresholds()
max_thread_delta = thresholds.get("max_thread_delta", 5)
threads_before = _get_thread_count()
with local_http_server() as (host, port):
url = f"http://{host}:{port}/json"
with Client(timeout=10) as client:
for _ in range(30):
resp = client.get(url)
assert resp.status_code == 200
threads_after = _get_thread_count()
delta = threads_after - threads_before
assert threads_after <= threads_before + max_thread_delta, (
f"Thread leak on real sockets: before={threads_before}, "
f"after={threads_after}, delta={delta}, threshold={max_thread_delta}"
)