tutti/packages/agent/activity-replication/conformance/conformance_test.go

package conformance

import (
	"context"
	"encoding/json"
	"errors"
	"strings"
	"testing"

	activityreplication "github.com/tutti-os/tutti/packages/agent/activity-replication"
)

func TestPublishedFixturesAreSelfConsistent(t *testing.T) {
	t.Parallel()

	for _, fixture := range ProjectionFixtures() {
		if err := activityreplication.ValidateBatch(fixture.WantBatch); err != nil {
			t.Errorf("projection fixture %q: invalid expected batch: %v", fixture.Name, err)
		}
		if len(fixture.Canonical) != len(fixture.WantBatch.Mutations) {
			t.Errorf("projection fixture %q: %d canonical snapshots, %d expected mutations", fixture.Name, len(fixture.Canonical), len(fixture.WantBatch.Mutations))
		}
	}

	for _, fixture := range SinkFixtures() {
		for _, step := range fixture.Steps {
			err := activityreplication.ValidateBatch(step.Batch)
			if step.WantRejection != nil {
				if step.WantRejection.Kind == activityreplication.RejectionSchema && err == nil {
					t.Errorf("sink fixture %q step %q: schema rejection batch validates", fixture.Name, step.Name)
				}
				continue
			}
			if err != nil {
				t.Errorf("sink fixture %q step %q: invalid batch: %v", fixture.Name, step.Name, err)
				continue
			}
			if len(step.WantAcknowledgements) != len(step.Batch.Mutations) {
				t.Errorf("sink fixture %q step %q: %d acknowledgements, want %d", fixture.Name, step.Name, len(step.WantAcknowledgements), len(step.Batch.Mutations))
				continue
			}
			for index, acknowledgement := range step.WantAcknowledgements {
				mutation := step.Batch.Mutations[index]
				if acknowledgement.MutationID != mutation.MutationID || acknowledgement.TransactionID != mutation.TransactionID {
					t.Errorf("sink fixture %q step %q: acknowledgement %d identity does not match mutation", fixture.Name, step.Name, index)
				}
			}
			result, err := activityreplication.SummarizeAcknowledgements(step.WantAcknowledgements)
			if err != nil || result != step.WantResult {
				t.Errorf("sink fixture %q step %q: summarized result %#v, want %#v, error %v", fixture.Name, step.Name, result, step.WantResult, err)
			}
		}
	}
}

func TestRunSinkRejectsAcknowledgementIdentityMismatch(t *testing.T) {
	t.Parallel()

	mutation := sessionMutation("mutation-1", "transaction-1", "title", 100)
	want := activityreplication.AcknowledgeApplied(mutation, 1)
	got := want
	got.TransactionID = "wrong-transaction"
	sink := &scriptedSink{report: ApplyReport{
		Result:           activityreplication.ApplyResult{AcceptedCount: 1, Cursor: 1},
		Acknowledgements: []activityreplication.MutationAcknowledgement{got},
	}}
	err := RunSink(context.Background(), sink, SinkFixture{
		Name: "identity mismatch",
		Steps: []SinkStep{{
			Name: "apply", Batch: batch(mutation),
			WantResult: activityreplication.ApplyResult{AcceptedCount: 1, Cursor: 1}, WantAcknowledgements: []activityreplication.MutationAcknowledgement{want},
		}},
	})
	if err == nil || !strings.Contains(err.Error(), "acknowledgements") {
		t.Fatalf("RunSink() error = %v, want acknowledgement mismatch", err)
	}
}

func TestRunSinkRejectsResultThatDoesNotSummarizeAcknowledgements(t *testing.T) {
	t.Parallel()

	mutation := sessionMutation("mutation-1", "transaction-1", "title", 100)
	acknowledgement := activityreplication.AcknowledgeApplied(mutation, 2)
	sink := &scriptedSink{report: ApplyReport{
		Result:           activityreplication.ApplyResult{AcceptedCount: 1, Cursor: 1},
		Acknowledgements: []activityreplication.MutationAcknowledgement{acknowledgement},
	}}
	err := RunSink(context.Background(), sink, SinkFixture{
		Name: "aggregate mismatch",
		Steps: []SinkStep{{
			Name: "apply", Batch: batch(mutation),
			WantResult: activityreplication.ApplyResult{AcceptedCount: 1, Cursor: 1}, WantAcknowledgements: []activityreplication.MutationAcknowledgement{acknowledgement},
		}},
	})
	if err == nil || !strings.Contains(err.Error(), "does not summarize") {
		t.Fatalf("RunSink() error = %v, want aggregate mismatch", err)
	}
}

func TestRunSinkRejectsMissingPersistedSessionScope(t *testing.T) {
	t.Parallel()

	mutation := sessionMutation("mutation-1", "transaction-1", "title", 100)
	acknowledgement := activityreplication.AcknowledgeApplied(mutation, 1)
	entity, err := json.Marshal(mutation.Session)
	if err != nil {
		t.Fatal(err)
	}
	sink := &scriptedSink{
		report: ApplyReport{
			Result:           activityreplication.ApplyResult{AcceptedCount: 1, Cursor: 1},
			Acknowledgements: []activityreplication.MutationAcknowledgement{acknowledgement},
		},
		snapshot: SinkSnapshot{Entity: entity},
		found:    true,
	}
	err = RunSink(context.Background(), sink, SinkFixture{
		Name: "missing persisted session scope",
		Steps: []SinkStep{{
			Name: "apply", Batch: batch(mutation),
			WantResult: activityreplication.ApplyResult{AcceptedCount: 1, Cursor: 1}, WantAcknowledgements: []activityreplication.MutationAcknowledgement{acknowledgement},
		}},
		WantSnapshots: []SnapshotExpectation{presentSnapshot(mutation)},
	})
	if err == nil || !strings.Contains(err.Error(), "session scope") {
		t.Fatalf("RunSink() error = %v, want missing session scope", err)
	}
}

type scriptedSink struct {
	report    ApplyReport
	err       error
	snapshot  SinkSnapshot
	found     bool
	lookupErr error
}

func (*scriptedSink) Reset(context.Context) error { return nil }

func (s *scriptedSink) Apply(context.Context, activityreplication.ChangeBatch) (ApplyReport, error) {
	return s.report, s.err
}

func (s *scriptedSink) Lookup(context.Context, activityreplication.EntityType, activityreplication.EntityKey) (SinkSnapshot, bool, error) {
	if s.lookupErr != nil || s.found {
		return s.snapshot, s.found, s.lookupErr
	}
	return SinkSnapshot{}, false, errors.New("unexpected lookup")
}