Add E2E integration tests with real ampd + Anvil

.github/workflows/e2e-tests.yml

@@ -0,0 +1,66 @@
+name: E2E Tests
+permissions:
+  contents: read
+
+on:
+  push:
+    branches: [ main ]
+    paths:
+      - 'src/**/*.py'
+      - 'tests/e2e/**'
+      - '.github/workflows/e2e-tests.yml'
+  pull_request:
+    branches: [ main ]
+    paths:
+      - 'src/**/*.py'
+      - 'tests/e2e/**'
+      - '.github/workflows/e2e-tests.yml'
+  workflow_dispatch:
+
+jobs:
+  e2e-tests:
+    runs-on: ubuntu-latest
+    timeout-minutes: 15
+
+    steps:
+    - uses: actions/checkout@v4
+
+    - name: Install uv
+      uses: astral-sh/setup-uv@v4
+      with:
+        enable-cache: true
+        cache-dependency-glob: "**/pyproject.toml"
+
+    - name: Set up Python 3.12
+      run: uv python install 3.12
+
+    - name: Install dependencies
+      run: uv sync --all-groups
+
+    - name: Install PostgreSQL
+      run: |
+        sudo apt-get update
+        sudo apt-get install -y postgresql
+        echo "/usr/lib/postgresql/$(ls /usr/lib/postgresql/)/bin" >> "$GITHUB_PATH"
+
+    - name: Install Foundry (anvil)
+      uses: foundry-rs/foundry-toolchain@v1
+
+    - name: Install amp tools
+      run: |
+        gh release download --repo edgeandnode/amp --pattern 'ampd-linux-x86_64' --output /usr/local/bin/ampd
+        chmod +x /usr/local/bin/ampd
+      env:
+        GH_TOKEN: ${{ github.token }}
+
+    - name: Verify tools
+      run: |
+        anvil --version
+        ampd --version
+        initdb --version
+        postgres --version
+
+    - name: Run E2E tests
+      run: |
+        uv run pytest tests/e2e/ -v --log-cli-level=INFO -m "e2e" -x
+

.gitignore

@@ -2,6 +2,8 @@
 .env
 .test.env
 *.env
+.amp
+.amp_state
 
 # Kubernetes secrets (NEVER commit these!)
 k8s/secret.yaml

Makefile

@@ -54,6 +54,11 @@ test-performance:
 	@echo "🏇 Running performance tests..."
 	$(PYTHON) pytest tests/performance/ -m "performance" -v --log-cli-level=ERROR
 
+# E2E tests (require anvil, ampd, Docker)
+test-e2e:
+	@echo "Running E2E tests..."
+	$(PYTHON) pytest tests/e2e/ -m "e2e" -v --log-cli-level=INFO -x
+
 # Code quality (using your ruff config)
 lint:
 	@echo "🔍 Linting code..."
@@ -133,6 +138,7 @@ help:
 	@echo "  make test-redis               - Run Redis tests"
 	@echo "  make test-snowflake           - Run Snowflake tests"
 	@echo "  make test-performance         - Run performance tests"
+	@echo "  make test-e2e                 - Run E2E tests (require anvil, ampd, Docker)"
 	@echo "  make lint                     - Lint code with ruff"
 	@echo "  make format                   - Format code with ruff"
 	@echo "  make test-setup               - Start test databases"

pyproject.toml

@@ -123,6 +123,7 @@ markers = [
     "iceberg: Tests requiring Apache Iceberg",
     "snowflake: Tests requiring Snowflake",
     "cloud: Tests requiring cloud storage access",
+    "e2e: End-to-end tests (require anvil, ampd, Docker)",
 ]
 
 [tool.ruff]

tests/conftest.py

@@ -461,6 +461,7 @@ def pytest_configure(config):
     config.addinivalue_line('markers', 'snowflake: Tests requiring Snowflake')
     config.addinivalue_line('markers', 'lmdb: Tests requiring LMDB')
     config.addinivalue_line('markers', 'slow: Slow tests (> 30 seconds)')
+    config.addinivalue_line('markers', 'e2e: End-to-end tests (require anvil, ampd, Docker)')
 
 
 # Utility fixtures for mocking

tests/e2e/conftest.py

