Initial version of stealthii
An aiohttp/httpx-shaped async HTTP client backed by a single, persistent, stealth-patched Chromium instance via Playwright. get/post/put/patch/ delete/head/download use context.request (real Chrome TLS/HTTP2 fingerprint, no tab needed); render()/screenshot()/page() escalate to full JS-executing page rendering for targets that need a JS challenge solved. Supports named sessions for cookie persistence/warm-up, and `async with stealth as page` to claim a tab directly, task-safe for concurrent use on one shared instance. Context/browser recycling bounds cookie and on-disk cache growth in a long-lived process. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_0116i7dYLNZTXNR7Lqf439bV
This commit is contained in:
@@ -0,0 +1,75 @@
|
||||
# retoor <retoor@molodetz.nl>
|
||||
import pytest
|
||||
|
||||
from stealthii import Stealthii, StealthResponse
|
||||
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
|
||||
|
||||
|
||||
def test_context_slot_starts_fresh():
|
||||
slot = _ContextSlot(context=object())
|
||||
assert slot.count == 0
|
||||
assert slot.patched is False
|
||||
@@ -0,0 +1,61 @@
|
||||
# retoor <retoor@molodetz.nl>
|
||||
import pytest
|
||||
|
||||
from stealthii import Stealthii
|
||||
|
||||
|
||||
@pytest.mark.online
|
||||
@pytest.mark.asyncio
|
||||
async def test_sannysoft_detection_suite_is_clean():
|
||||
"""Regression guard: every check on bot.sannysoft.com must pass under
|
||||
our stealth patches. If this starts failing, a Chromium upgrade likely
|
||||
changed something the STEALTH_INIT_JS patches assume."""
|
||||
s = Stealthii()
|
||||
try:
|
||||
async with s.page() as pg:
|
||||
await pg.goto('https://bot.sannysoft.com/', wait_until='networkidle', timeout=20000)
|
||||
rows = await pg.evaluate(
|
||||
"""() => Array.from(document.querySelectorAll('table tr')).map(tr =>
|
||||
Array.from(tr.querySelectorAll('td')).map(td => td.innerText.trim()).join(' | ')
|
||||
)"""
|
||||
)
|
||||
finally:
|
||||
await s.shutdown()
|
||||
|
||||
failures = [r for r in rows if 'FAIL' in r or 'failed' in r]
|
||||
assert not failures, f'stealth detection regressions: {failures}'
|
||||
|
||||
|
||||
@pytest.mark.online
|
||||
@pytest.mark.asyncio
|
||||
async def test_async_with_instance_claims_isolated_concurrent_tabs():
|
||||
"""`async with stealth as pg` must behave like `stealth.page()`, and be
|
||||
safe to use concurrently from multiple tasks on one shared instance."""
|
||||
import asyncio
|
||||
|
||||
s = Stealthii()
|
||||
|
||||
async def worker():
|
||||
async with s as pg:
|
||||
await pg.goto('https://example.com')
|
||||
return await pg.title()
|
||||
|
||||
try:
|
||||
titles = await asyncio.gather(*[worker() for _ in range(3)])
|
||||
finally:
|
||||
await s.shutdown()
|
||||
assert titles == ['Example Domain'] * 3
|
||||
|
||||
|
||||
@pytest.mark.online
|
||||
@pytest.mark.asyncio
|
||||
async def test_fast_path_uses_real_chromium_fingerprint():
|
||||
"""The no-tab get()/post() fast path must carry a genuine Chromium TLS
|
||||
fingerprint, distinct from a plain Python HTTP stack's."""
|
||||
s = Stealthii()
|
||||
try:
|
||||
info = await s.fingerprint_info()
|
||||
finally:
|
||||
await s.shutdown()
|
||||
assert info['hash']
|
||||
assert info['mode'] == 'ja4'
|
||||
Reference in New Issue
Block a user