retoorandClaude Sonnet 5
423170e1bb
A Stealthii instance binds its lock, browser, and contexts to whichever asyncio event loop is running the first time it is used. Using it again from a different loop previously hung forever (the lock/semaphore/ Playwright transport are all bound to the dead first loop and never wake the waiting coroutine). Detected in rsearch's test suite, where pytest-asyncio's default per-test-function event loop broke the module-level Stealthii singleton shared across tests. Now raises StealthUnavailableError immediately with an actionable message. README documents the constraint and the pytest-asyncio config fix (asyncio_default_test_loop_scope = session). Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_0116i7dYLNZTXNR7Lqf439bV
85 lines
2.4 KiB
Python
85 lines
2.4 KiB
Python
# retoor <retoor@molodetz.nl>
|
|
import pytest
|
|
|
|
from stealthii import Stealthii, StealthResponse, StealthUnavailableError
|
|
from stealthii.client import _ContextSlot
|
|
|
|
|
|
def test_construction_is_lazy_and_cheap():
|
|
s = Stealthii()
|
|
assert s._browser is None
|
|
assert s._playwright is None
|
|
assert s._contexts == {}
|
|
|
|
|
|
def test_invalid_fingerprint_rejected():
|
|
with pytest.raises(ValueError):
|
|
Stealthii(fingerprint='ja5')
|
|
|
|
|
|
def test_defaults():
|
|
s = Stealthii()
|
|
assert s.fingerprint == 'ja4'
|
|
assert s.max_tabs == 4
|
|
assert s.max_retries == 3
|
|
assert s.recycle_context_after_count == 500
|
|
assert s.recycle_context_after_s == 1800.0
|
|
|
|
|
|
def test_build_body_json():
|
|
headers = {}
|
|
kwargs = Stealthii._build_body(headers, data=None, json={'a': 1}, files=None)
|
|
assert kwargs == {'data': '{"a": 1}'}
|
|
assert headers['Content-Type'] == 'application/json'
|
|
|
|
|
|
def test_build_body_form_dict():
|
|
kwargs = Stealthii._build_body({}, data={'x': '1'}, json=None, files=None)
|
|
assert kwargs == {'form': {'x': '1'}}
|
|
|
|
|
|
def test_build_body_raw_bytes():
|
|
kwargs = Stealthii._build_body({}, data=b'raw', json=None, files=None)
|
|
assert kwargs == {'data': b'raw'}
|
|
|
|
|
|
def test_build_body_none():
|
|
assert Stealthii._build_body({}, data=None, json=None, files=None) == {}
|
|
|
|
|
|
def test_build_body_multipart_with_files_and_fields():
|
|
kwargs = Stealthii._build_body(
|
|
{}, data={'field': 'value'}, json=None,
|
|
files={'upload': ('name.txt', b'content')},
|
|
)
|
|
multipart = kwargs['multipart']
|
|
assert multipart['field'] == 'value'
|
|
assert multipart['upload']['name'] == 'name.txt'
|
|
assert multipart['upload']['buffer'] == b'content'
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_stealth_response_shapes_like_aiohttp():
|
|
resp = StealthResponse(200, {'Content-Type': 'application/json'}, b'{"ok": true}', 'https://x/')
|
|
assert resp.status == 200
|
|
assert await resp.text() == '{"ok": true}'
|
|
assert await resp.json() == {'ok': True}
|
|
assert await resp.read() == b'{"ok": true}'
|
|
async with resp as r:
|
|
assert r is resp
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_lock_raises_clearly_on_event_loop_mismatch():
|
|
s = Stealthii()
|
|
s._get_lock()
|
|
s._loop = object()
|
|
with pytest.raises(StealthUnavailableError, match='different asyncio event loop'):
|
|
s._get_lock()
|
|
|
|
|
|
def test_context_slot_starts_fresh():
|
|
slot = _ContextSlot(context=object())
|
|
assert slot.count == 0
|
|
assert slot.patched is False
|