timeline: includeManaged source parity + delta/deep-link contract tests (#1147)

Author: nadaverell

internal/timeline/sqlite_store_test.go

@@ -1148,215 +1148,6 @@ func TestSQLiteStore_Migration_AddsClusterContext(t *testing.T) {
 	}
 }
 
-// A relist re-emits the same informer id; INSERT ... ON CONFLICT must leave the
-// original row untouched (no k8s_event mutation path) instead of duplicating it.
-// Mirrors TestMemoryStore_DedupesIdenticalInformerID.
-func TestSQLiteStore_DedupesIdenticalInformerID(t *testing.T) {
-	store, cleanup := createTestSQLiteStore(t)
-	defer cleanup()
-	ctx := context.Background()
-
-	rv := "100"
-	add := NewInformerEvent("Deployment", "apps/v1", "team-a", "web", "uid-1", rv, EventTypeAdd, HealthHealthy, nil, nil, nil, nil)
-	relistUpdate := NewInformerEvent("Deployment", "apps/v1", "team-a", "web", "uid-1", rv, EventTypeUpdate, HealthHealthy, nil, nil, nil, nil)
-	if add.ID != relistUpdate.ID {
-		t.Fatalf("relist add/update must share an id: %q vs %q", add.ID, relistUpdate.ID)
-	}
-
-	if err := store.Append(ctx, add); err != nil {
-		t.Fatalf("Append add: %v", err)
-	}
-	if err := store.Append(ctx, relistUpdate); err != nil {
-		t.Fatalf("Append relistUpdate: %v", err)
-	}
-
-	got, err := store.Query(ctx, QueryOptions{Limit: 10, IncludeManaged: true})
-	if err != nil {
-		t.Fatalf("Query: %v", err)
-	}
-	if len(got) != 1 {
-		t.Fatalf("expected 1 deduped row, got %d: %+v", len(got), got)
-	}
-	// The original add is kept — the relist did NOT overwrite it into an update.
-	if got[0].EventType != EventTypeAdd {
-		t.Fatalf("informer relist must not mutate the row, got event_type %q want %q", got[0].EventType, EventTypeAdd)
-	}
-
-	del := NewInformerEvent("Deployment", "apps/v1", "team-a", "web", "uid-1", rv, EventTypeDelete, HealthUnknown, nil, nil, nil, nil)
-	if del.ID == add.ID {
-		t.Fatalf("delete must get a distinct id, got %q for both", del.ID)
-	}
-	if err := store.Append(ctx, del); err != nil {
-		t.Fatalf("Append del: %v", err)
-	}
-	got, err = store.Query(ctx, QueryOptions{Limit: 10, IncludeManaged: true})
-	if err != nil {
-		t.Fatalf("Query: %v", err)
-	}
-	if len(got) != 2 {
-		t.Fatalf("expected add + delete = 2 rows, got %d", len(got))
-	}
-}
-
-// A K8s Event's count bump reuses the uid-based id; the store must upsert the
-// mutable fields (count/message/timestamp) in place, not drop the bump or append
-// a duplicate. Mirrors TestMemoryStore_K8sEventCountBumpUpserts.
-func TestSQLiteStore_K8sEventCountBumpUpserts(t *testing.T) {
-	store, cleanup := createTestSQLiteStore(t)
-	defer cleanup()
-	ctx := context.Background()
-
-	base := time.Now()
-	mk := func(count int32, message string, ts time.Time) TimelineEvent {
-		return TimelineEvent{
-			ID: "k8s-uid-1", Timestamp: ts, Source: SourceK8sEvent,
-			Kind: "Pod", Namespace: "team-a", Name: "web-abc",
-			EventType: EventTypeWarning, Reason: "BackOff", Message: message, Count: count,
-		}
-	}
-	if err := store.Append(ctx, mk(1, "back-off 10s", base)); err != nil {
-		t.Fatalf("Append first: %v", err)
-	}
-	if err := store.Append(ctx, mk(5, "back-off 40s", base.Add(30*time.Second))); err != nil {
-		t.Fatalf("Append bump: %v", err)
-	}
-
-	got, err := store.Query(ctx, QueryOptions{Limit: 10, IncludeManaged: true, IncludeK8sEvents: true})
-	if err != nil {
-		t.Fatalf("Query: %v", err)
-	}
-	if len(got) != 1 {
-		t.Fatalf("expected 1 upserted row, got %d: %+v", len(got), got)
-	}
-	if got[0].Count != 5 {
-		t.Fatalf("expected refreshed count 5, got %d", got[0].Count)
-	}
-	if got[0].Message != "back-off 40s" {
-		t.Fatalf("expected refreshed message, got %q", got[0].Message)
-	}
-	if !got[0].Timestamp.Equal(base.Add(30 * time.Second)) {
-		t.Fatalf("expected refreshed timestamp %v, got %v", base.Add(30*time.Second), got[0].Timestamp)
-	}
-}
-
-// An out-of-order older revision of the same K8s Event uid must NOT clobber the
-// newer row already stored. Mirrors the recency guard in
-// TestMemoryStore_K8sEventBumpMovesToRecency.
-func TestSQLiteStore_K8sEventStaleBumpIgnored(t *testing.T) {
-	store, cleanup := createTestSQLiteStore(t)
-	defer cleanup()
-	ctx := context.Background()
-
-	base := time.Now()
-	mk := func(count int32, ts time.Time) TimelineEvent {
-		return TimelineEvent{
-			ID: "k8s-uid-1", Timestamp: ts, Source: SourceK8sEvent,
-			Kind: "Pod", Namespace: "team-a", Name: "web-abc",
-			EventType: EventTypeWarning, Reason: "BackOff", Count: count,
-		}
-	}
-	// Newest revision arrives first, then a stale older relay of the same uid.
-	if err := store.Append(ctx, mk(5, base.Add(30*time.Second))); err != nil {
-		t.Fatalf("Append newer: %v", err)
-	}
-	if err := store.Append(ctx, mk(1, base)); err != nil {
-		t.Fatalf("Append stale: %v", err)
-	}
-
-	got, err := store.Query(ctx, QueryOptions{Limit: 10, IncludeManaged: true, IncludeK8sEvents: true})
-	if err != nil {
-		t.Fatalf("Query: %v", err)
-	}
-	if len(got) != 1 {
-		t.Fatalf("expected 1 row, got %d: %+v", len(got), got)
-	}
-	if got[0].Count != 5 {
-		t.Fatalf("stale relay must not clobber newer row, got count %d want 5", got[0].Count)
-	}
-	if !got[0].Timestamp.Equal(base.Add(30 * time.Second)) {
-		t.Fatalf("stale relay must not roll back timestamp, got %v", got[0].Timestamp)
-	}
-}
-
-// Seq is the delta cursor: SinceSeq must return exactly what arrived after the
-// cursor, keyed on arrival order — a late event with an older timestamp still
-// lands ahead of the cursor. Mirrors TestMemoryStore_SeqCursor.
-func TestSQLiteStore_SeqCursor(t *testing.T) {
-	store, cleanup := createTestSQLiteStore(t)
-	defer cleanup()
-	ctx := context.Background()
-
-	base := time.Now()
-	mk := func(id string, ts time.Time) TimelineEvent {
-		return TimelineEvent{
-			ID: id, Timestamp: ts, Source: SourceInformer,
-			Kind: "Pod", Namespace: "default", Name: id, EventType: EventTypeUpdate,
-		}
-	}
-	if err := store.AppendBatch(ctx, []TimelineEvent{mk("a", base), mk("b", base.Add(time.Second))}); err != nil {
-		t.Fatalf("AppendBatch: %v", err)
-	}
-	all, err := store.Query(ctx, QueryOptions{Limit: 10, IncludeManaged: true})
-	if err != nil {
-		t.Fatalf("Query: %v", err)
-	}
-	if len(all) != 2 || all[0].Seq == 0 || all[1].Seq == 0 || all[0].Seq == all[1].Seq {
-		t.Fatalf("expected 2 events with distinct non-zero seqs, got %+v", all)
-	}
-	cursor := max(all[0].Seq, all[1].Seq)
-
-	if err := store.Append(ctx, mk("late", base.Add(-time.Hour))); err != nil {
-		t.Fatalf("Append late: %v", err)
-	}
-	delta, err := store.Query(ctx, QueryOptions{Limit: 10, IncludeManaged: true, SinceSeq: cursor})
-	if err != nil {
-		t.Fatalf("Query delta: %v", err)
-	}
-	if len(delta) != 1 || delta[0].ID != "late" {
-		t.Fatalf("expected exactly the late arrival past the cursor, got %+v", delta)
-	}
-	if delta[0].Seq <= cursor {
-		t.Fatalf("late arrival seq %d must exceed cursor %d", delta[0].Seq, cursor)
-	}
-}
-
-// A K8s Event count bump upserts in place but must take a fresh seq, or a
-// delta reader holding a cursor past the original arrival never sees the bump.
-// Mirrors TestMemoryStore_K8sEventBumpAdvancesSeq.
-func TestSQLiteStore_K8sEventBumpAdvancesSeq(t *testing.T) {
-	store, cleanup := createTestSQLiteStore(t)
-	defer cleanup()
-	ctx := context.Background()
-
-	base := time.Now()
-	mk := func(count int32, ts time.Time) TimelineEvent {
-		return TimelineEvent{
-			ID: "k8s-uid-1", Timestamp: ts, Source: SourceK8sEvent,
-			Kind: "Pod", Namespace: "default", Name: "web-abc",
-			EventType: EventTypeWarning, Reason: "BackOff", Count: count,
-		}
-	}
-	if err := store.Append(ctx, mk(1, base)); err != nil {
-		t.Fatalf("Append first: %v", err)
-	}
-	first, err := store.Query(ctx, QueryOptions{Limit: 10, IncludeManaged: true, IncludeK8sEvents: true})
-	if err != nil || len(first) != 1 {
-		t.Fatalf("Query first: %v %+v", err, first)
-	}
-	cursor := first[0].Seq
-
-	if err := store.Append(ctx, mk(5, base.Add(30*time.Second))); err != nil {
-		t.Fatalf("Append bump: %v", err)
-	}
-	delta, err := store.Query(ctx, QueryOptions{Limit: 10, IncludeManaged: true, IncludeK8sEvents: true, SinceSeq: cursor})
-	if err != nil {
-		t.Fatalf("Query delta: %v", err)
-	}
-	if len(delta) != 1 || delta[0].Count != 5 {
-		t.Fatalf("expected the bumped row past the cursor, got %+v", delta)
-	}
-}
-
 // A database created before the seq column existed must come back with arrival
 // numbers backfilled from rowid, so pre-upgrade rows are addressable by cursor.
 func TestSQLiteStore_SeqMigrationBackfillsFromRowid(t *testing.T) {
@@ -1452,76 +1243,6 @@ func TestSQLiteStore_ResourceCreatedAtRoundtrip(t *testing.T) {
 	}
 }
 
