1 files changed, 58 insertions, 0 deletions

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)
+        }
+}