mirror of
https://github.com/vladkens/twscrape.git
synced 2026-10-15 18:54:00 -04:00
1096 lines
32 KiB
Python
1096 lines
32 KiB
Python
import json
|
|
from collections import OrderedDict
|
|
from contextlib import aclosing
|
|
|
|
import pytest
|
|
|
|
import twscrape.queue_client as queue_client_module
|
|
from twscrape.account import Account
|
|
from twscrape.accounts_pool import AccountsPool
|
|
from twscrape.http import ConnectError, HttpMethod, NetworkError, Response
|
|
from twscrape.queue_client import (
|
|
GqlFeaturesOutdatedError,
|
|
QueueClient,
|
|
ReqParams,
|
|
XClIdGenStore,
|
|
)
|
|
from twscrape.utils import utc
|
|
from twscrape.xclid import XClIdAccountError, XClIdGen, XClIdParseError
|
|
|
|
from .mock_http import MockClient
|
|
|
|
URL = "https://example.com/api"
|
|
CF = tuple[AccountsPool, QueueClient, MockClient]
|
|
REAL_XCLID_STORE_GET = XClIdGenStore.get.__func__
|
|
|
|
|
|
async def get_locked(pool: AccountsPool) -> set[str]:
|
|
rep = await pool.get_all()
|
|
return {x.username for x in rep if x.locks.get("SearchTimeline", None) is not None}
|
|
|
|
|
|
async def get_inactive(pool: AccountsPool) -> set[str]:
|
|
rep = await pool.get_all()
|
|
return {x.username for x in rep if not x.active}
|
|
|
|
|
|
async def test_lock_account_when_used(client_fixture: CF):
|
|
pool, client, mock = client_fixture
|
|
|
|
locked = await get_locked(pool)
|
|
assert len(locked) == 0
|
|
|
|
await client.__aenter__()
|
|
locked = await get_locked(pool)
|
|
assert len(locked) == 1
|
|
assert "user1" in locked
|
|
|
|
mock.add_response(json={"foo": "bar"})
|
|
rep = await client.get(URL)
|
|
assert rep is not None
|
|
assert rep.json() == {"foo": "bar"}
|
|
|
|
locked = await get_locked(pool)
|
|
assert len(locked) == 1
|
|
assert "user1" in locked
|
|
|
|
await client.__aexit__(None, None, None)
|
|
locked = await get_locked(pool)
|
|
assert len(locked) == 0
|
|
|
|
|
|
async def test_do_not_switch_account_on_200(client_fixture: CF):
|
|
pool, client, mock = client_fixture
|
|
|
|
await client.__aenter__()
|
|
locked1 = await get_locked(pool)
|
|
assert len(locked1) == 1
|
|
|
|
for x in range(3):
|
|
mock.add_response(json={"foo": x})
|
|
rep = await client.get(URL)
|
|
assert rep is not None
|
|
assert rep.json() == {"foo": x}
|
|
|
|
locked2 = await get_locked(pool)
|
|
assert locked1 == locked2
|
|
|
|
await client.__aexit__(None, None, None)
|
|
assert len(await get_locked(pool)) == 0
|
|
|
|
|
|
async def test_switch_acc_on_http_error(client_fixture: CF):
|
|
pool, client, mock = client_fixture
|
|
|
|
await client.__aenter__()
|
|
locked1 = await get_locked(pool)
|
|
assert len(locked1) == 1
|
|
|
|
mock.add_response(status_code=403, json={})
|
|
mock.add_response(json={"foo": "2"})
|
|
|
|
rep = await client.get(URL)
|
|
assert rep is not None
|
|
assert rep.json() == {"foo": "2"}
|
|
|
|
locked2 = await get_locked(pool)
|
|
assert len(locked2) == 2
|
|
|
|
await client.__aexit__(None, None, None)
|
|
locked3 = await get_locked(pool)
|
|
assert len(locked3) == 1
|
|
assert locked1 == locked3
|
|
|
|
|
|
async def test_retry_with_same_acc_on_network_error(client_fixture: CF, monkeypatch):
|
|
pool, client, mock = client_fixture
|
|
|
|
sleeps = []
|
|
|
|
async def fake_sleep(secs):
|
|
sleeps.append(secs)
|
|
|
|
monkeypatch.setattr("twscrape.queue_client.asyncio.sleep", fake_sleep)
|
|
|
|
await client.__aenter__()
|
|
locked1 = await get_locked(pool)
|
|
assert len(locked1) == 1
|
|
|
|
mock.add_exception(NetworkError("timeout"))
|
|
mock.add_response(json={"foo": "2"})
|
|
|
|
rep = await client.get(URL)
|
|
assert rep is not None
|
|
assert rep.json() == {"foo": "2"}
|
|
assert sleeps == [2]
|
|
|
|
assert await get_locked(pool) == locked1
|
|
|
|
username = getattr(rep, "__username", None)
|
|
assert username is not None
|
|
|
|
|
|
async def test_network_error_rotates_account_after_3_failures(client_fixture: CF, monkeypatch):
|
|
pool, client, mock = client_fixture
|
|
|
|
sleeps = []
|
|
|
|
async def fake_sleep(secs):
|
|
sleeps.append(secs)
|
|
|
|
monkeypatch.setattr("twscrape.queue_client.asyncio.sleep", fake_sleep)
|
|
|
|
await client.__aenter__()
|
|
assert await get_locked(pool) == {"user1"}
|
|
|
|
for _ in range(3):
|
|
mock.add_exception(NetworkError("timeout"))
|
|
mock.add_response(json={"ok": True})
|
|
|
|
rep = await client.get(URL)
|
|
assert rep is not None
|
|
assert rep.json() == {"ok": True}
|
|
assert getattr(rep, "__username", None) == "user2"
|
|
assert sleeps == [2, 4]
|
|
|
|
# user1 is short-locked (~60s transport lock, not the 15-min unknown-error lock)
|
|
user1 = next(x for x in await pool.get_all() if x.username == "user1")
|
|
lock_secs = (user1.locks["SearchTimeline"] - utc.now()).total_seconds()
|
|
assert 0 < lock_secs <= 61
|
|
|
|
await client.__aexit__(None, None, None)
|
|
|
|
|
|
async def test_network_error_counter_resets_on_account_change(client_fixture: CF, monkeypatch):
|
|
pool, client, mock = client_fixture
|
|
|
|
async def fake_sleep(secs):
|
|
pass
|
|
|
|
monkeypatch.setattr("twscrape.queue_client.asyncio.sleep", fake_sleep)
|
|
|
|
await client.__aenter__()
|
|
|
|
# 3 failures rotate user1 -> user2, which must get its own 3 tries: if the
|
|
# counter carried over, the first failure on user2 would rotate it too and
|
|
# exhaust the pool (request would return None)
|
|
for _ in range(5):
|
|
mock.add_exception(NetworkError("timeout"))
|
|
mock.add_response(json={"ok": True})
|
|
|
|
rep = await client.get(URL)
|
|
assert rep is not None
|
|
assert rep.json() == {"ok": True}
|
|
assert getattr(rep, "__username", None) == "user2"
|
|
|
|
await client.__aexit__(None, None, None)
|
|
|
|
|
|
async def test_ctx_closed_on_break(client_fixture: CF):
|
|
pool, client, mock = client_fixture
|
|
|
|
async def get_data_stream():
|
|
async with client as c:
|
|
counter = 0
|
|
while True:
|
|
counter += 1
|
|
check_retry = counter == 2
|
|
before_ctx = c.ctx
|
|
|
|
if check_retry:
|
|
mock.add_response(status_code=403, json={"counter": counter})
|
|
mock.add_response(json={"counter": counter})
|
|
else:
|
|
mock.add_response(json={"counter": counter})
|
|
|
|
rep = await c.get(URL)
|
|
|
|
if check_retry:
|
|
assert before_ctx != c.ctx
|
|
elif before_ctx is not None:
|
|
assert before_ctx == c.ctx
|
|
|
|
assert rep is not None
|
|
assert rep.json() == {"counter": counter}
|
|
yield rep.json()["counter"]
|
|
|
|
if counter == 9:
|
|
return
|
|
|
|
async with aclosing(get_data_stream()) as gen:
|
|
async for x in gen:
|
|
if x == 3:
|
|
break
|
|
|
|
assert client.ctx is None
|
|
|
|
|
|
async def test_queue_client_passes_effective_proxy_to_xclid(pool_mock: AccountsPool, monkeypatch):
|
|
mock = MockClient()
|
|
seen = {}
|
|
|
|
class FakeXClIdGen:
|
|
def calc(self, *args, **kwargs):
|
|
return "mocked-clid"
|
|
|
|
async def fake_get(cls, username, proxy=None, cookies=None, fresh=False):
|
|
seen["username"] = username
|
|
seen["proxy"] = proxy
|
|
seen["fresh"] = fresh
|
|
return FakeXClIdGen()
|
|
|
|
monkeypatch.setattr(Account, "make_client", lambda self, proxy=None: mock)
|
|
monkeypatch.setattr(XClIdGenStore, "get", classmethod(fake_get))
|
|
|
|
await pool_mock.add_account(
|
|
"user1",
|
|
"pass1",
|
|
"email1",
|
|
"email_pass1",
|
|
proxy="127.0.0.1:7897",
|
|
cookies="auth_token=token1; ct0=csrf1",
|
|
)
|
|
await pool_mock.set_active("user1", True)
|
|
|
|
client = QueueClient(pool_mock, "SearchTimeline")
|
|
await client.__aenter__()
|
|
|
|
mock.add_response(json={"ok": True})
|
|
rep = await client.get(URL)
|
|
|
|
assert rep is not None
|
|
assert seen == {
|
|
"username": "user1",
|
|
"proxy": "http://127.0.0.1:7897",
|
|
"fresh": False,
|
|
}
|
|
|
|
await client.__aexit__(None, None, None)
|
|
|
|
|
|
# --- ConnectError ---
|
|
|
|
|
|
async def test_connect_error_cools_account_and_rotates_after_3_retries(
|
|
client_fixture: CF, monkeypatch
|
|
):
|
|
pool, client, mock = client_fixture
|
|
|
|
sleeps = []
|
|
|
|
async def fake_sleep(secs):
|
|
sleeps.append(secs)
|
|
|
|
monkeypatch.setattr("twscrape.queue_client.asyncio.sleep", fake_sleep)
|
|
|
|
await client.__aenter__()
|
|
assert await get_locked(pool) == {"user1"}
|
|
|
|
mock.add_exception(ConnectError("refused"))
|
|
mock.add_exception(ConnectError("refused"))
|
|
mock.add_exception(ConnectError("refused"))
|
|
mock.add_response(json={"ok": True})
|
|
|
|
rep = await client.get(URL)
|
|
assert rep is not None
|
|
assert rep.json() == {"ok": True}
|
|
assert getattr(rep, "__username", None) == "user2"
|
|
assert sleeps == [2, 4]
|
|
|
|
# user1 got the short transport cooldown lock, not the 15min unknown-error lock
|
|
user1 = next(x for x in await pool.get_all() if x.username == "user1")
|
|
lock_secs = (user1.locks["SearchTimeline"] - utc.now()).total_seconds()
|
|
assert 0 < lock_secs <= 61
|
|
|
|
await client.__aexit__(None, None, None)
|
|
|
|
|
|
async def test_connect_error_recovers_before_3_retries(client_fixture: CF, monkeypatch):
|
|
pool, client, mock = client_fixture
|
|
|
|
async def fake_sleep(secs):
|
|
pass
|
|
|
|
monkeypatch.setattr("twscrape.queue_client.asyncio.sleep", fake_sleep)
|
|
|
|
await client.__aenter__()
|
|
|
|
mock.add_exception(ConnectError("refused"))
|
|
mock.add_exception(ConnectError("refused"))
|
|
mock.add_response(json={"ok": True})
|
|
|
|
rep = await client.get(URL)
|
|
assert rep is not None
|
|
assert rep.json() == {"ok": True}
|
|
assert getattr(rep, "__username", None) == "user1"
|
|
|
|
await client.__aexit__(None, None, None)
|
|
|
|
|
|
async def test_alternating_categories_trip_total_safety_net(client_fixture: CF, monkeypatch):
|
|
pool, client, mock = client_fixture
|
|
|
|
async def fake_sleep(secs):
|
|
pass
|
|
|
|
monkeypatch.setattr("twscrape.queue_client.asyncio.sleep", fake_sleep)
|
|
|
|
await client.__aenter__()
|
|
assert await get_locked(pool) == {"user1"}
|
|
|
|
# neither category alone reaches its own limit (3), but alternating between
|
|
# them should still trip the combined total-failure safety net (4)
|
|
mock.add_exception(ConnectError("refused"))
|
|
mock.add_exception(RuntimeError("boom"))
|
|
mock.add_exception(ConnectError("refused"))
|
|
mock.add_exception(RuntimeError("boom"))
|
|
mock.add_response(json={"ok": True})
|
|
|
|
rep = await client.get(URL)
|
|
assert rep is not None
|
|
assert getattr(rep, "__username", None) == "user2"
|
|
|
|
user1 = next(x for x in await pool.get_all() if x.username == "user1")
|
|
assert "SearchTimeline" in user1.locks
|
|
|
|
await client.__aexit__(None, None, None)
|
|
|
|
|
|
async def test_transport_error_retry_budget_is_per_account(client_fixture: CF, monkeypatch):
|
|
pool, client, mock = client_fixture
|
|
|
|
async def fake_sleep(secs):
|
|
pass
|
|
|
|
monkeypatch.setattr("twscrape.queue_client.asyncio.sleep", fake_sleep)
|
|
|
|
await client.__aenter__()
|
|
|
|
# 3 failures rotate user1 -> user2, which must get its own fresh budget: if the
|
|
# counter carried over, the first failure on user2 would immediately rotate it
|
|
# too and exhaust the whole (2-account) pool.
|
|
for _ in range(5):
|
|
mock.add_exception(NetworkError("timeout"))
|
|
mock.add_response(json={"ok": True})
|
|
|
|
rep = await client.get(URL)
|
|
assert rep is not None
|
|
assert rep.json() == {"ok": True}
|
|
assert getattr(rep, "__username", None) == "user2"
|
|
|
|
await client.__aexit__(None, None, None)
|
|
|
|
await client.__aexit__(None, None, None)
|
|
|
|
|
|
# --- Rate limit ---
|
|
|
|
|
|
async def test_rate_limit_locks_account_and_switches(client_fixture: CF):
|
|
pool, client, mock = client_fixture
|
|
await client.__aenter__()
|
|
assert "user1" in await get_locked(pool)
|
|
|
|
future_ts = 9999999999
|
|
mock.add_response(
|
|
headers={"x-rate-limit-remaining": "0", "x-rate-limit-reset": str(future_ts)},
|
|
)
|
|
mock.add_response(json={"ok": True})
|
|
|
|
rep = await client.get(URL)
|
|
assert rep is not None
|
|
assert rep.json() == {"ok": True}
|
|
|
|
user1 = next(x for x in await pool.get_all() if x.username == "user1")
|
|
assert user1.locks.get("SearchTimeline") is not None
|
|
|
|
await client.__aexit__(None, None, None)
|
|
|
|
|
|
# --- Ban / inactive ---
|
|
|
|
|
|
async def test_ban_88_marks_account_inactive(client_fixture: CF):
|
|
pool, client, mock = client_fixture
|
|
await client.__aenter__()
|
|
|
|
mock.add_response(
|
|
json={"errors": [{"code": 88, "message": "Rate limit exceeded"}]},
|
|
headers={"x-rate-limit-remaining": "1"},
|
|
)
|
|
mock.add_response(json={"ok": True})
|
|
|
|
rep = await client.get(URL)
|
|
assert rep is not None
|
|
assert "user1" in await get_inactive(pool)
|
|
|
|
await client.__aexit__(None, None, None)
|
|
|
|
|
|
async def test_ban_326_marks_account_inactive(client_fixture: CF):
|
|
pool, client, mock = client_fixture
|
|
await client.__aenter__()
|
|
|
|
mock.add_response(
|
|
json={"errors": [{"code": 326, "message": "Authorization: Denied by access control"}]},
|
|
)
|
|
mock.add_response(json={"ok": True})
|
|
|
|
rep = await client.get(URL)
|
|
assert rep is not None
|
|
assert "user1" in await get_inactive(pool)
|
|
|
|
await client.__aexit__(None, None, None)
|
|
|
|
|
|
async def test_ban_32_marks_account_inactive(client_fixture: CF):
|
|
pool, client, mock = client_fixture
|
|
await client.__aenter__()
|
|
|
|
mock.add_response(
|
|
json={"errors": [{"code": 32, "message": "Could not authenticate you"}]},
|
|
)
|
|
mock.add_response(json={"ok": True})
|
|
|
|
rep = await client.get(URL)
|
|
assert rep is not None
|
|
assert "user1" in await get_inactive(pool)
|
|
|
|
await client.__aexit__(None, None, None)
|
|
|
|
|
|
async def test_403_no_errors_marks_account_inactive(client_fixture: CF):
|
|
pool, client, mock = client_fixture
|
|
await client.__aenter__()
|
|
|
|
mock.add_response(status_code=403, json={})
|
|
mock.add_response(json={"ok": True})
|
|
|
|
rep = await client.get(URL)
|
|
assert rep is not None
|
|
assert "user1" in await get_inactive(pool)
|
|
|
|
await client.__aexit__(None, None, None)
|
|
|
|
|
|
# --- Cloudflare / HTML block ---
|
|
|
|
|
|
async def test_cloudflare_block_returns_none(client_fixture: CF):
|
|
pool, client, mock = client_fixture
|
|
await client.__aenter__()
|
|
|
|
mock.add_response(
|
|
status_code=403,
|
|
text="<html>blocked</html>",
|
|
headers={"content-type": "text/html", "cf-ray": "abc123"},
|
|
)
|
|
|
|
rep = await client.get(URL)
|
|
assert rep is None
|
|
|
|
await client.__aexit__(None, None, None)
|
|
|
|
|
|
async def test_html_block_without_cf_returns_none(client_fixture: CF):
|
|
pool, client, mock = client_fixture
|
|
await client.__aenter__()
|
|
|
|
mock.add_response(
|
|
status_code=503,
|
|
text="<html>error</html>",
|
|
headers={"content-type": "text/html"},
|
|
)
|
|
|
|
rep = await client.get(URL)
|
|
assert rep is None
|
|
|
|
await client.__aexit__(None, None, None)
|
|
|
|
|
|
# --- _check_rep branches ---
|
|
|
|
|
|
async def test_131_with_user_data_continues(client_fixture: CF):
|
|
pool, client, mock = client_fixture
|
|
await client.__aenter__()
|
|
|
|
mock.add_response(
|
|
json={
|
|
"errors": [{"code": 131, "message": "Dependency: Internal error"}],
|
|
"data": {"user": {"id": "123"}},
|
|
}
|
|
)
|
|
|
|
rep = await client.get(URL)
|
|
assert rep is not None
|
|
|
|
await client.__aexit__(None, None, None)
|
|
|
|
|
|
async def test_131_without_user_data_continues(client_fixture: CF):
|
|
pool, client, mock = client_fixture
|
|
await client.__aenter__()
|
|
|
|
mock.add_response(json={"errors": [{"code": 131, "message": "Dependency: Internal error"}]})
|
|
|
|
rep = await client.get(URL)
|
|
assert rep is not None
|
|
|
|
await client.__aexit__(None, None, None)
|
|
|
|
|
|
async def test_gql_features_outdated_raises(client_fixture: CF):
|
|
pool, client, mock = client_fixture
|
|
await client.__aenter__()
|
|
|
|
mock.add_response(
|
|
json={"errors": [{"code": 336, "message": "The following features cannot be null: foo"}]}
|
|
)
|
|
|
|
with pytest.raises(GqlFeaturesOutdatedError):
|
|
await client.get(URL)
|
|
|
|
await client.__aexit__(None, None, None)
|
|
|
|
# the account must survive: not deactivated, not left locked
|
|
assert await get_inactive(pool) == set()
|
|
assert await get_locked(pool) == set()
|
|
|
|
|
|
async def test_gql_features_self_healed_and_retried(client_fixture: CF, monkeypatch):
|
|
pool, client, mock = client_fixture
|
|
await client.__aenter__()
|
|
|
|
sent: list[dict] = []
|
|
original = mock.request
|
|
|
|
async def spy(method: HttpMethod, url: str, **kwargs) -> Response:
|
|
sent.append(dict(kwargs.get("params") or {}))
|
|
return await original(method, url, **kwargs)
|
|
|
|
monkeypatch.setattr(mock, "request", spy)
|
|
|
|
mock.add_response(
|
|
json={
|
|
"errors": [{"code": 336, "message": "The following features cannot be null: foo, bar"}]
|
|
}
|
|
)
|
|
mock.add_response(json={"data": {"user": {"id": 1}}})
|
|
|
|
params: ReqParams = {"variables": "{}", "features": json.dumps({"baz": True})}
|
|
rep = await client.get(URL, params=params)
|
|
assert rep is not None
|
|
|
|
# the flags X named were added and the request replayed
|
|
assert len(sent) == 2
|
|
assert json.loads(sent[0]["features"]) == {"baz": True}
|
|
assert json.loads(sent[1]["features"]) == {"baz": True, "foo": True, "bar": True}
|
|
|
|
await client.__aexit__(None, None, None)
|
|
|
|
# the account must survive: not deactivated, not left locked
|
|
assert await get_inactive(pool) == set()
|
|
assert await get_locked(pool) == set()
|
|
|
|
|
|
async def test_gql_features_self_heal_persists_across_requests(client_fixture: CF, monkeypatch):
|
|
_pool, client, mock = client_fixture
|
|
await client.__aenter__()
|
|
|
|
sent = []
|
|
original = mock.request
|
|
|
|
async def spy(method: HttpMethod, url: str, **kwargs) -> Response:
|
|
sent.append(json.loads(kwargs["params"]["features"]))
|
|
return await original(method, url, **kwargs)
|
|
|
|
monkeypatch.setattr(mock, "request", spy)
|
|
|
|
mock.add_response(
|
|
json={"errors": [{"code": 336, "message": "The following features cannot be null: foo"}]}
|
|
)
|
|
mock.add_response(json={"data": {"user": {"id": 1}}})
|
|
mock.add_response(json={"data": {"user": {"id": 1}}})
|
|
|
|
assert await client.get(URL, params={"variables": "{}", "features": "{}"}) is not None
|
|
assert await client.get(URL, params={"variables": "{}", "features": "{}"}) is not None
|
|
assert sent == [{}, {"foo": True}, {"foo": True}]
|
|
|
|
await client.__aexit__(None, None, None)
|
|
|
|
|
|
async def test_gql_features_self_heal_retries_only_once(client_fixture: CF):
|
|
_pool, client, mock = client_fixture
|
|
await client.__aenter__()
|
|
|
|
# X keeps rejecting even with the flags set — give up rather than loop
|
|
for _ in range(2):
|
|
mock.add_response(
|
|
json={
|
|
"errors": [{"code": 336, "message": "The following features cannot be null: foo"}]
|
|
}
|
|
)
|
|
|
|
with pytest.raises(GqlFeaturesOutdatedError):
|
|
await client.get(URL, params={"variables": "{}", "features": "{}"})
|
|
|
|
await client.__aexit__(None, None, None)
|
|
|
|
|
|
async def test_gql_features_outdated_raises_when_not_first(client_fixture: CF):
|
|
_pool, client, mock = client_fixture
|
|
await client.__aenter__()
|
|
|
|
mock.add_response(
|
|
json={
|
|
"errors": [
|
|
{"code": 999, "message": "Some unfamiliar error"},
|
|
{"code": 336, "message": "The following features cannot be null: foo"},
|
|
]
|
|
}
|
|
)
|
|
|
|
with pytest.raises(GqlFeaturesOutdatedError):
|
|
await client.get(URL)
|
|
|
|
await client.__aexit__(None, None, None)
|
|
|
|
|
|
async def test_missing_status_error_ignored(client_fixture: CF):
|
|
pool, client, mock = client_fixture
|
|
await client.__aenter__()
|
|
|
|
mock.add_response(
|
|
json={"errors": [{"code": -1, "message": "_Missing: No status found with that ID"}]}
|
|
)
|
|
|
|
rep = await client.get(URL)
|
|
assert rep is not None
|
|
|
|
await client.__aexit__(None, None, None)
|
|
|
|
|
|
async def test_authorization_error_200_ignored(client_fixture: CF):
|
|
pool, client, mock = client_fixture
|
|
await client.__aenter__()
|
|
|
|
mock.add_response(
|
|
json={"errors": [{"code": -1, "message": "Authorization: Denied by unknown rule"}]}
|
|
)
|
|
|
|
rep = await client.get(URL)
|
|
assert rep is not None
|
|
|
|
await client.__aexit__(None, None, None)
|
|
|
|
|
|
async def test_unknown_error_warned_once(client_fixture: CF, monkeypatch):
|
|
_pool, client, mock = client_fixture
|
|
logs = []
|
|
monkeypatch.setattr(queue_client_module.LogOnce, "seen", OrderedDict())
|
|
monkeypatch.setattr(
|
|
queue_client_module.logger, "log", lambda level, msg: logs.append((level, msg))
|
|
)
|
|
await client.__aenter__()
|
|
|
|
for _ in range(2):
|
|
mock.add_response(json={"errors": [{"code": 999, "message": "Some unfamiliar error"}]})
|
|
rep = await client.get(URL)
|
|
assert rep is not None
|
|
|
|
assert [level for level, _msg in logs] == ["WARNING"]
|
|
|
|
await client.__aexit__(None, None, None)
|
|
|
|
|
|
async def test_unknown_error_statuses_are_logged_separately(client_fixture: CF, monkeypatch):
|
|
_pool, client, mock = client_fixture
|
|
logs = []
|
|
monkeypatch.setattr(queue_client_module.LogOnce, "seen", OrderedDict())
|
|
monkeypatch.setattr(
|
|
queue_client_module.logger, "log", lambda level, msg: logs.append((level, msg))
|
|
)
|
|
await client.__aenter__()
|
|
|
|
for status_code in (200, 500):
|
|
mock.add_response(
|
|
status_code=status_code,
|
|
json={"errors": [{"code": 999, "message": "Some unfamiliar error"}]},
|
|
)
|
|
rep = await client.get(URL)
|
|
assert rep is not None
|
|
|
|
assert [level for level, _msg in logs] == ["WARNING", "WARNING"]
|
|
await client.__aexit__(None, None, None)
|
|
|
|
|
|
async def test_loadshed_without_data_retries_same_account(client_fixture: CF, monkeypatch):
|
|
_pool, client, mock = client_fixture
|
|
sleeps = []
|
|
|
|
async def fake_sleep(secs):
|
|
sleeps.append(secs)
|
|
|
|
monkeypatch.setattr("twscrape.queue_client.asyncio.sleep", fake_sleep)
|
|
await client.__aenter__()
|
|
|
|
mock.add_response(json={"errors": [{"code": -1, "message": "LoadShed: Unspecified"}]})
|
|
mock.add_response(json={"ok": True})
|
|
|
|
rep = await client.get(URL)
|
|
|
|
assert rep is not None
|
|
assert rep.json() == {"ok": True}
|
|
assert getattr(rep, "__username", None) == "user1"
|
|
assert sleeps == [2]
|
|
await client.__aexit__(None, None, None)
|
|
|
|
|
|
async def test_api_error_with_data_is_throttled(client_fixture: CF, monkeypatch):
|
|
_pool, client, mock = client_fixture
|
|
logs = []
|
|
monkeypatch.setattr(queue_client_module.LogOnce, "pending", OrderedDict())
|
|
monkeypatch.setattr(
|
|
queue_client_module.logger, "log", lambda level, msg: logs.append((level, msg))
|
|
)
|
|
await client.__aenter__()
|
|
|
|
errors = [
|
|
{"code": -1, "message": "Dependency: Unspecified"},
|
|
{"code": -1, "message": "DeadlineExceeded: Unspecified"},
|
|
{"code": -1, "message": "LoadShed: Unspecified"},
|
|
]
|
|
for response_errors in (errors, list(reversed(errors))):
|
|
mock.add_response(
|
|
json={
|
|
"data": {"search_by_raw_query": {"items": [1]}},
|
|
"errors": response_errors,
|
|
}
|
|
)
|
|
rep = await client.get(URL)
|
|
assert rep is not None
|
|
|
|
assert [level for level, _msg in logs] == ["DEBUG"]
|
|
assert "user1" not in logs[0][1]
|
|
assert "SearchTimeline" in logs[0][1]
|
|
await client.__aexit__(None, None, None)
|
|
|
|
|
|
@pytest.mark.parametrize("data", [None, {}, {"user": None}])
|
|
async def test_api_error_without_useful_data_is_warned(client_fixture: CF, monkeypatch, data):
|
|
_pool, client, mock = client_fixture
|
|
logs = []
|
|
monkeypatch.setattr(queue_client_module.LogOnce, "seen", OrderedDict())
|
|
monkeypatch.setattr(
|
|
queue_client_module.logger, "log", lambda level, msg: logs.append((level, msg))
|
|
)
|
|
await client.__aenter__()
|
|
|
|
mock.add_response(
|
|
json={
|
|
"data": data,
|
|
"errors": [{"code": -1, "message": "Dependency: Unspecified"}],
|
|
}
|
|
)
|
|
rep = await client.get(URL)
|
|
|
|
assert rep is not None
|
|
assert [level for level, _msg in logs] == ["WARNING"]
|
|
await client.__aexit__(None, None, None)
|
|
|
|
|
|
async def test_unhandled_status_code_locks_and_retries(client_fixture: CF):
|
|
pool, client, mock = client_fixture
|
|
await client.__aenter__()
|
|
|
|
mock.add_response(status_code=500, json={})
|
|
mock.add_response(json={"ok": True})
|
|
|
|
rep = await client.get(URL)
|
|
assert rep is not None
|
|
assert rep.json() == {"ok": True}
|
|
|
|
user1 = next(x for x in await pool.get_all() if x.username == "user1")
|
|
assert "SearchTimeline" in user1.locks
|
|
assert int(user1.locks["SearchTimeline"].timestamp()) > utc.ts() + 60 * 10
|
|
|
|
await client.__aexit__(None, None, None)
|
|
|
|
|
|
async def test_no_active_accounts_returns_none(pool_mock: AccountsPool):
|
|
client = QueueClient(pool_mock, "SearchTimeline")
|
|
rep = await client.get(URL)
|
|
assert rep is None
|
|
|
|
|
|
async def test_unknown_exception_retries_then_locks_account(client_fixture: CF):
|
|
pool, client, mock = client_fixture
|
|
await client.__aenter__()
|
|
|
|
mock.add_exception(RuntimeError("boom"))
|
|
mock.add_exception(RuntimeError("boom"))
|
|
mock.add_exception(RuntimeError("boom"))
|
|
mock.add_response(json={"ok": True})
|
|
|
|
rep = await client.get(URL)
|
|
assert rep is not None
|
|
assert rep.json() == {"ok": True}
|
|
|
|
user1 = next(x for x in await pool.get_all() if x.username == "user1")
|
|
assert "SearchTimeline" in user1.locks
|
|
|
|
await client.__aexit__(None, None, None)
|
|
|
|
|
|
async def test_invalid_json_body_falls_back_to_raw_text(client_fixture: CF):
|
|
pool, client, mock = client_fixture
|
|
await client.__aenter__()
|
|
|
|
mock.add_invalid_json_response(text="not-json")
|
|
rep = await client.get(URL)
|
|
assert rep is not None
|
|
|
|
await client.__aexit__(None, None, None)
|
|
|
|
|
|
async def test_close_ctx_noop_when_ctx_is_none(pool_mock: AccountsPool):
|
|
client = QueueClient(pool_mock, "SearchTimeline")
|
|
# ctx is None — _close_ctx must be a no-op
|
|
await client._close_ctx()
|
|
|
|
|
|
async def test_404_retries_exhaust_and_abort(client_fixture: CF):
|
|
from unittest.mock import patch
|
|
|
|
pool, client, mock = client_fixture
|
|
await client.__aenter__()
|
|
|
|
mock.add_response(status_code=404, json={})
|
|
mock.add_response(status_code=404, json={})
|
|
mock.add_response(status_code=404, json={})
|
|
|
|
with patch("twscrape.queue_client.asyncio.sleep"):
|
|
rep = await client.get(URL)
|
|
assert rep is None
|
|
|
|
await client.__aexit__(None, None, None)
|
|
|
|
|
|
async def test_404_refreshes_generator_and_retries(client_fixture: CF, monkeypatch):
|
|
pool, client, mock = client_fixture
|
|
fresh_values = []
|
|
|
|
class FakeXClIdGen:
|
|
def calc(self, *args, **kwargs):
|
|
return "mocked-clid"
|
|
|
|
async def fake_get(cls, username, proxy=None, cookies=None, fresh=False):
|
|
fresh_values.append(fresh)
|
|
return FakeXClIdGen()
|
|
|
|
async def no_sleep(_):
|
|
return None
|
|
|
|
monkeypatch.setattr(XClIdGenStore, "get", classmethod(fake_get))
|
|
monkeypatch.setattr("twscrape.queue_client.asyncio.sleep", no_sleep)
|
|
mock.add_response(status_code=404, json={})
|
|
mock.add_response(json={"ok": True})
|
|
|
|
rep = await client.get(URL)
|
|
|
|
assert rep is not None
|
|
assert rep.json() == {"ok": True}
|
|
assert fresh_values == [False, True]
|
|
await client.__aexit__(None, None, None)
|
|
|
|
|
|
async def test_xclid_store_reuses_username_across_proxies_and_refreshes(monkeypatch):
|
|
created = []
|
|
|
|
async def fake_create(proxy=None, cookies=None):
|
|
gen = object()
|
|
created.append((proxy, cookies, gen))
|
|
return gen
|
|
|
|
monkeypatch.setattr(XClIdGen, "create", staticmethod(fake_create))
|
|
XClIdGenStore.items.clear()
|
|
try:
|
|
first = await REAL_XCLID_STORE_GET(
|
|
XClIdGenStore,
|
|
"user1",
|
|
proxy="http://proxy-one.test",
|
|
cookies={"auth_token": "one", "ct0": "one"},
|
|
)
|
|
reused = await REAL_XCLID_STORE_GET(
|
|
XClIdGenStore,
|
|
"user1",
|
|
proxy="http://proxy-two.test",
|
|
cookies={"auth_token": "two", "ct0": "two"},
|
|
)
|
|
refreshed = await REAL_XCLID_STORE_GET(
|
|
XClIdGenStore,
|
|
"user1",
|
|
proxy="http://proxy-two.test",
|
|
cookies={"auth_token": "two", "ct0": "two"},
|
|
fresh=True,
|
|
)
|
|
finally:
|
|
XClIdGenStore.items.clear()
|
|
|
|
assert reused is first
|
|
assert refreshed is not first
|
|
assert created == [
|
|
(
|
|
"http://proxy-one.test",
|
|
{"auth_token": "one", "ct0": "one"},
|
|
first,
|
|
),
|
|
(
|
|
"http://proxy-two.test",
|
|
{"auth_token": "two", "ct0": "two"},
|
|
refreshed,
|
|
),
|
|
]
|
|
|
|
|
|
async def test_queue_client_passes_account_cookies_to_xclid(pool_mock: AccountsPool, monkeypatch):
|
|
mock = MockClient()
|
|
seen = {}
|
|
|
|
class FakeXClIdGen:
|
|
def calc(self, *args, **kwargs):
|
|
return "mocked-clid"
|
|
|
|
async def fake_get(cls, username, proxy=None, cookies=None, fresh=False):
|
|
seen["username"] = username
|
|
seen["cookies"] = cookies
|
|
seen["fresh"] = fresh
|
|
return FakeXClIdGen()
|
|
|
|
monkeypatch.setattr(Account, "make_client", lambda self, proxy=None: mock)
|
|
monkeypatch.setattr(XClIdGenStore, "get", classmethod(fake_get))
|
|
|
|
await pool_mock.add_account(
|
|
"user1",
|
|
"pass1",
|
|
"email1",
|
|
"email_pass1",
|
|
cookies='{"auth_token": "abc", "ct0": "def"}',
|
|
)
|
|
await pool_mock.set_active("user1", True)
|
|
|
|
client = QueueClient(pool_mock, "SearchTimeline")
|
|
await client.__aenter__()
|
|
|
|
mock.add_response(json={"ok": True})
|
|
rep = await client.get(URL)
|
|
|
|
assert rep is not None
|
|
assert seen == {
|
|
"username": "user1",
|
|
"cookies": {"auth_token": "abc", "ct0": "def"},
|
|
"fresh": False,
|
|
}
|
|
|
|
await client.__aexit__(None, None, None)
|
|
|
|
|
|
async def test_missing_required_cookies_deactivates_and_rotates(
|
|
pool_mock: AccountsPool, monkeypatch
|
|
):
|
|
mock = MockClient()
|
|
monkeypatch.setattr(Account, "make_client", lambda self, proxy=None: mock)
|
|
pool_mock._order_by = "username"
|
|
await pool_mock.add_account("user1", "pass1", "email1", "email_pass1", cookies="ct0=csrf1")
|
|
await pool_mock.set_active("user1", True)
|
|
await pool_mock.add_account(
|
|
"user2",
|
|
"pass2",
|
|
"email2",
|
|
"email_pass2",
|
|
cookies="auth_token=token2; ct0=csrf2",
|
|
)
|
|
mock.add_response(json={"ok": True})
|
|
|
|
client = QueueClient(pool_mock, "SearchTimeline")
|
|
rep = await client.get(URL)
|
|
|
|
assert rep is not None
|
|
assert getattr(rep, "__username") == "user2"
|
|
user1 = await pool_mock.get("user1")
|
|
assert user1.active is False
|
|
assert user1.error_msg == "Missing authentication cookies"
|
|
await client.__aexit__(None, None, None)
|
|
|
|
|
|
async def test_xclid_account_error_locks_queue_and_rotates(pool_mock: AccountsPool, monkeypatch):
|
|
mock = MockClient()
|
|
monkeypatch.setattr(Account, "make_client", lambda self, proxy=None: mock)
|
|
pool_mock._order_by = "username"
|
|
for index in (1, 2):
|
|
await pool_mock.add_account(
|
|
f"user{index}",
|
|
f"pass{index}",
|
|
f"email{index}",
|
|
f"email_pass{index}",
|
|
cookies=f"auth_token=token{index}; ct0=csrf{index}",
|
|
)
|
|
|
|
original_get = XClIdGenStore.get.__func__
|
|
|
|
async def fake_get(cls, username, proxy=None, cookies=None, fresh=False):
|
|
if username == "user1":
|
|
raise XClIdAccountError("Logged-out X web app")
|
|
return await original_get(cls, username, proxy, cookies, fresh)
|
|
|
|
monkeypatch.setattr(XClIdGenStore, "get", classmethod(fake_get))
|
|
mock.add_response(json={"ok": True})
|
|
|
|
client = QueueClient(pool_mock, "SearchTimeline")
|
|
rep = await client.get(URL)
|
|
|
|
assert rep is not None
|
|
assert getattr(rep, "__username") == "user2"
|
|
user1 = await pool_mock.get("user1")
|
|
assert user1.active is True
|
|
assert int(user1.locks["SearchTimeline"].timestamp()) > utc.ts() + 60 * 10
|
|
await client.__aexit__(None, None, None)
|
|
|
|
|
|
async def test_xclid_parse_error_aborts_without_account_state_change(
|
|
pool_mock: AccountsPool, monkeypatch
|
|
):
|
|
mock = MockClient()
|
|
monkeypatch.setattr(Account, "make_client", lambda self, proxy=None: mock)
|
|
await pool_mock.add_account(
|
|
"user1",
|
|
"pass1",
|
|
"email1",
|
|
"email_pass1",
|
|
proxy="http://secret:[email protected]:7897",
|
|
cookies="auth_token=token1; ct0=csrf1",
|
|
)
|
|
messages = []
|
|
|
|
async def fake_get(cls, username, proxy=None, cookies=None, fresh=False):
|
|
raise XClIdParseError("Signing script not found (3/3 assets loaded)")
|
|
|
|
monkeypatch.setattr(XClIdGenStore, "get", classmethod(fake_get))
|
|
monkeypatch.setattr("twscrape.queue_client.logger.error", messages.append)
|
|
|
|
client = QueueClient(pool_mock, "SearchTimeline")
|
|
rep = await client.get(URL)
|
|
|
|
assert rep is None
|
|
user1 = await pool_mock.get("user1")
|
|
assert user1.active is True
|
|
assert "SearchTimeline" not in user1.locks
|
|
assert len(messages) == 1
|
|
assert "Signing script not found (3/3 assets loaded)" in messages[0]
|
|
assert "username=user1" in messages[0]
|
|
assert "queue=SearchTimeline" in messages[0]
|
|
assert "backend=unknown" in messages[0]
|
|
assert "proxy=True" in messages[0]
|
|
assert "secret" not in messages[0]
|