-// A bump that lost its enrichment (tombstone expired) must not erase the
-// owner/labels/createdAt the row already knows; a bump that carries fresh
-// enrichment wins. Mirrors TestMemoryStore_K8sEventBumpPreservesEnrichment.
-func TestSQLiteStore_K8sEventBumpPreservesEnrichment(t *testing.T) {
-	store, cleanup := createTestSQLiteStore(t)
-	defer cleanup()
-	ctx := context.Background()
-
-	base := time.Now()
-	born := base.Add(-time.Hour).Truncate(time.Millisecond)
-	if err := store.Append(ctx, TimelineEvent{
-		ID: "k8s-uid-1", Timestamp: base, Source: SourceK8sEvent,
-		Kind: "Pod", Namespace: "default", Name: "web-abc",
-		EventType: EventTypeWarning, Reason: "BackOff", Count: 1,
-		CreatedAt: &born,
-		Owner:     &OwnerInfo{Kind: "ReplicaSet", Name: "web"},
-		Labels:    map[string]string{"app": "web"},
-	}); err != nil {
-		t.Fatalf("Append enriched: %v", err)
-	}
-	if err := store.Append(ctx, TimelineEvent{
-		ID: "k8s-uid-1", Timestamp: base.Add(30 * time.Second), Source: SourceK8sEvent,
-		Kind: "Pod", Namespace: "default", Name: "web-abc",
-		EventType: EventTypeWarning, Reason: "BackOff", Count: 5,
-	}); err != nil {
-		t.Fatalf("Append bare bump: %v", err)
-	}
-
-	got, err := store.Query(ctx, QueryOptions{Limit: 10, IncludeManaged: true, IncludeK8sEvents: true})
-	if err != nil || len(got) != 1 {
-		t.Fatalf("Query: %v %+v", err, got)
-	}
-	if got[0].Count != 5 {
-		t.Fatalf("expected bumped count 5, got %d", got[0].Count)
-	}
-	if got[0].CreatedAt == nil || !got[0].CreatedAt.Equal(born) {
-		t.Fatalf("bare bump erased CreatedAt: %+v", got[0].CreatedAt)
-	}
-	if got[0].Owner == nil || got[0].Owner.Name != "web" {
-		t.Fatalf("bare bump erased Owner: %+v", got[0].Owner)
-	}
-	if got[0].Labels["app"] != "web" {
-		t.Fatalf("bare bump erased Labels: %+v", got[0].Labels)
-	}
-
-	// The inverse: a bump that carries enrichment fills a row that lacked it.
-	if err := store.Append(ctx, TimelineEvent{
-		ID: "k8s-uid-2", Timestamp: base, Source: SourceK8sEvent,
-		Kind: "Pod", Namespace: "default", Name: "late-enriched",
-		EventType: EventTypeWarning, Reason: "BackOff", Count: 1,
-	}); err != nil {
-		t.Fatalf("Append bare first: %v", err)
-	}
-	if err := store.Append(ctx, TimelineEvent{
-		ID: "k8s-uid-2", Timestamp: base.Add(time.Minute), Source: SourceK8sEvent,
-		Kind: "Pod", Namespace: "default", Name: "late-enriched",
-		EventType: EventTypeWarning, Reason: "BackOff", Count: 2,
-		Owner: &OwnerInfo{Kind: "Job", Name: "batch"},
-	}); err != nil {
-		t.Fatalf("Append enriched bump: %v", err)
-	}
-	row, err := store.GetEvent(ctx, "k8s-uid-2")
-	if err != nil || row == nil {
-		t.Fatalf("GetEvent: %v %+v", err, row)
-	}
-	if row.Owner == nil || row.Owner.Name != "batch" {
-		t.Fatalf("enriched bump did not fill Owner: %+v", row.Owner)
-	}
-}
-
 // Timestamps are stored as TEXT and ordered lexically; informer events carry
 // the host's local zone while K8s events are UTC. Without UTC normalization on
 // write, a local-zone wall clock sorts ahead of an earlier-instant UTC string.
