Add real transaction to E2E setup and clean up query tests

Author: incrypto32

tests/e2e/conftest.py

@@ -15,6 +15,7 @@
 from .helpers.process_manager import (
     get_free_port,
     mine_blocks,
+    send_eth,
     spawn_ampd,
     spawn_anvil,
     wait_for_ampd_ready,
@@ -57,6 +58,13 @@ def _setup_amp_stack(num_blocks: int = 10, end_block: str | None = 'latest'):
         }
 
         anvil_proc, anvil_url = spawn_anvil(log_dir)
+        # Send a transaction so transactions table has data
+        send_eth(
+            anvil_url,
+            from_addr='0xf39Fd6e51aad88F6F4ce6aB8827279cffFb92266',
+            to_addr='0x70997970C51812dc3A010C7d01b50e0d17dc79C8',
+            value_wei=10**18,
+        )
         mine_blocks(anvil_url, num_blocks)
 
         config_path = generate_ampd_config(

tests/e2e/helpers/process_manager.py

@@ -105,6 +105,22 @@ def mine_blocks(anvil_url: str, count: int) -> None:
         resp.raise_for_status()
 
 
+def send_eth(anvil_url: str, from_addr: str, to_addr: str, value_wei: int) -> str:
+    """Send an ETH transfer on Anvil. Returns the transaction hash."""
+    with httpx.Client() as client:
+        resp = client.post(
+            anvil_url,
+            json={
+                'jsonrpc': '2.0',
+                'method': 'eth_sendTransaction',
+                'params': [{'from': from_addr, 'to': to_addr, 'value': hex(value_wei)}],
+                'id': 1,
+            },
+        )
+        resp.raise_for_status()
+        return resp.json()['result']
+
+
 def evm_snapshot(anvil_url: str) -> str:
     """Take a snapshot of the current anvil state. Returns snapshot ID."""
     with httpx.Client() as client:

tests/e2e/test_queries.py

@@ -5,13 +5,16 @@
 
 pytestmark = pytest.mark.e2e
 
+# genesis + 1 tx block (auto-mined) + 10 mined blocks
+EXPECTED_BLOCKS = 12
+
 
 def test_query_blocks(e2e_client):
     table = e2e_client.sql('SELECT block_num, hash FROM anvil.blocks ORDER BY block_num').to_arrow()
 
-    assert len(table) == 11
+    assert len(table) == EXPECTED_BLOCKS
     block_nums = table.column('block_num').to_pylist()
-    assert block_nums == list(range(11))
+    assert block_nums == list(range(EXPECTED_BLOCKS))
     assert all(h is not None for h in table.column('hash').to_pylist())
 
 
@@ -29,29 +32,19 @@ def test_blocks_schema(e2e_client):
 
 def test_query_transactions(e2e_client):
     table = e2e_client.sql('SELECT * FROM anvil.transactions LIMIT 5').to_arrow()
+    assert len(table) >= 1
     expected_cols = {'block_num', 'tx_hash', 'from', 'to', 'value', 'gas_used'}
     assert expected_cols.issubset(set(table.column_names))
 
 
-def test_query_logs(e2e_client):
-    table = e2e_client.sql('SELECT * FROM anvil.logs LIMIT 5').to_arrow()
-    expected_cols = {'block_num', 'log_index', 'address', 'data'}
-    assert expected_cols.issubset(set(table.column_names))
-
-
 def test_query_with_where_clause(e2e_client):
     table = e2e_client.sql('SELECT block_num FROM anvil.blocks WHERE block_num > 5 ORDER BY block_num').to_arrow()
 
-    assert len(table) == 5
-    assert table.column('block_num').to_pylist() == [6, 7, 8, 9, 10]
-
-
-def test_query_with_aggregation(e2e_client):
-    table = e2e_client.sql('SELECT COUNT(*) AS cnt FROM anvil.blocks').to_arrow()
-    assert table.column('cnt').to_pylist()[0] == 11
+    assert len(table) == EXPECTED_BLOCKS - 6
+    assert table.column('block_num').to_pylist() == list(range(6, EXPECTED_BLOCKS))
 
 
 def test_isolated_server(amp_test_server):
     """Verify function-scoped stack works on its own ports."""
     table = amp_test_server.client.sql('SELECT COUNT(*) AS cnt FROM anvil.blocks').to_arrow()
-    assert table.column('cnt').to_pylist()[0] == 11
+    assert table.column('cnt').to_pylist()[0] == EXPECTED_BLOCKS

tests/e2e/test_streaming.py

@@ -79,16 +79,22 @@ def test_reorg_detection(reorg_server):
     flight_port = reorg_server.ports['flight']
     client = reorg_server.client
 
-    # Snapshot at block 10
+    # Find current chain tip
+    tip = client.sql('SELECT MAX(block_num) AS tip FROM anvil.blocks').to_arrow()
+    tip_block = tip.column('tip').to_pylist()[0]
+
     snapshot_id = evm_snapshot(anvil_url)
 
-    # Mine blocks 11-15, wait for ingestion
+    # Mine 5 blocks past the tip, wait for ingestion
     mine_blocks(anvil_url, 5)
-    wait_for_block(flight_port, 15)
+    first_new = tip_block + 1
+    last_new = tip_block + 5
+    wait_for_block(flight_port, last_new)
 
     # Capture pre-reorg hashes
     pre_reorg = client.sql(
-        'SELECT block_num, hash FROM anvil.blocks WHERE block_num >= 11 ORDER BY block_num'
+        f'SELECT block_num, hash FROM anvil.blocks '
+        f'WHERE block_num >= {first_new} ORDER BY block_num'
     ).to_arrow()
     assert len(pre_reorg) == 5
     pre_hashes = pre_reorg.column('hash').to_pylist()
@@ -97,15 +103,16 @@ def test_reorg_detection(reorg_server):
     # writes new blocks whose prev_hash won't match the stored hash,
     # triggering fork detection and re-materialization.
     evm_revert(anvil_url, snapshot_id)
-    mine_blocks(anvil_url, 10)  # blocks 11-20, different hashes
+    mine_blocks(anvil_url, 10)
 
     # Wait for ampd to detect reorg and re-ingest
     timeout = 30
     post_hashes = None
     start = time.monotonic()
     while time.monotonic() - start < timeout:
         post_reorg = client.sql(
-            'SELECT block_num, hash FROM anvil.blocks WHERE block_num >= 11 AND block_num <= 15 ORDER BY block_num'
+            f'SELECT block_num, hash FROM anvil.blocks '
+            f'WHERE block_num >= {first_new} AND block_num <= {last_new} ORDER BY block_num'
         ).to_arrow()
         if len(post_reorg) == 5:
             post_hashes = post_reorg.column('hash').to_pylist()