Testing Projections

⚠️

causet projections doctor and causet projections test referenced below are proposed — there is no CLI equivalent in causet-cli today. Use causet build validate / causet build compile for the real, shipped compiler checks. See CLI Overview.

Projections have two testing layers: static validation by the compiler and runtime validation through integration tests. Both are necessary. The compiler catches structural errors; integration tests catch behavioral errors.

Layer 1: Compiler validation

The causet-compiler is the first line of defense. It validates:

All source_events reference declared event types
All derive expressions reference fields that exist on the referenced event
All fields: entries use supported SQL types
All primary_key columns are declared in fields:
Aggregate columns are declared in fields: with numeric types
Query from references a declared projection
Query where operators are valid
No reserved field names (type, ts, entity_id) used as payload fields

Run the compiler locally before pushing:

causet build validate --runtime concert-app
causet build compile --runtime concert-app --out dist/

--strict treats warnings as errors. Enable it in CI.

If the compiler passes, the projection schema is structurally sound. Behavioral correctness requires integration tests.

Layer 2: Integration tests

Pattern

Deploy IR → Submit intent → Event emitted → Projection updates → Query → Assert

A complete integration test cycle:

# 1. Deploy app IR to a test fork
causet deploy \
  --platform test-platform \
  --app concert-app \
  --fork test-$(git rev-parse --short HEAD)
 
# 2. Submit an intent
curl -X POST \
  "http://runtime/v1/platforms/test-platform/applications/concert-app/forks/test-abc123/intents" \
  -H "Content-Type: application/json" \
  -d '{"type": "UpdateArtist", "entity_id": "artist_1", "info": {"artist_name": "The Cure"}, "ranking": {"popularity_score": 9800}}'
 
# 3. Wait for projection to update (eventual consistency — poll with timeout)
sleep 1
 
# 4. Query the projection
curl -X POST \
  "http://query-service/v1/platforms/test-platform/applications/concert-app/forks/test-abc123/queries/artist_by_id" \
  -d '{"entity_id": "artist_1"}'
 
# 5. Assert expected row

In a test framework

def test_artist_leaderboard_updates_on_artist_updated():
    fork = deploy_test_fork()
 
    submit_intent(fork, {
        "type": "UpdateArtist",
        "entity_id": "artist_1",
        "info": {"artist_name": "The Cure"},
        "ranking": {"popularity_score": 9800}
    })
 
    # Poll with timeout for eventual consistency
    row = wait_for_projection_row(
        fork=fork,
        query="artist_by_id",
        params={"entity_id": "artist_1"},
        timeout_seconds=10
    )
 
    assert row["artist_name"] == "The Cure"
    assert row["popularity_score"] == 9800

Testing projection failures

Verify that invalid events produce a captured failure record — not a silent drop.

def test_projection_failure_captured_on_invalid_payload():
    fork = deploy_test_fork()
 
    # Submit intent with missing required field
    submit_intent(fork, {
        "type": "UpdateArtist",
        "entity_id": "artist_1",
        # Missing: ranking.popularity_score — derive will fail
        "info": {"artist_name": "The Cure"}
    })
 
    # Verify failure record created
    failures = get_projection_failures(
        fork=fork,
        projection="artist_leaderboard",
        status="open"
    )
    assert len(failures) == 1
    assert failures[0]["eventType"] == "ArtistUpdated"
    assert "popularity_score" in failures[0]["errorMessage"]

Testing rebuilds

Verify that a truncated projection table is fully repopulated from event history.

def test_projection_rebuilds_correctly():
    fork = deploy_test_fork()
 
    # Submit 10 events
    for i in range(10):
        submit_intent(fork, {
            "type": "UpdateArtist",
            "entity_id": f"artist_{i}",
            "info": {"artist_name": f"Artist {i}"},
            "ranking": {"popularity_score": i * 100}
        })
 
    wait_for_lag_zero(fork=fork, timeout=30)
 
    # Truncate the table
    db.execute(f"TRUNCATE TABLE {fork.schema}.artist_leaderboard")
 
    # Reset offset and replay
    reset_consumer_offset(fork=fork, to="earliest")
    restart_projection_worker(fork=fork)
 
    wait_for_lag_zero(fork=fork, timeout=30)
 
    # Verify all 10 rows repopulated
    result = query(fork=fork, query="all_artists")
    assert len(result) == 10

