spb/coinexplorer Public MIT

1# Author: Simon-Pierre Boucher2# Mail: contact@spboucher.ai3"""Unit checks for the EVM ingestion loop (plain asserts — run with python)."""45import pathlib6import sys7import threading89sys.path.insert(0, str(pathlib.Path(__file__).parent.parent))1011from indexer import db12from indexer.decode import TRANSFER_TOPIC13from indexer.ingest import ChainIndexer14from indexer.rpc import RpcError1516USDT = "0xdac17f958d2ee523a2206206994597c13d831ec7"17CFG = {"rpcs": ["http://unused.invalid"], "max_range": 8, "confirmations": 2}18TOKENS = [{"symbol": "USDT", "address": USDT, "decimals": 6, "native": True}]192021def fake_log(block, log_index=0, amount=1_000_000):22    pad = lambda a: "0x" + a[2:].rjust(64, "0")23    return {24        "address": USDT,25        "topics": [TRANSFER_TOPIC, pad("0x" + "11" * 20), pad("0x" + "22" * 20)],26        "data": hex(amount),27        "blockNumber": hex(block),28        "blockHash": "0x" + f"{block:x}".rjust(64, "b"),29        "transactionHash": "0x" + f"{block:x}{log_index:x}".rjust(64, "a"),30        "logIndex": hex(log_index),31    }323334class FakePool:35    """Serves synthetic logs; refuses getLogs ranges wider than `cap`."""3637    def __init__(self, cap=2, blocks_with_logs=()):38        self.cap = cap39        self.blocks = set(blocks_with_logs)40        self.getlogs_calls = []41        self.rotations = 04243    def call(self, method, params=None):44        if method == "eth_getLogs":45            f = params[0]46            frm, to = int(f["fromBlock"], 16), int(f["toBlock"], 16)47            self.getlogs_calls.append((frm, to))48            if to - frm + 1 > self.cap:49                raise RpcError(-32005, "query returned more than 10000 results")50            return [fake_log(b) for b in range(frm, to + 1) if b in self.blocks]51        if method == "eth_getBlockByNumber":52            n = int(params[0], 16)53            return {"number": hex(n), "timestamp": hex(1_785_000_000 + n),54                    "hash": "0x" + f"{n:x}".rjust(64, "b"),55                    "parentHash": "0x" + f"{n - 1:x}".rjust(64, "b")}56        raise AssertionError(f"unexpected method {method}")5758    def batch(self, calls):59        return [self.call(m, p) for m, p in calls]6061    def rotate(self):62        self.rotations += 16364    @property65    def current_url(self):66        return "http://fake"6768    def stats(self):69        return []707172def make_indexer(pool):73    ix = ChainIndexer("testchain", CFG, TOKENS, ":memory:")74    ix.pool = pool75    ix.conn = db.connect(":memory:")76    return ix777879def test_try_range_halves_and_stays_contiguous():80    pool = FakePool(cap=2, blocks_with_logs={101, 103, 105})81    ix = make_indexer(pool)82    stop = threading.Event()83    covered, rows = 100, 084    while covered < 106:85        end, n = ix.try_range(covered + 1, 106, stop)86        assert end > covered, "no forward progress"87        covered, rows = end, rows + n88    assert covered == 10689    assert rows == 3, f"expected 3 transfers, got {rows}"90    assert ix.range == 2, "range should have halved to the node's cap"91    # every accepted sweep must be adjacent to the previous one (no gaps)92    accepted = [c for c in pool.getlogs_calls if c[1] - c[0] + 1 <= pool.cap]93    for (f1, t1), (f2, _) in zip(accepted, accepted[1:]):94        assert f2 == t1 + 1, f"gap between sweeps: ..{t1} then {f2}.."959697def test_decode_lands_in_db():98    pool = FakePool(cap=10, blocks_with_logs={42})99    ix = make_indexer(pool)100    n = ix.process_range(40, 45)101    assert n == 1102    row = ix.conn.execute("SELECT * FROM transfers").fetchone()103    assert row is not None104    (chain, block, _bh, _tx, _li, ts, token, symbol, frm, to, amount, decimals) = row105    assert (chain, block, token, symbol) == ("testchain", 42, USDT, "USDT")106    assert frm == "0x" + "11" * 20 and to == "0x" + "22" * 20107    assert amount == "1000000" and decimals == 6108    assert ts == 1_785_000_000 + 42109110111def test_rollback_removes_reorged_rows():112    pool = FakePool(cap=10, blocks_with_logs={10, 11, 12})113    ix = make_indexer(pool)114    ix.process_range(10, 12)115    db.set_cursor(ix.conn, "testchain", 12, "0xdead")116    db.rollback(ix.conn, "testchain", 10)117    left = [r[0] for r in ix.conn.execute("SELECT block FROM transfers")]118    assert left == [10], f"only block 10 should survive, got {left}"119    assert db.get_cursor(ix.conn, "testchain")[0] == 10120121122def test_parent_link_check():123    pool = FakePool(cap=10)124    ix = make_indexer(pool)125    good = "0x" + f"{50:x}".rjust(64, "b")   # header(51).parentHash126    assert ix.links_to_cursor(50, good)127    assert not ix.links_to_cursor(50, "0x" + "f" * 64)128129130if __name__ == "__main__":131    for name, fn in sorted(globals().items()):132        if name.startswith("test_"):133            fn()134            print(f"ok {name}")135    print("all ingest tests passed")136