@@ -1559,58 +1280,6 @@ func TestSQLiteStore_UTCTimestampOrdering(t *testing.T) {
 	}
 }
 
-// Delta reads must page by ascending arrival order so a burst larger than the
-// page limit isn't skipped as the server advances the cursor by each page's max
-// seq. Successive polls must cover every unseen event with no gaps.
-func TestSQLiteStore_DeltaPagesAscendingUnderLimit(t *testing.T) {
-	store, cleanup := createTestSQLiteStore(t)
-	defer cleanup()
-	ctx := context.Background()
-
-	base := time.Now()
-	for i := 0; i < 6; i++ {
-		ev := TimelineEvent{
-			ID: fmt.Sprintf("e%d", i), Timestamp: base.Add(time.Duration(i) * time.Second),
-			Source: SourceInformer, Kind: "Deployment", Namespace: "default",
-			Name: fmt.Sprintf("e%d", i), EventType: EventTypeUpdate,
-		}
-		if err := store.Append(ctx, ev); err != nil {
-			t.Fatalf("Append %d: %v", i, err)
-		}
-	}
-
-	cursor := int64(1)
-	seen := map[int64]bool{}
-	for polls := 0; polls < 10; polls++ {
-		page, err := store.Query(ctx, QueryOptions{Limit: 2, IncludeManaged: true, SinceSeq: cursor})
-		if err != nil {
-			t.Fatalf("delta query: %v", err)
-		}
-		if len(page) == 0 {
-			break
-		}
-		if len(page) == 2 && page[0].Seq >= page[1].Seq {
-			t.Fatalf("delta page not ascending by seq: %d,%d", page[0].Seq, page[1].Seq)
-		}
-		var maxSeq int64
-		for _, e := range page {
-			if seen[e.Seq] {
-				t.Fatalf("seq %d returned twice across polls", e.Seq)
-			}
-			seen[e.Seq] = true
-			if e.Seq > maxSeq {
-				maxSeq = e.Seq
-			}
-		}
-		cursor = maxSeq
-	}
-	for want := int64(2); want <= 6; want++ {
-		if !seen[want] {
-			t.Fatalf("delta paging skipped seq %d; seen=%v", want, seen)
-		}
-	}
-}
-
 // seq must not rewind when retention empties the table: a live client's cursor
 // (seq>N) would otherwise go dead with no forced resync while the store lives.
 func TestSQLiteStore_SeqSurvivesEmptyTable(t *testing.T) {

internal/timeline/store_conformance_test.go

@@ -0,0 +1,30 @@
+package timeline
+
+import (
+	"os"
+	"path/filepath"
+	"testing"
+
+	"github.com/skyhook-io/radar/pkg/timeline/storetest"
+)
+
+// The shared EventStore contract lives in pkg/timeline/storetest and runs here
+// against SQLiteStore; MemoryStore runs the same suite from the pkg module.
+// Only SQLite-specific behavior (migrations, quarantine, retention, timestamp
+// encoding) is tested in this package's own files.
+func TestStoreConformance_SQLite(t *testing.T) {
+	storetest.RunConformance(t, func(t *testing.T) EventStore {
+		t.Helper()
+		dir, err := os.MkdirTemp("", "timeline-conformance-*")
+		if err != nil {
+			t.Fatalf("MkdirTemp: %v", err)
+		}
+		t.Cleanup(func() { os.RemoveAll(dir) })
+		store, err := NewSQLiteStore(filepath.Join(dir, "test.db"))
+		if err != nil {
+			t.Fatalf("NewSQLiteStore: %v", err)
+		}
+		t.Cleanup(func() { store.Close() })
+		return store
+	})
+}

pkg/timeline/conformance_test.go

@@ -0,0 +1,14 @@
+package timeline_test
+
+import (
+	"testing"
+
+	timeline "github.com/skyhook-io/radar/pkg/timeline"
+	"github.com/skyhook-io/radar/pkg/timeline/storetest"
+)
+
+func TestMemoryStore_Conformance(t *testing.T) {
+	storetest.RunConformance(t, func(t *testing.T) timeline.EventStore {
+		return timeline.NewMemoryStore(100)
+	})
+}