Fixture events

Fixture events are pre-defined test event payloads for each projection. They serve as canonical examples of valid inputs and can be used in:

Pre-deploy validation (causet projections doctor)
Unit testing of derive expressions
DLQ replay testing

Store fixture events alongside your .causet files:

projections/
  artist_leaderboard.projections.causet
  artist_leaderboard.fixture.json

{
  "type": "ArtistUpdated",
  "entity_id": "fixture_artist_1",
  "ts": 1719340800000,
  "info": {
    "artist_name": "Fixture Artist",
    "genre": "Electronic"
  },
  "ranking": {
    "popularity_score": 5000
  }
}

Test the fixture against the handler:

causet projections test \
  --projection artist_leaderboard \
  --fixture ./projections/artist_leaderboard.fixture.json

Testing aggregates

Aggregate columns require multiple events to verify correctly.

def test_checkin_count_aggregate():
    fork = deploy_test_fork()
    show_id = "show_test_1"
 
    # Submit 5 check-ins
    for _ in range(5):
        submit_intent(fork, {
            "type": "RecordCheckin",
            "entity_id": show_id,
            "show_id": show_id,
            "venue_id": "venue_1"
        })
 
    wait_for_lag_zero(fork=fork, timeout=10)
 
    result = query(fork=fork, query="show_attendance", params={"show_id": show_id})
    assert result[0]["checkin_count"] == 5
 
    # Submit 2 revocations
    for _ in range(2):
        submit_intent(fork, {
            "type": "RevokeCheckin",
            "entity_id": show_id,
            "show_id": show_id,
            "venue_id": "venue_1"
        })
 
    wait_for_lag_zero(fork=fork, timeout=10)
 
    result = query(fork=fork, query="show_attendance", params={"show_id": show_id})
    assert result[0]["checkin_count"] == 3

Testing multi-event projections

Verify that upsert and delete behave correctly for multi-event projections.

def test_artist_delete_removes_row():
    fork = deploy_test_fork()
 
    submit_intent(fork, {
        "type": "UpdateArtist",
        "entity_id": "artist_delete_test",
        "info": {"artist_name": "Gone Soon"},
        "ranking": {"popularity_score": 100}
    })
 
    wait_for_lag_zero(fork=fork, timeout=10)
 
    # Row exists
    result = query(fork=fork, query="artist_by_id", params={"entity_id": "artist_delete_test"})
    assert len(result) == 1
 
    submit_intent(fork, {
        "type": "DeleteArtist",
        "entity_id": "artist_delete_test"
    })
 
    wait_for_lag_zero(fork=fork, timeout=10)
 
    # Row removed
    result = query(fork=fork, query="artist_by_id", params={"entity_id": "artist_delete_test"})
    assert len(result) == 0

CI integration

Add projection tests to your CI pipeline:

# .github/workflows/test.yml
- name: Compile IR
  run: causet build compile --runtime . --out dist/
 
- name: Run integration tests
  run: pytest tests/projections/ -v
  env:
    CAUSET_RUNTIME_URL: http://localhost:8080
    CAUSET_QUERY_URL: http://localhost:8081
    CAUSET_TEST_PLATFORM: ci-platform
    CAUSET_TEST_APP: concert-app
 
- name: Validate projections
  run: |
    causet projections doctor \
      --platform ci-platform \
      --app concert-app \
      --fork ci-$(git rev-parse --short HEAD) \
      --fail-on-error

Fail the CI run if the compiler, tests, or doctor report any errors. Projection issues caught in CI are orders of magnitude cheaper to fix than projection failures in production.

Schema Versioning Examples