test(s3/lifecycle): meta-log Event builder + monotonic clock fixture (#9406)
* test(s3/lifecycle): meta-log Event builder + monotonic clock fixture Several test files build *reader.Event ad-hoc; consolidate the common shape into the lifecycletest package as task #12 spec calls out ("fixture meta-log generator"). New tests using the builder don't have to thread Mtime / ShardID / leaf-name semantics by hand, and existing helpers can migrate over time without churning this PR. NewCreate / NewDelete / NewUpdate cover the three event shapes; WithSize / WithModTime / WithTtlSec / WithVersionID / WithExtended / WithChunks / WithBootstrapVersion / WithShardID compose deterministic overrides. ShardID defaults to s3lifecycle.ShardID(bucket, key) so events route through the same shard the production reader would. MetaLogClock issues monotonic timestamps with a configurable step (default 1s); concurrent-safe so fan-out fixtures don't have to lock externally. 15 unit tests pin every option, the IsCreate/IsDelete/IsUpdate discriminators, leaf-name extraction for nested keys, ShardID derivation, option-ordering semantics, the concurrent clock contract under -race, and a Peek-doesn't-advance check. * test(s3/lifecycle): address review comments on event builder - leafOf strips trailing slashes before splitting so directory-key fixtures (e.g. "folder/") get the slashless leaf "folder" — pre-fix it returned "" which would break router tests for directory markers. - NewUpdate now seeds OldEntry.Attributes.Mtime with the event ts (matching NewDelete), so a downstream router that compares mtimes doesn't see a synthetic 1970 epoch on the pre-update state. - New WithOldSize / WithOldChunks / WithOldModTime options let Update events configure pre-update state independently. The unprefixed variants still target NewEntry on Update events; the With Old* options are no-ops on Create (no OldEntry to mutate) and never bleed into NewEntry. 5 new tests pin: directory-key + multi-slash leaf extraction; OldEntry mtime default on Update; the WithOld* targeting + Create-event no-bleed contract.
This commit is contained in:
Chris Lu
GitHub
parent
9d20e71883
commit
edbe7ab140
286
weed/s3api/s3lifecycle/lifecycletest/eventbuilder.go
Normal file
286
weed/s3api/s3lifecycle/lifecycletest/eventbuilder.go
Normal file
@ -0,0 +1,286 @@
|
||||
package lifecycletest
|
||||
|
||||
import (
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
|
||||
"github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants"
|
||||
"github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle"
|
||||
"github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle/reader"
|
||||
)
|
||||
|
||||
// EventOption mutates a reader.Event during construction. Options compose
|
||||
// in order — later options override earlier ones if they touch the same
|
||||
// field, which lets a default+override pattern work without surprises.
|
||||
type EventOption func(*reader.Event)
|
||||
|
||||
// WithSize sets the Size on whichever entry the event populates (NewEntry
|
||||
// for Create / Update, OldEntry for Delete). Constructors that build
|
||||
// both entries apply the size to NewEntry.
|
||||
func WithSize(bytes int64) EventOption {
|
||||
return func(e *reader.Event) {
|
||||
if e.NewEntry != nil && e.NewEntry.Attributes != nil {
|
||||
e.NewEntry.Attributes.FileSize = uint64(bytes)
|
||||
} else if e.OldEntry != nil && e.OldEntry.Attributes != nil {
|
||||
e.OldEntry.Attributes.FileSize = uint64(bytes)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// WithModTime sets Mtime/MtimeNs on the populated entry.
|
||||
func WithModTime(t time.Time) EventOption {
|
||||
return func(e *reader.Event) {
|
||||
secs, nanos := t.Unix(), int32(t.Nanosecond())
|
||||
if e.NewEntry != nil && e.NewEntry.Attributes != nil {
|
||||
e.NewEntry.Attributes.Mtime = secs
|
||||
e.NewEntry.Attributes.MtimeNs = nanos
|
||||
} else if e.OldEntry != nil && e.OldEntry.Attributes != nil {
|
||||
e.OldEntry.Attributes.Mtime = secs
|
||||
e.OldEntry.Attributes.MtimeNs = nanos
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// WithTtlSec sets the TTL on the populated entry. Used by tests that
|
||||
// exercise the lifecycle metadata-only delete path (TtlSec > 0 means
|
||||
// the volume reclaims chunks naturally).
|
||||
func WithTtlSec(ttl int32) EventOption {
|
||||
return func(e *reader.Event) {
|
||||
if e.NewEntry != nil && e.NewEntry.Attributes != nil {
|
||||
e.NewEntry.Attributes.TtlSec = ttl
|
||||
} else if e.OldEntry != nil && e.OldEntry.Attributes != nil {
|
||||
e.OldEntry.Attributes.TtlSec = ttl
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// WithVersionID stamps Seaweed-X-Amz-Version-Id on the populated entry's
|
||||
// Extended map. Used by versioning-aware router and dispatcher tests.
|
||||
func WithVersionID(versionID string) EventOption {
|
||||
return func(e *reader.Event) {
|
||||
var entry *filer_pb.Entry
|
||||
if e.NewEntry != nil {
|
||||
entry = e.NewEntry
|
||||
} else if e.OldEntry != nil {
|
||||
entry = e.OldEntry
|
||||
}
|
||||
if entry == nil {
|
||||
return
|
||||
}
|
||||
if entry.Extended == nil {
|
||||
entry.Extended = map[string][]byte{}
|
||||
}
|
||||
entry.Extended[s3_constants.ExtVersionIdKey] = []byte(versionID)
|
||||
}
|
||||
}
|
||||
|
||||
// WithExtended sets an arbitrary Extended key/value on the populated entry.
|
||||
func WithExtended(key string, value []byte) EventOption {
|
||||
return func(e *reader.Event) {
|
||||
var entry *filer_pb.Entry
|
||||
if e.NewEntry != nil {
|
||||
entry = e.NewEntry
|
||||
} else if e.OldEntry != nil {
|
||||
entry = e.OldEntry
|
||||
}
|
||||
if entry == nil {
|
||||
return
|
||||
}
|
||||
if entry.Extended == nil {
|
||||
entry.Extended = map[string][]byte{}
|
||||
}
|
||||
entry.Extended[key] = value
|
||||
}
|
||||
}
|
||||
|
||||
// WithChunks attaches FileChunks to the populated entry. Identity-CAS in
|
||||
// LifecycleDelete uses the head chunk's FID, so tests that exercise CAS
|
||||
// drift need at least one chunk. For Update events the chunks land on
|
||||
// NewEntry; use WithOldChunks to target the pre-update state.
|
||||
func WithChunks(chunks ...*filer_pb.FileChunk) EventOption {
|
||||
return func(e *reader.Event) {
|
||||
if e.NewEntry != nil {
|
||||
e.NewEntry.Chunks = append(e.NewEntry.Chunks, chunks...)
|
||||
} else if e.OldEntry != nil {
|
||||
e.OldEntry.Chunks = append(e.OldEntry.Chunks, chunks...)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// WithOldSize sets FileSize on OldEntry specifically. Use on Update
|
||||
// events to configure the pre-update size (the WithSize default lands
|
||||
// on NewEntry when both are populated).
|
||||
func WithOldSize(bytes int64) EventOption {
|
||||
return func(e *reader.Event) {
|
||||
if e.OldEntry != nil && e.OldEntry.Attributes != nil {
|
||||
e.OldEntry.Attributes.FileSize = uint64(bytes)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// WithOldChunks attaches FileChunks to OldEntry specifically. Use on
|
||||
// Update events to configure the pre-update chunk list (WithChunks
|
||||
// targets NewEntry when both are populated).
|
||||
func WithOldChunks(chunks ...*filer_pb.FileChunk) EventOption {
|
||||
return func(e *reader.Event) {
|
||||
if e.OldEntry != nil {
|
||||
e.OldEntry.Chunks = append(e.OldEntry.Chunks, chunks...)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// WithOldModTime sets Mtime/MtimeNs on OldEntry specifically. Update
|
||||
// events whose pre-update mtime should differ from the event timestamp
|
||||
// use this to override the default.
|
||||
func WithOldModTime(t time.Time) EventOption {
|
||||
return func(e *reader.Event) {
|
||||
if e.OldEntry != nil && e.OldEntry.Attributes != nil {
|
||||
e.OldEntry.Attributes.Mtime = t.Unix()
|
||||
e.OldEntry.Attributes.MtimeNs = int32(t.Nanosecond())
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// WithBootstrapVersion attaches a BootstrapVersion to the event.
|
||||
// Bootstrap-walk events use this to carry per-version state the live
|
||||
// meta-log doesn't see.
|
||||
func WithBootstrapVersion(bv *reader.BootstrapVersion) EventOption {
|
||||
return func(e *reader.Event) { e.BootstrapVersion = bv }
|
||||
}
|
||||
|
||||
// WithShardID overrides the computed ShardID. Most tests should leave
|
||||
// it as the s3lifecycle.ShardID default to mirror production routing.
|
||||
func WithShardID(shard int) EventOption {
|
||||
return func(e *reader.Event) { e.ShardID = shard }
|
||||
}
|
||||
|
||||
// NewCreate builds a Create event (NewEntry populated, OldEntry nil) at
|
||||
// the given timestamp. Bucket and key are required; everything else is
|
||||
// derived (Mtime defaults to ts, FileSize defaults to 0). Apply options
|
||||
// to override.
|
||||
func NewCreate(bucket, key string, ts time.Time, opts ...EventOption) *reader.Event {
|
||||
e := &reader.Event{
|
||||
TsNs: ts.UnixNano(),
|
||||
Bucket: bucket,
|
||||
Key: key,
|
||||
ShardID: s3lifecycle.ShardID(bucket, key),
|
||||
NewEntry: &filer_pb.Entry{
|
||||
Name: leafOf(key),
|
||||
Attributes: &filer_pb.FuseAttributes{
|
||||
Mtime: ts.Unix(),
|
||||
MtimeNs: int32(ts.Nanosecond()),
|
||||
},
|
||||
},
|
||||
}
|
||||
for _, o := range opts {
|
||||
o(e)
|
||||
}
|
||||
return e
|
||||
}
|
||||
|
||||
// NewDelete builds a Delete event (OldEntry populated, NewEntry nil).
|
||||
// Tests usually pass the entry the prior Create produced as a snapshot
|
||||
// of pre-delete state; this helper builds a minimal stand-in.
|
||||
func NewDelete(bucket, key string, ts time.Time, opts ...EventOption) *reader.Event {
|
||||
e := &reader.Event{
|
||||
TsNs: ts.UnixNano(),
|
||||
Bucket: bucket,
|
||||
Key: key,
|
||||
ShardID: s3lifecycle.ShardID(bucket, key),
|
||||
OldEntry: &filer_pb.Entry{
|
||||
Name: leafOf(key),
|
||||
Attributes: &filer_pb.FuseAttributes{
|
||||
Mtime: ts.Unix(),
|
||||
MtimeNs: int32(ts.Nanosecond()),
|
||||
},
|
||||
},
|
||||
}
|
||||
for _, o := range opts {
|
||||
o(e)
|
||||
}
|
||||
return e
|
||||
}
|
||||
|
||||
// NewUpdate builds an Update event (both OldEntry and NewEntry populated).
|
||||
// Same defaults as NewCreate; options apply to NewEntry per the rules in
|
||||
// each option's doc.
|
||||
func NewUpdate(bucket, key string, ts time.Time, opts ...EventOption) *reader.Event {
|
||||
e := &reader.Event{
|
||||
TsNs: ts.UnixNano(),
|
||||
Bucket: bucket,
|
||||
Key: key,
|
||||
ShardID: s3lifecycle.ShardID(bucket, key),
|
||||
OldEntry: &filer_pb.Entry{
|
||||
Name: leafOf(key),
|
||||
Attributes: &filer_pb.FuseAttributes{
|
||||
Mtime: ts.Unix(),
|
||||
MtimeNs: int32(ts.Nanosecond()),
|
||||
},
|
||||
},
|
||||
NewEntry: &filer_pb.Entry{
|
||||
Name: leafOf(key),
|
||||
Attributes: &filer_pb.FuseAttributes{
|
||||
Mtime: ts.Unix(),
|
||||
MtimeNs: int32(ts.Nanosecond()),
|
||||
},
|
||||
},
|
||||
}
|
||||
for _, o := range opts {
|
||||
o(e)
|
||||
}
|
||||
return e
|
||||
}
|
||||
|
||||
// MetaLogClock produces monotonically increasing timestamps for fixture
|
||||
// generation. Each call to Next advances by Step (default 1s) so tests
|
||||
// don't have to thread a counter through every helper invocation. Safe
|
||||
// for concurrent use.
|
||||
type MetaLogClock struct {
|
||||
mu sync.Mutex
|
||||
now time.Time
|
||||
step time.Duration
|
||||
}
|
||||
|
||||
// NewMetaLogClock returns a clock that ticks forward Step on every Next
|
||||
// call. Step defaults to 1s when zero.
|
||||
func NewMetaLogClock(start time.Time, step time.Duration) *MetaLogClock {
|
||||
if step <= 0 {
|
||||
step = time.Second
|
||||
}
|
||||
return &MetaLogClock{now: start, step: step}
|
||||
}
|
||||
|
||||
// Next returns the current timestamp and advances by Step.
|
||||
func (c *MetaLogClock) Next() time.Time {
|
||||
c.mu.Lock()
|
||||
defer c.mu.Unlock()
|
||||
t := c.now
|
||||
c.now = c.now.Add(c.step)
|
||||
return t
|
||||
}
|
||||
|
||||
// Peek returns what Next() would return without advancing.
|
||||
func (c *MetaLogClock) Peek() time.Time {
|
||||
c.mu.Lock()
|
||||
defer c.mu.Unlock()
|
||||
return c.now
|
||||
}
|
||||
|
||||
// leafOf returns the basename of a slash-separated key. Filer entries
|
||||
// store only the leaf name; tests that mirror production layout need
|
||||
// the same shape. Trailing slashes are stripped first so directory-key
|
||||
// fixtures (e.g. "folder/") get the slashless leaf "folder" — the
|
||||
// production directory-marker write path stores the name without the
|
||||
// trailing slash.
|
||||
func leafOf(key string) string {
|
||||
for len(key) > 0 && key[len(key)-1] == '/' {
|
||||
key = key[:len(key)-1]
|
||||
}
|
||||
for i := len(key) - 1; i >= 0; i-- {
|
||||
if key[i] == '/' {
|
||||
return key[i+1:]
|
||||
}
|
||||
}
|
||||
return key
|
||||
}
|
||||
240
weed/s3api/s3lifecycle/lifecycletest/eventbuilder_test.go
Normal file
240
weed/s3api/s3lifecycle/lifecycletest/eventbuilder_test.go
Normal file
@ -0,0 +1,240 @@
|
||||
package lifecycletest
|
||||
|
||||
import (
|
||||
"sync"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
|
||||
"github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants"
|
||||
"github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle"
|
||||
"github.com/stretchr/testify/assert"
|
||||
"github.com/stretchr/testify/require"
|
||||
)
|
||||
|
||||
func TestNewCreate_PopulatesNewEntryOnly(t *testing.T) {
|
||||
t0 := time.Unix(1700000000, 123)
|
||||
e := NewCreate("bk", "obj.txt", t0)
|
||||
require.NotNil(t, e.NewEntry)
|
||||
assert.Nil(t, e.OldEntry)
|
||||
assert.True(t, e.IsCreate())
|
||||
assert.Equal(t, t0.UnixNano(), e.TsNs)
|
||||
assert.Equal(t, "bk", e.Bucket)
|
||||
assert.Equal(t, "obj.txt", e.Key)
|
||||
assert.Equal(t, "obj.txt", e.NewEntry.Name)
|
||||
// Mtime defaults to the event timestamp; tests that don't care can
|
||||
// rely on this rather than threading time through every call.
|
||||
assert.Equal(t, t0.Unix(), e.NewEntry.Attributes.Mtime)
|
||||
assert.Equal(t, int32(t0.Nanosecond()), e.NewEntry.Attributes.MtimeNs)
|
||||
}
|
||||
|
||||
func TestNewDelete_PopulatesOldEntryOnly(t *testing.T) {
|
||||
t0 := time.Unix(1700000000, 0)
|
||||
e := NewDelete("bk", "obj.txt", t0)
|
||||
require.NotNil(t, e.OldEntry)
|
||||
assert.Nil(t, e.NewEntry)
|
||||
assert.True(t, e.IsDelete())
|
||||
assert.Equal(t, "obj.txt", e.OldEntry.Name)
|
||||
}
|
||||
|
||||
func TestNewUpdate_PopulatesBothEntries(t *testing.T) {
|
||||
t0 := time.Unix(1700000000, 0)
|
||||
e := NewUpdate("bk", "obj.txt", t0)
|
||||
require.NotNil(t, e.OldEntry)
|
||||
require.NotNil(t, e.NewEntry)
|
||||
assert.False(t, e.IsCreate())
|
||||
assert.False(t, e.IsDelete())
|
||||
// OldEntry's Mtime defaults to the event ts so its mirror of pre-
|
||||
// update state reflects the same wall-clock origin as NewEntry —
|
||||
// a downstream router that compares mtimes won't see a synthetic
|
||||
// 1970 epoch.
|
||||
assert.Equal(t, t0.Unix(), e.OldEntry.Attributes.Mtime)
|
||||
assert.Equal(t, t0.Unix(), e.NewEntry.Attributes.Mtime)
|
||||
}
|
||||
|
||||
func TestNewCreate_NestedKeyUsesLeafName(t *testing.T) {
|
||||
// Filer entries store only the leaf name; mirroring that in the
|
||||
// fixture keeps router/dispatcher tests realistic.
|
||||
e := NewCreate("bk", "a/b/c.txt", time.Unix(0, 0))
|
||||
assert.Equal(t, "c.txt", e.NewEntry.Name)
|
||||
}
|
||||
|
||||
func TestNewCreate_DirectoryKeyUsesSlashlessLeaf(t *testing.T) {
|
||||
// Directory-marker objects (S3 keys ending in "/") store the leaf
|
||||
// name without the trailing slash on the filer side. A pre-fix
|
||||
// regression returned "" here.
|
||||
e := NewCreate("bk", "folder/", time.Unix(0, 0))
|
||||
assert.Equal(t, "folder", e.NewEntry.Name)
|
||||
|
||||
// Nested directory key strips both the trailing slash AND the
|
||||
// parent prefix.
|
||||
e2 := NewCreate("bk", "a/b/folder/", time.Unix(0, 0))
|
||||
assert.Equal(t, "folder", e2.NewEntry.Name)
|
||||
|
||||
// Multiple trailing slashes collapse.
|
||||
e3 := NewCreate("bk", "folder///", time.Unix(0, 0))
|
||||
assert.Equal(t, "folder", e3.NewEntry.Name)
|
||||
}
|
||||
|
||||
func TestNewCreate_ShardIDDerivedFromKey(t *testing.T) {
|
||||
// ShardID matches s3lifecycle.ShardID(bucket, key) so the event
|
||||
// routes through the same shard the production reader would
|
||||
// classify it under.
|
||||
e := NewCreate("bk", "obj.txt", time.Unix(0, 0))
|
||||
assert.Equal(t, s3lifecycle.ShardID("bk", "obj.txt"), e.ShardID)
|
||||
}
|
||||
|
||||
func TestEventOption_WithSize(t *testing.T) {
|
||||
e := NewCreate("bk", "k", time.Unix(0, 0), WithSize(4096))
|
||||
assert.Equal(t, uint64(4096), e.NewEntry.Attributes.FileSize)
|
||||
}
|
||||
|
||||
func TestEventOption_WithSizeAppliesToOldEntryOnDelete(t *testing.T) {
|
||||
// Delete events have OldEntry only; WithSize must apply there.
|
||||
e := NewDelete("bk", "k", time.Unix(0, 0), WithSize(8192))
|
||||
assert.Equal(t, uint64(8192), e.OldEntry.Attributes.FileSize)
|
||||
}
|
||||
|
||||
func TestEventOption_WithModTimeOverridesDefault(t *testing.T) {
|
||||
// Default Mtime is event ts; explicit WithModTime overrides.
|
||||
t0 := time.Unix(1700000000, 0)
|
||||
override := time.Unix(1700000123, 456)
|
||||
e := NewCreate("bk", "k", t0, WithModTime(override))
|
||||
assert.Equal(t, override.Unix(), e.NewEntry.Attributes.Mtime)
|
||||
assert.Equal(t, int32(override.Nanosecond()), e.NewEntry.Attributes.MtimeNs)
|
||||
}
|
||||
|
||||
func TestEventOption_WithTtlSec(t *testing.T) {
|
||||
// WithTtlSec drives the lifecycle metadata-only delete path; the
|
||||
// gate fires when the live entry's TtlSec > 0.
|
||||
e := NewCreate("bk", "k", time.Unix(0, 0), WithTtlSec(300))
|
||||
assert.Equal(t, int32(300), e.NewEntry.Attributes.TtlSec)
|
||||
}
|
||||
|
||||
func TestEventOption_WithVersionID(t *testing.T) {
|
||||
e := NewCreate("bk", "k", time.Unix(0, 0), WithVersionID("v_abc"))
|
||||
assert.Equal(t, []byte("v_abc"), e.NewEntry.Extended[s3_constants.ExtVersionIdKey])
|
||||
}
|
||||
|
||||
func TestEventOption_WithExtendedKeyValue(t *testing.T) {
|
||||
e := NewCreate("bk", "k", time.Unix(0, 0), WithExtended("Custom-Tag", []byte("v1")))
|
||||
assert.Equal(t, []byte("v1"), e.NewEntry.Extended["Custom-Tag"])
|
||||
}
|
||||
|
||||
func TestEventOption_WithChunks(t *testing.T) {
|
||||
c1 := &filer_pb.FileChunk{FileId: "1,abc"}
|
||||
c2 := &filer_pb.FileChunk{FileId: "1,def"}
|
||||
e := NewCreate("bk", "k", time.Unix(0, 0), WithChunks(c1, c2))
|
||||
require.Len(t, e.NewEntry.Chunks, 2)
|
||||
assert.Equal(t, "1,abc", e.NewEntry.Chunks[0].FileId)
|
||||
assert.Equal(t, "1,def", e.NewEntry.Chunks[1].FileId)
|
||||
}
|
||||
|
||||
func TestEventOption_WithShardIDOverrides(t *testing.T) {
|
||||
e := NewCreate("bk", "obj.txt", time.Unix(0, 0), WithShardID(7))
|
||||
assert.Equal(t, 7, e.ShardID)
|
||||
}
|
||||
|
||||
func TestEventOption_WithOldSizeTargetsOldEntryOnUpdate(t *testing.T) {
|
||||
// On Update events, WithSize lands on NewEntry; WithOldSize lets
|
||||
// the caller configure pre-update state independently. Both options
|
||||
// in a single call should produce distinct sizes on the two entries.
|
||||
t0 := time.Unix(1700000000, 0)
|
||||
e := NewUpdate("bk", "k", t0, WithSize(200), WithOldSize(100))
|
||||
assert.Equal(t, uint64(200), e.NewEntry.Attributes.FileSize, "WithSize lands on NewEntry")
|
||||
assert.Equal(t, uint64(100), e.OldEntry.Attributes.FileSize, "WithOldSize lands on OldEntry")
|
||||
}
|
||||
|
||||
func TestEventOption_WithOldChunksTargetsOldEntryOnUpdate(t *testing.T) {
|
||||
newChunk := &filer_pb.FileChunk{FileId: "1,new"}
|
||||
oldChunk := &filer_pb.FileChunk{FileId: "1,old"}
|
||||
t0 := time.Unix(1700000000, 0)
|
||||
e := NewUpdate("bk", "k", t0, WithChunks(newChunk), WithOldChunks(oldChunk))
|
||||
require.Len(t, e.NewEntry.Chunks, 1)
|
||||
require.Len(t, e.OldEntry.Chunks, 1)
|
||||
assert.Equal(t, "1,new", e.NewEntry.Chunks[0].FileId)
|
||||
assert.Equal(t, "1,old", e.OldEntry.Chunks[0].FileId)
|
||||
}
|
||||
|
||||
func TestEventOption_WithOldModTimeTargetsOldEntryOnUpdate(t *testing.T) {
|
||||
// Pre-update mtime can lag the event timestamp; WithOldModTime lets
|
||||
// a router test pin the noncurrent-clock origin.
|
||||
t0 := time.Unix(1700000000, 0)
|
||||
older := time.Unix(1699000000, 500)
|
||||
e := NewUpdate("bk", "k", t0, WithOldModTime(older))
|
||||
assert.Equal(t, older.Unix(), e.OldEntry.Attributes.Mtime)
|
||||
assert.Equal(t, int32(500), e.OldEntry.Attributes.MtimeNs)
|
||||
// NewEntry's mtime stays at the default ts.
|
||||
assert.Equal(t, t0.Unix(), e.NewEntry.Attributes.Mtime)
|
||||
}
|
||||
|
||||
func TestEventOption_WithOldOptionsAreNoOpsOnCreate(t *testing.T) {
|
||||
// Create events have no OldEntry; the WithOld* options must not
|
||||
// panic and must not leak fields into NewEntry.
|
||||
t0 := time.Unix(1700000000, 0)
|
||||
e := NewCreate("bk", "k", t0,
|
||||
WithOldSize(999),
|
||||
WithOldChunks(&filer_pb.FileChunk{FileId: "phantom"}),
|
||||
WithOldModTime(time.Unix(1, 0)),
|
||||
)
|
||||
assert.Nil(t, e.OldEntry)
|
||||
assert.Equal(t, uint64(0), e.NewEntry.Attributes.FileSize, "WithOldSize must not bleed to NewEntry")
|
||||
assert.Empty(t, e.NewEntry.Chunks, "WithOldChunks must not bleed to NewEntry")
|
||||
assert.Equal(t, t0.Unix(), e.NewEntry.Attributes.Mtime, "WithOldModTime must not bleed to NewEntry")
|
||||
}
|
||||
|
||||
func TestEventOption_LaterOverridesEarlier(t *testing.T) {
|
||||
// Apply order matters: later options win on the same field. Pins
|
||||
// the documented ordering so tests can compose default+override
|
||||
// patterns without surprises.
|
||||
e := NewCreate("bk", "k", time.Unix(0, 0),
|
||||
WithSize(100),
|
||||
WithSize(200),
|
||||
)
|
||||
assert.Equal(t, uint64(200), e.NewEntry.Attributes.FileSize)
|
||||
}
|
||||
|
||||
func TestMetaLogClock_DefaultStepIsOneSecond(t *testing.T) {
|
||||
c := NewMetaLogClock(time.Unix(1000, 0), 0)
|
||||
t1 := c.Next()
|
||||
t2 := c.Next()
|
||||
assert.Equal(t, time.Second, t2.Sub(t1))
|
||||
}
|
||||
|
||||
func TestMetaLogClock_CustomStep(t *testing.T) {
|
||||
c := NewMetaLogClock(time.Unix(1000, 0), 250*time.Millisecond)
|
||||
t1 := c.Next()
|
||||
t2 := c.Next()
|
||||
assert.Equal(t, 250*time.Millisecond, t2.Sub(t1))
|
||||
}
|
||||
|
||||
func TestMetaLogClock_PeekDoesNotAdvance(t *testing.T) {
|
||||
c := NewMetaLogClock(time.Unix(1000, 0), time.Second)
|
||||
first := c.Peek()
|
||||
again := c.Peek()
|
||||
assert.True(t, first.Equal(again), "Peek must not advance the clock")
|
||||
advanced := c.Next()
|
||||
assert.True(t, first.Equal(advanced), "Next returns what Peek returned")
|
||||
}
|
||||
|
||||
func TestMetaLogClock_ConcurrentNextNoRace(t *testing.T) {
|
||||
// Tests that produce events from many goroutines (e.g. fan-out
|
||||
// fixtures) need the clock to serialize without deadlock or
|
||||
// duplicate timestamps. -race catches a regression that drops
|
||||
// the lock.
|
||||
c := NewMetaLogClock(time.Unix(0, 0), time.Microsecond)
|
||||
const N = 64
|
||||
seen := sync.Map{}
|
||||
var wg sync.WaitGroup
|
||||
wg.Add(N)
|
||||
for i := 0; i < N; i++ {
|
||||
go func() {
|
||||
defer wg.Done()
|
||||
seen.Store(c.Next(), struct{}{})
|
||||
}()
|
||||
}
|
||||
wg.Wait()
|
||||
count := 0
|
||||
seen.Range(func(_, _ any) bool { count++; return true })
|
||||
assert.Equal(t, N, count, "every Next must produce a unique timestamp")
|
||||
}
|
||||
Loading…
Reference in New Issue
Block a user
