spb/coinexplorer Public MIT

1# Author: Simon-Pierre Boucher2# Mail: contact@spboucher.ai3"""Unit checks for the RPC layer (plain asserts — run with python)."""45import pathlib6import sys7import time89sys.path.insert(0, str(pathlib.Path(__file__).parent.parent))1011from indexer.rpc import Endpoint, RpcPool, TokenBucket, _is_rate_limit121314def test_token_bucket_paces():15    b = TokenBucket(rate=10, burst=2)16    t0 = time.monotonic()17    for _ in range(6):  # burst of 2 free, then 4 paced at 10/s ≈ 0.4s18        b.acquire()19    elapsed = time.monotonic() - t020    assert 0.3 <= elapsed <= 1.0, f"pacing off: {elapsed:.2f}s"212223def test_rate_limit_classification():24    # failover-worthy (rate limits dressed as JSON-RPC errors)25    assert _is_rate_limit("rate limited")26    assert _is_rate_limit("Too many requests, slow down")27    assert _is_rate_limit("You've reached the usage limit for your current plan")28    # caller-worthy (query too big — range halving is the right response)29    assert not _is_rate_limit("query returned more than 10000 results")30    assert not _is_rate_limit("block range is too wide")31    assert not _is_rate_limit("Log response size exceeded")32    assert not _is_rate_limit("limit exceeded")  # BSC getLogs cap — ambiguous, caller decides333435def test_health_scoring_and_cooldown():36    e = Endpoint("https://example.invalid")37    assert e.score == 1.0 and e.available()38    e.record(False)39    e.record(False)40    assert e.score < 1.0 and not e.available()  # cooling down41    assert e.consec_fail == 242    e.cooldown_until = 0  # simulate recovery43    e.record(True, latency_ms=50)44    assert e.consec_fail == 0 and e.available()454647def test_pool_prefers_healthy_endpoint():48    pool = RpcPool(["https://a.invalid", "https://b.invalid"])49    a, b = pool.endpoints50    a.record(False)  # a benched + score down51    assert pool._pick() is b52    assert pool.current_url == "https://b.invalid"535455def test_endpoint_spec_dict():56    e = Endpoint({"url": "https://x.invalid", "rps": 1, "burst": 3})57    assert e.bucket.rate == 1 and e.bucket.burst == 3585960if __name__ == "__main__":61    for name, fn in sorted(globals().items()):62        if name.startswith("test_"):63            fn()64            print(f"ok {name}")65    print("all rpc tests passed")66