@@ -0,0 +1,157 @@
+"""E2E test fixtures: session-scoped and function-scoped ampd + Anvil infrastructure."""
+
+import logging
+import shutil
+import tempfile
+from dataclasses import dataclass
+from pathlib import Path
+
+import pytest
+
+from amp.client import Client
+
+from .helpers.config import copy_anvil_manifest, generate_ampd_config, generate_provider_toml
+from .helpers.dataset_manager import DatasetManager
+from .helpers.process_manager import (
+    get_free_port,
+    mine_blocks,
+    send_eth,
+    spawn_ampd,
+    spawn_anvil,
+    wait_for_ampd_ready,
+    wait_for_data_ready,
+)
+
+logger = logging.getLogger(__name__)
+
+
+def _check_deps():
+    missing = [b for b in ('anvil', 'ampd', 'initdb', 'postgres') if not shutil.which(b)]
+    if missing:
+        pytest.fail(f'Missing binaries: {", ".join(missing)}')
+
+
+@dataclass
+class AmpTestServer:
+    """An ampd + Anvil stack with a connected client."""
+
+    client: Client
+    anvil_url: str
+    admin_url: str
+    ports: dict
+
+
+def _setup_amp_stack(num_blocks: int = 10, end_block: str | None = 'latest'):
+    """Spin up anvil + ampd + register + deploy.
+
+    Returns (AmpTestServer, cleanup_fn).
+    """
+    temp_dir = Path(tempfile.mkdtemp(prefix='amp_e2e_'))
+    anvil_proc = None
+    ampd_proc = None
+    try:
+        log_dir = temp_dir / 'logs'
+        ports = {
+            'admin': get_free_port(),
+            'flight': get_free_port(),
+            'jsonl': get_free_port(),
+        }
+
+        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(
+            temp_dir,
+            ports['admin'],
+            ports['flight'],
+            ports['jsonl'],
+        )
+        generate_provider_toml(temp_dir, anvil_url)
+        manifest_path = copy_anvil_manifest(temp_dir)
+
+        ampd_proc = spawn_ampd(config_path, log_dir)
+        wait_for_ampd_ready(ports['admin'])
+
+        admin_url = f'http://127.0.0.1:{ports["admin"]}'
+        manager = DatasetManager(admin_url)
+        try:
+            manager.register_provider('anvil', temp_dir / 'provider_sources' / 'anvil.toml')
+            manager.register_dataset('_', 'anvil', manifest_path, '0.0.1')
+            manager.deploy_dataset('_', 'anvil', '0.0.1', end_block=end_block)
+            wait_for_data_ready(ports['flight'])
+        finally:
+            manager.close()
+    except Exception:
+        if ampd_proc:
+            ampd_proc.terminate()
+        if anvil_proc:
+            anvil_proc.terminate()
+        shutil.rmtree(temp_dir, ignore_errors=True)
+        raise
+
+    server = AmpTestServer(
+        client=Client(query_url=f'grpc://127.0.0.1:{ports["flight"]}'),
+        anvil_url=anvil_url,
+        admin_url=admin_url,
+        ports=ports,
+    )
+
+    def cleanup():
+        ampd_proc.terminate()
+        anvil_proc.terminate()
+        shutil.rmtree(temp_dir, ignore_errors=True)
+
+    return server, cleanup
+
+
+def _amp_fixture(**stack_kwargs):
+    """Yield an AmpTestServer, handling skip-check and cleanup."""
+    _check_deps()
+    server, cleanup = _setup_amp_stack(**stack_kwargs)
+    yield server
+    cleanup()
+
+
+# ---------------------------------------------------------------------------
+# Session-scoped fixtures (read-only query tests)
+# ---------------------------------------------------------------------------
+
+
+@pytest.fixture(scope='session')
+def e2e_server():
+    yield from _amp_fixture()
+
+
+@pytest.fixture(scope='session')
+def e2e_client(e2e_server):
+    return e2e_server.client
+
+
+@pytest.fixture(scope='session')
+def continuous_server():
+    """Session-scoped ampd + Anvil with continuous deploy."""
+    yield from _amp_fixture(end_block=None)
+
+
+# ---------------------------------------------------------------------------
+# Function-scoped fixtures (tests that mutate chain state)
+# ---------------------------------------------------------------------------
+
+
+@pytest.fixture()
+def amp_test_server():
+    """Isolated ampd + Anvil stack for a single test."""
+    yield from _amp_fixture()
+
+
+@pytest.fixture()
+def reorg_server():
+    """Isolated ampd + Anvil stack for reorg testing."""
+    yield from _amp_fixture(end_block=None)