package transfer import ( "errors" "testing" "time" ) type fakeClock struct { now time.Time } func (f *fakeClock) Now() time.Time { return f.now } func (f *fakeClock) Advance(d time.Duration) { f.now = f.now.Add(d) } func TestManagerOutgoingLifecycle(t *testing.T) { clock := &fakeClock{now: time.Unix(100, 0)} manager := NewManagerWithClock(clock.Now) snapshot, err := manager.StartOutgoing(Descriptor{ ID: "tx-1", Size: 100, Checksum: "sum-1", Metadata: Metadata{"kind": "file"}, }) if err != nil { t.Fatalf("StartOutgoing failed: %v", err) } if got, want := snapshot.State, StateNegotiating; got != want { t.Fatalf("start state = %v, want %v", got, want) } if got, want := snapshot.Channel, DataChannel; got != want { t.Fatalf("channel = %q, want %q", got, want) } clock.Advance(time.Second) if _, err := manager.Activate("tx-1"); err != nil { t.Fatalf("Activate failed: %v", err) } clock.Advance(time.Second) if _, err := manager.RecordSend("tx-1", 60); err != nil { t.Fatalf("RecordSend failed: %v", err) } clock.Advance(time.Second) if _, err := manager.SetAckedBytes("tx-1", 40); err != nil { t.Fatalf("SetAckedBytes failed: %v", err) } if _, err := manager.RecordRetry("tx-1"); err != nil { t.Fatalf("RecordRetry failed: %v", err) } if _, err := manager.RecordTimeout("tx-1"); err != nil { t.Fatalf("RecordTimeout failed: %v", err) } if _, err := manager.Pause("tx-1"); err != nil { t.Fatalf("Pause failed: %v", err) } clock.Advance(time.Second) if _, err := manager.Resume("tx-1", 40); err != nil { t.Fatalf("Resume failed: %v", err) } if _, err := manager.BeginCommit("tx-1"); err != nil { t.Fatalf("BeginCommit failed: %v", err) } clock.Advance(time.Second) snapshot, err = manager.Complete("tx-1") if err != nil { t.Fatalf("Complete failed: %v", err) } if got, want := snapshot.State, StateDone; got != want { t.Fatalf("complete state = %v, want %v", got, want) } if got, want := snapshot.SentBytes, int64(60); got != want { t.Fatalf("sent bytes = %d, want %d", got, want) } if got, want := snapshot.AckedBytes, int64(40); got != want { t.Fatalf("acked bytes = %d, want %d", got, want) } if got, want := snapshot.InflightBytes, int64(20); got != want { t.Fatalf("inflight bytes = %d, want %d", got, want) } if got, want := snapshot.RetryCount, 1; got != want { t.Fatalf("retry count = %d, want %d", got, want) } if got, want := snapshot.TimeoutCount, 1; got != want { t.Fatalf("timeout count = %d, want %d", got, want) } if _, err := manager.SetStage("tx-1", "chunk"); err != nil { t.Fatalf("SetStage failed: %v", err) } if _, err := manager.SetFailureStage("tx-1", "chunk"); err != nil { t.Fatalf("SetFailureStage failed: %v", err) } if snapshot.CompletedAt == 0 { t.Fatal("completed timestamp should be set") } if got := snapshot.Metadata["kind"]; got != "file" { t.Fatalf("metadata kind = %q, want file", got) } snapshot, ok := manager.Snapshot("tx-1") if !ok { t.Fatal("snapshot should still exist") } if got, want := snapshot.Stage, "chunk"; got != want { t.Fatalf("stage = %q, want %q", got, want) } if got, want := snapshot.LastFailureStage, "chunk"; got != want { t.Fatalf("last failure stage = %q, want %q", got, want) } } func TestManagerIncomingResumeAndVerify(t *testing.T) { clock := &fakeClock{now: time.Unix(200, 0)} manager := NewManagerWithClock(clock.Now) snapshot, err := manager.StartIncoming(Descriptor{ ID: "rx-1", Channel: ControlChannel, Size: 64, }) if err != nil { t.Fatalf("StartIncoming failed: %v", err) } if got, want := snapshot.State, StatePrepared; got != want { t.Fatalf("prepared state = %v, want %v", got, want) } clock.Advance(time.Second) snapshot, err = manager.Resume("rx-1", 16) if err != nil { t.Fatalf("Resume failed: %v", err) } if got, want := snapshot.ReceivedBytes, int64(16); got != want { t.Fatalf("received bytes after resume = %d, want %d", got, want) } if _, err := manager.RecordReceive("rx-1", 20); err != nil { t.Fatalf("RecordReceive failed: %v", err) } if _, err := manager.BeginVerify("rx-1"); err != nil { t.Fatalf("BeginVerify failed: %v", err) } clock.Advance(time.Second) snapshot, err = manager.Complete("rx-1") if err != nil { t.Fatalf("Complete failed: %v", err) } if got, want := snapshot.State, StateDone; got != want { t.Fatalf("complete state = %v, want %v", got, want) } if got, want := snapshot.ReceivedBytes, int64(36); got != want { t.Fatalf("received bytes = %d, want %d", got, want) } if got, want := snapshot.Channel, ControlChannel; got != want { t.Fatalf("channel = %q, want %q", got, want) } } func TestManagerValidatesIDsAndSortedSnapshots(t *testing.T) { manager := NewManager() if _, err := manager.StartOutgoing(Descriptor{}); !errors.Is(err, ErrTransferIDEmpty) { t.Fatalf("empty id error = %v, want %v", err, ErrTransferIDEmpty) } if _, err := manager.StartOutgoing(Descriptor{ID: "b"}); err != nil { t.Fatalf("StartOutgoing b failed: %v", err) } if _, err := manager.StartIncoming(Descriptor{ID: "a"}); err != nil { t.Fatalf("StartIncoming a failed: %v", err) } if _, err := manager.StartOutgoing(Descriptor{ID: "b"}); !errors.Is(err, ErrTransferExists) { t.Fatalf("duplicate id error = %v, want %v", err, ErrTransferExists) } if _, err := manager.RecordSend("missing", 1); !errors.Is(err, ErrTransferNotFound) { t.Fatalf("missing transfer error = %v, want %v", err, ErrTransferNotFound) } if _, err := manager.RecordReceive("a", -1); !errors.Is(err, ErrTransferBytesInvalid) { t.Fatalf("negative bytes error = %v, want %v", err, ErrTransferBytesInvalid) } snapshots := manager.Snapshots() if len(snapshots) != 2 { t.Fatalf("snapshot count = %d, want 2", len(snapshots)) } if snapshots[0].ID != "a" || snapshots[1].ID != "b" { t.Fatalf("snapshot order = [%s %s], want [a b]", snapshots[0].ID, snapshots[1].ID) } }