amadan
e2e: smoke assertions for the copy point, the clock and Serial
TestBridgeCopyFiresOncePerStagerRound stages a move and finds copy: firing once per stage request on the stager, never on the target or a pass with no request. TestBridgeBoxesReportThemselves now finds the snapshots stamped at the wall clock before any Set and at the bridge clock after Advance, and a bump made during a pass consumed by Serial's second pass. Both fail on 4456d778.
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
1 file changed,
+120
−16
internal/e2e/selfserve/bridge_test.go+120 −16
diff --git a/internal/e2e/selfserve/bridge_test.go b/internal/e2e/selfserve/bridge_test.go| index 30ac54e26..1ad42da60 100644 |
| --- a/internal/e2e/selfserve/bridge_test.go |
| +++ b/internal/e2e/selfserve/bridge_test.go |
| @@ -9,12 +9,16 @@ package selfserve |
| import ( |
| "context" |
| + "encoding/json" |
| "path/filepath" |
| "slices" |
| "testing" |
| + "time" |
| "github.com/carlosframework/platform/internal/console" |
| "github.com/carlosframework/platform/internal/host" |
| + "github.com/carlosframework/platform/internal/moves" |
| + "github.com/carlosframework/platform/internal/provision" |
| "github.com/carlosframework/platform/internal/release" |
| "github.com/carlosframework/platform/internal/status" |
| ) |
| @@ -23,8 +27,10 @@ import ( |
| // store a console uses, with that console beside it, one pass and one |
| // status tick on both boxes leave each box's status block naming its own |
| // box id, instance id and lane, and the source's naming the instance's host |
| -// facts; the serial is consumed; and the pass and status fault points fire |
| -// once per box. |
| +// facts, stamped at the wall clock; the pass and status fault points fire |
| +// once per box; a bump made during a pass is consumed by Serial's next |
| +// pass; and once the test advances the clock the status snapshot is |
| +// stamped at the bridge's time, not the wall's. |
| func TestBridgeBoxesReportThemselves(t *testing.T) { |
| ctx := context.Background() |
| store := release.NewDirStore(t.TempDir()) |
| @@ -63,20 +69,11 @@ func TestBridgeBoxesReportThemselves(t *testing.T) { |
| if len(fired) != 4 { |
| t.Errorf("fault points fired: %v, want the pass and status points of both boxes", fired) |
| } |
| - if err := release.BumpSerial(ctx, store, e.Clock().Now()); err != nil { |
| - t.Fatal(err) |
| - } |
| - if err := e.Serial(ctx); err != nil { |
| - t.Fatal(err) |
| - } |
| - |
| - boxes, complete, err := status.LoadBoxesComplete(ctx, store) |
| - if err != nil || !complete { |
| - t.Fatalf("status boxes: complete %v, %v", complete, err) |
| - } |
| - byLabel := map[string]*status.BoxMetrics{} |
| - for _, b := range boxes { |
| - byLabel[b.Label] = b.Metrics |
| + boxes, byLabel, generated := loadBoxes(t, store) |
| + for label, at := range generated { |
| + if d := time.Since(at); d < 0 || d > time.Second { |
| + t.Errorf("%s's snapshot is stamped %v, %v from the wall clock before any Set", label, at, d) |
| + } |
| } |
| for label, lane := range map[string]string{host.E2ESourceBox: "data-uk", host.E2ETargetBox: "data-ie"} { |
| h := e.Box(label) |
| @@ -104,4 +101,111 @@ func TestBridgeBoxesReportThemselves(t *testing.T) { |
| if e.Box(host.E2ESourceBox).DataBucket() != "data-ie" || e.Box(host.E2ETargetBox).DataBucket() != "data-uk" { |
| t.Errorf("data buckets %q and %q", e.Box(host.E2ESourceBox).DataBucket(), e.Box(host.E2ETargetBox).DataBucket()) |
| } |
| + |
| + // Serial: a bump made during the first pass is the next pass's to |
| + // consume, and Serial runs it. |
| + _, etag0, err := release.ReadSerial(ctx, store) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + passes, bumped := 0, false |
| + e.Fault(host.FaultPass+host.E2ESourceBox, func() { |
| + passes++ |
| + if !bumped { |
| + bumped = true |
| + if err := release.BumpSerial(ctx, store, time.Now()); err != nil { |
| + t.Error(err) |
| + } |
| + } |
| + }) |
| + if err := e.Serial(ctx); err != nil { |
| + t.Fatal(err) |
| + } |
| + e.Fault(host.FaultPass+host.E2ESourceBox, nil) |
| + if _, etag1, err := release.ReadSerial(ctx, store); err != nil || etag1 == etag0 || passes != 2 { |
| + t.Errorf("Serial ran %d passes and the serial's etag is %q (was %q, %v); want the bump consumed by a second pass", passes, etag1, etag0, err) |
| + } |
| + |
| + // The clock: once the test advances it, the status snapshot is stamped |
| + // at the bridge's time. |
| + e.Clock().Advance(2 * time.Hour) |
| + if err := e.Status(ctx); err != nil { |
| + t.Fatal(err) |
| + } |
| + _, _, generated = loadBoxes(t, store) |
| + for label, at := range generated { |
| + if !at.Equal(e.Clock().Now()) { |
| + t.Errorf("%s's snapshot is stamped %v after Advance, want the bridge clock's %v", label, at, e.Clock().Now()) |
| + } |
| + } |
| +} |
| + |
| +// TestBridgeCopyFiresOncePerStagerRound: in the two-box world, a staged |
| +// move's stager runs one copy round per stage request, and copy: fires once |
| +// at each round's start, on the stager only, never on a pass with no new |
| +// request. |
| +func TestBridgeCopyFiresOncePerStagerRound(t *testing.T) { |
| + ctx := context.Background() |
| + store := release.NewDirStore(t.TempDir()) |
| + e := host.NewE2E(t, store) |
| + copies := map[string]int{} |
| + for _, label := range []string{host.E2ESourceBox, host.E2ETargetBox} { |
| + p := host.FaultCopy + label |
| + e.Fault(p, func() { copies[p]++ }) |
| + } |
| + const id = "mv_stagedstagedstagedstagedab" |
| + move := func(mut func(*provision.Record) error) { |
| + t.Helper() |
| + if _, err := provision.SetRecordMove(ctx, store, e.Account(), host.E2EApp, host.E2EHost, mut); err != nil { |
| + t.Fatal(err) |
| + } |
| + } |
| + src, dst := host.FaultCopy+host.E2ESourceBox, host.FaultCopy+host.E2ETargetBox |
| + move(provision.OfflineStage(id, "", host.E2ETargetRegion, "op@example.com", |
| + provision.StageTarget{Stager: host.E2ESourceBox, Bucket: e.Box(host.E2ETargetBox).DataBucket(), Region: "eu-west-2"}, e.Clock().Now(), "")) |
| + for round := 1; round <= 3; round++ { |
| + if round > 1 { |
| + move(provision.OfflineRestage(id, e.Clock().Now().Add(time.Duration(round)*time.Second), provision.Receipt{})) |
| + } |
| + for range 2 { // the second pass has no new request: no round |
| + if err := e.Pass(ctx); err != nil { |
| + t.Fatal(err) |
| + } |
| + } |
| + if copies[src] != round || copies[dst] != 0 { |
| + t.Fatalf("after stage request %d: copy fired %d times on the stager and %d on the target, want %d and 0", round, copies[src], copies[dst], round) |
| + } |
| + } |
| + if !stagerReported(t, store, id) { |
| + t.Fatal("the stager never reported a staged round") |
| + } |
| +} |
| + |
| +// stagerReported is whether the stager's report for move id says staged. |
| +func stagerReported(t *testing.T, store release.Store, id string) bool { |
| + t.Helper() |
| + body, _, err := store.Get(context.Background(), moves.OfflineReportKey(host.E2EHost, id, moves.OfflineRoleStager, host.E2ESourceBox)) |
| + if err != nil { |
| + t.Fatalf("the stager report: %v", err) |
| + } |
| + var r moves.OfflineReport |
| + if err := json.Unmarshal(body, &r); err != nil { |
| + t.Fatal(err) |
| + } |
| + return r.Phase == moves.OfflinePhaseStaged |
| +} |
| + |
| +// loadBoxes reads every box's status snapshot: the boxes, their metrics and |
| +// their generated-at stamps by label. |
| +func loadBoxes(t *testing.T, store release.Store) ([]status.Box, map[string]*status.BoxMetrics, map[string]time.Time) { |
| + t.Helper() |
| + boxes, complete, err := status.LoadBoxesComplete(context.Background(), store) |
| + if err != nil || !complete { |
| + t.Fatalf("status boxes: complete %v, %v", complete, err) |
| + } |
| + metrics, generated := map[string]*status.BoxMetrics{}, map[string]time.Time{} |
| + for _, b := range boxes { |
| + metrics[b.Label], generated[b.Label] = b.Metrics, b.GeneratedAt |
| + } |
| + return boxes, metrics, generated |
| } |