From 50139c994e539f2f902fc0c06ab29ee93c1bc03c Mon Sep 17 00:00:00 2001
From: sangtran <tranquangsang12.7@gmail.com>
Date: Sat, 12 Sep 2026 16:35:06 +0700
Subject: refactor(event-time): make source timestamps authoritative

---
 core/operator/watermark_test.go | 58 +++++++++++++++++++++++++++++++++++++++++
 1 file changed, 58 insertions(+)
 create mode 100644 core/operator/watermark_test.go

(limited to 'core/operator/watermark_test.go')

diff --git a/core/operator/watermark_test.go b/core/operator/watermark_test.go
new file mode 100644
index 0000000..0a0495e
--- /dev/null
+++ b/core/operator/watermark_test.go
@@ -0,0 +1,58 @@
+package operator_test
+
+import (
+	"context"
+	"flux/core"
+	"flux/core/operator"
+	"testing"
+)
+
+type timestampEvent struct {
+	EventTime int64
+}
+
+func TestWatermarkGeneratorPreservesAssignedEventTimestamp(t *testing.T) {
+	ctx := context.Background()
+
+	op := operator.NewWatermarkGeneratorOperator[timestampEvent](0)
+
+	if err := op.Open(ctx); err != nil {
+		t.Fatalf("open operator failed: %v", err)
+	}
+
+	t.Cleanup(func() {
+		if err := op.Close(ctx); err != nil {
+			t.Errorf("close operator: %v", err)
+		}
+	})
+
+	out := &captureEmitter[timestampEvent]{}
+	input := core.NewDataRecord("user-1", timestampEvent{EventTime: 999}, 40)
+
+	if err := op.Process(ctx, input, out); err != nil {
+		t.Fatalf("process operator failed: %v", err)
+	}
+
+	records := out.Records()
+	if len(records) == 0 {
+		t.Fatal("operator emitted no records")
+	}
+
+	got := records[0]
+
+	if got.Type != core.RecordData {
+		t.Fatalf("first output type: got %v, want RecordData", got.Type)
+	}
+
+	if got.Key != "user-1" {
+		t.Errorf("data key: got %q, want %q", got.Key, "user-1")
+	}
+
+	if got.Value != input.Value {
+		t.Errorf("data value: got %+v, want %+v", got.Value, input.Value)
+	}
+
+	if got.Timestamp != 40 {
+		t.Errorf("data timestamp: got %d, want %d", got.Timestamp, 40)
+	}
+}
-- 
cgit v1.3.1