136 lines
4.3 KiB
Python
136 lines
4.3 KiB
Python
"""unit test: a session's close cannot park the caller indefinitely
|
|
|
|
run from the repo root:
|
|
uv run pytest tests/test_session_close_budget.py
|
|
|
|
offline — no network. curl_cffi's close awaits the multi handle with no timeout
|
|
of its own, so a wedged transfer would park whatever closes the session,
|
|
including an `async with` exit. These build a real session and wedge the
|
|
backend's close, asserting the bound lives on the session itself so every call
|
|
site gets it without arranging anything.
|
|
"""
|
|
|
|
import asyncio
|
|
import logging
|
|
|
|
from aioproxies import AioProxies
|
|
|
|
from monitor_utils import sessions
|
|
from monitor_utils.sessions import BoundedCloseSession, get_default_session
|
|
|
|
|
|
def _proxies(tmp_path) -> AioProxies:
|
|
"""A one-entry pool in the host:port:user:pass format monitors ship."""
|
|
proxy_file = tmp_path / "proxies.txt"
|
|
proxy_file.write_text("proxy.example.io:8080:user:secret\n")
|
|
return AioProxies.from_file(path=str(proxy_file))
|
|
|
|
|
|
async def _wedge_close(session) -> asyncio.Event:
|
|
"""Replace the backend's close with one that never returns.
|
|
|
|
Builds the backend session first so TLSSession.close actually reaches the
|
|
backend rather than short-circuiting on an unbuilt session.
|
|
"""
|
|
|
|
session.session # force lazy construction
|
|
started = asyncio.Event()
|
|
|
|
async def wedged(inner_session):
|
|
started.set()
|
|
await asyncio.sleep(3600)
|
|
|
|
session.backend.close = wedged
|
|
return started
|
|
|
|
|
|
async def _really_close(session, original_close):
|
|
"""Close for real, so the test does not leak the curl handle it allocated."""
|
|
session.backend.close = original_close
|
|
await session.close()
|
|
|
|
|
|
def test_close_is_abandoned_when_it_exceeds_the_budget(monkeypatch, caplog, tmp_path):
|
|
"""A close that never returns is abandoned and logged, not awaited forever."""
|
|
|
|
monkeypatch.setattr(sessions, "SESSION_CLOSE_TIMEOUT_SECONDS", 0.05)
|
|
proxies = _proxies(tmp_path)
|
|
|
|
async def run():
|
|
session = get_default_session(proxies, timeout=8)
|
|
original_close = session.backend.close
|
|
started = await _wedge_close(session)
|
|
|
|
# wait_for is the assertion: an unbounded close never returns here
|
|
await asyncio.wait_for(session.close(), timeout=5)
|
|
|
|
assert started.is_set()
|
|
await _really_close(session, original_close)
|
|
|
|
with caplog.at_level(logging.ERROR, logger="monitor_utils.sessions"):
|
|
asyncio.run(run())
|
|
|
|
assert any(
|
|
record.message == "session close exceeded budget, abandoning session"
|
|
for record in caplog.records
|
|
)
|
|
|
|
|
|
def test_async_with_exit_is_bounded(monkeypatch, caplog, tmp_path):
|
|
"""The bound applies through `async with`, which is how monitors close."""
|
|
|
|
monkeypatch.setattr(sessions, "SESSION_CLOSE_TIMEOUT_SECONDS", 0.05)
|
|
proxies = _proxies(tmp_path)
|
|
|
|
async def run():
|
|
session = get_default_session(proxies, timeout=8)
|
|
original_close = session.backend.close
|
|
started = await _wedge_close(session)
|
|
|
|
async with session:
|
|
pass
|
|
|
|
assert started.is_set()
|
|
await _really_close(session, original_close)
|
|
|
|
with caplog.at_level(logging.ERROR, logger="monitor_utils.sessions"):
|
|
asyncio.run(asyncio.wait_for(run(), timeout=5))
|
|
|
|
assert any(
|
|
record.message == "session close exceeded budget, abandoning session"
|
|
for record in caplog.records
|
|
)
|
|
|
|
|
|
def test_healthy_close_completes_and_is_not_logged(caplog, tmp_path):
|
|
"""The normal path closes fully and reports nothing."""
|
|
|
|
proxies = _proxies(tmp_path)
|
|
|
|
async def run():
|
|
session = get_default_session(proxies, timeout=8)
|
|
session.session # force lazy construction so close has work to do
|
|
await session.close()
|
|
return session
|
|
|
|
with caplog.at_level(logging.ERROR, logger="monitor_utils.sessions"):
|
|
session = asyncio.run(run())
|
|
|
|
assert session._is_closed()
|
|
assert not caplog.records
|
|
|
|
|
|
def test_get_default_session_returns_a_bounded_session(tmp_path):
|
|
"""The factory hands out the bounded type, not a bare TLSSession."""
|
|
|
|
proxies = _proxies(tmp_path)
|
|
|
|
async def run():
|
|
session = get_default_session(proxies, timeout=8)
|
|
try:
|
|
return isinstance(session, BoundedCloseSession)
|
|
finally:
|
|
await session.close()
|
|
|
|
assert asyncio.run(run())
|