From 2a8045de58365254e16bcc1c1fc08b8c3ef002f5 Mon Sep 17 00:00:00 2001 From: Cristian Magherusan-Stanciu Date: Tue, 13 Oct 2026 03:43:36 +0200 Subject: [PATCH] test(purchase): prove the plan-step money path end to end and its crash points S9: an Azure step 2 priced from the unsuppressed stored row buys through executeAndFinalize; the same step priced from the suppressed row, or repriced between notify and run, is refused before the provider and the step fails. S11: crash before commit persists nothing; crash after the email re-sends and promotes without resolving again; after the provider call a re-drive reuses DeriveIdempotencyToken(lineage, index) for every provider, and providers without a duplicate guard are refused on re-drive. Refs #631 Co-Authored-By: Claude Sonnet 5.5 --- .../plan_step_crash_integration_test.go | 80 +++++++++ internal/purchase/plan_step_proofs_test.go | 163 ++++++++++++++++++ 2 files changed, 243 insertions(+) create mode 100644 internal/purchase/plan_step_crash_integration_test.go create mode 100644 internal/purchase/plan_step_proofs_test.go diff --git a/internal/purchase/plan_step_crash_integration_test.go b/internal/purchase/plan_step_crash_integration_test.go new file mode 100644 index 00000000..d90e1e9c --- /dev/null +++ b/internal/purchase/plan_step_crash_integration_test.go @@ -0,0 +1,80 @@ +//go:build integration + +package purchase + +import ( + "context" + "errors" + "sync/atomic" + "testing" + "time" + + "github.com/LeanerCloud/cloud-commitments-platform/internal/config" + "github.com/jackc/pgx/v5" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// S11 crash points on real PostgreSQL. The resolver and email sender are +// fixtures; the store is real with one injected failure each. The two other +// points of the table are covered elsewhere: "after commit, before email" by +// TestAttachIntegration_EmailFailureLeavesPendingAndNextTickDoesNotResolveAgain, +// and "after the provider call" by TestRedriveReusesDerivedTokensForEveryProvider. + +type suppressionFailingStore struct{ *config.PostgresStore } + +func (suppressionFailingStore) CreateSuppressionTx(context.Context, pgx.Tx, *config.PurchaseSuppression) error { + return errors.New("injected crash while writing suppressions") +} + +// rotateLosingStore drops the first token rotation, as a crash between the +// email going out and the notified promotion would. +type rotateLosingStore struct { + *config.PostgresStore + rotations atomic.Int32 +} + +func (s *rotateLosingStore) RotatePendingApprovalToken(ctx context.Context, id, hash string, exp time.Time) (bool, error) { + if s.rotations.Add(1) == 1 { + return false, errors.New("injected crash after the email") + } + return s.PostgresStore.RotatePendingApprovalToken(ctx, id, hash, exp) +} + +func TestCrash_BeforeCommitPersistsNothing(t *testing.T) { + ctx := context.Background() + store, pool := rampStepStore(ctx, t) + attachGlobal(ctx, t, store) + plan := attachPlanInDB(ctx, t, store, "crash-before-commit", time.Now().Add(24*time.Hour)) + sender := newCountingSender(ctx) + m := &Manager{config: suppressionFailingStore{store}, email: sender, resolver: &fakeResolver{recs: []config.RecommendationRecord{attachRec("rec-a", 3)}}} + + require.False(t, m.sendPlanNotification(ctx, plan)) + + assert.Empty(t, stepRowsInDB(ctx, t, store, plan.ID), "the row insert rolled back with the failed suppressions") + assert.Zero(t, suppressedTotal(ctx, t, pool)) + assert.Zero(t, sender.sends) +} + +func TestCrash_AfterEmailBeforeNotifiedResendsAndPromotes(t *testing.T) { + ctx := context.Background() + store, _ := rampStepStore(ctx, t) + attachGlobal(ctx, t, store) + plan := attachPlanInDB(ctx, t, store, "crash-after-email", time.Now().Add(24*time.Hour)) + sender := newCountingSender(ctx) + res := &fakeResolver{recs: []config.RecommendationRecord{attachRec("rec-a", 3)}} + m := &Manager{config: &rotateLosingStore{PostgresStore: store}, email: sender, resolver: res} + + require.False(t, m.sendPlanNotification(ctx, plan)) + rows := stepRowsInDB(ctx, t, store, plan.ID) + require.Len(t, rows, 1) + assert.Equal(t, "pending", rows[0].Status, "the scheduler refuses a pending row") + assert.Equal(t, 1, sender.sends) + + require.True(t, m.sendPlanNotification(ctx, plan), "the next tick re-sends and promotes") + rows = stepRowsInDB(ctx, t, store, plan.ID) + require.Len(t, rows, 1) + assert.Equal(t, "notified", rows[0].Status) + assert.Equal(t, 2, sender.sends) + assert.EqualValues(t, 1, res.calls.Load(), "no second resolve") +} diff --git a/internal/purchase/plan_step_proofs_test.go b/internal/purchase/plan_step_proofs_test.go new file mode 100644 index 00000000..272f8054 --- /dev/null +++ b/internal/purchase/plan_step_proofs_test.go @@ -0,0 +1,163 @@ +package purchase + +import ( + "context" + "sync" + "testing" + + "github.com/LeanerCloud/cloud-commitments-go/pkg/common" + "github.com/LeanerCloud/cloud-commitments-go/pkg/provider" + "github.com/LeanerCloud/cloud-commitments-platform/internal/config" + "github.com/LeanerCloud/cloud-commitments-platform/internal/planstep" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/mock" + "github.com/stretchr/testify/require" +) + +// Mock store, mock provider and fixture recommendations (nothing reaches a +// cloud). They cover the plan-step money path that the attach PR (#836) opened. + +// azureStepTwo is step 2 of an Azure plan: the stored (unsuppressed) row holds +// 20 units, step 1 suppressed 10, and the plan buys 25% of what is shown. +func azureStepTwo(t *testing.T) (stored config.RecommendationRecord, resolved config.RecommendationRecord) { + t.Helper() + acct := "acct-T" + stored = config.RecommendationRecord{ + ID: "azure-1", Provider: "azure", Service: "compute", Region: "westeurope", ResourceType: "Standard_D2s_v3", + Term: 1, Payment: "monthly", Count: 20, UpfrontCost: 1234.56, MonthlyCost: new(77.31), Savings: 400.17, CloudAccountID: &acct, Selected: true, + } + shown := stored + shown.Count = 10 + plan := &config.PurchasePlan{ID: "p", Services: map[string]config.ServiceConfig{"azure:compute": {Provider: "azure", Service: "compute", Enabled: true, Term: 1, Payment: "no-upfront", Coverage: 100}}, RampSchedule: config.RampSchedule{PercentPerStep: 25}} + recs, err := planstep.Resolve(context.Background(), azureStepSource{[]config.RecommendationRecord{shown}}, azureStepStore{[]config.RecommendationRecord{stored}}, plan, []string{acct}) + require.NoError(t, err) + require.Len(t, recs, 1) + require.Equal(t, 2, recs[0].Count) + return stored, recs[0] +} + +// S9 through executeAndFinalize (the funnel every executor uses): a step priced +// from the unsuppressed stored row buys; the same step priced from the +// suppressed row is refused before the provider; a re-collection that changes +// the price between notify and run is refused and the step fails. +func TestAzureStepTwo_ExecuteAndFinalize(t *testing.T) { + stored, resolved := azureStepTwo(t) + // The pre-fix shape: the count lowered by suppression and sizing, the + // totals left at the unsuppressed 20 units. + suppressedPriced := resolved + suppressedPriced.UpfrontCost = stored.UpfrontCost + suppressedPriced.MonthlyCost = stored.MonthlyCost + repriced := stored + repriced.UpfrontCost++ // Azure re-collection between notify and run + + for _, tc := range []struct { + name string + rec config.RecommendationRecord + current config.RecommendationRecord + wantCalls int + wantErr bool + }{ + {"resolved from the stored row buys", resolved, stored, 1, false}, + {"priced from the suppressed row is refused", suppressedPriced, stored, 0, true}, + {"price changed after notify is refused", resolved, repriced, 0, true}, + } { + t.Run(tc.name, func(t *testing.T) { + exec := armedExecution("azure", "compute", 0, "running") + exec.Recommendations = []config.RecommendationRecord{tc.rec} + mgr, store, recorder := armedHarness(t, common.ServiceCompute) + store.On("ListStoredRecommendations", mock.Anything, config.RecommendationFilter{Provider: "azure", RequireAzurePricingMigration: true}). + Return([]config.RecommendationRecord{tc.current}, nil).Maybe() + + err := mgr.executeAndFinalize(context.Background(), exec) + + assert.Equal(t, tc.wantCalls, recorder.count()) + if tc.wantErr { + require.ErrorIs(t, err, ErrStaleAzurePricing) + assert.Equal(t, "failed", exec.Status, "the step fails instead of buying a repriced commitment") + } else { + require.NoError(t, err) + assert.Equal(t, "completed", exec.Status) + } + }) + } +} + +// tokenCapture runs processPurchaseRecommendations for exec on the given +// provider and returns the IdempotencyToken each +// PurchaseCommitment call received, keyed by resource type. +func tokenCapture(t *testing.T, providerName string, exec *config.PurchaseExecution) map[string]string { + t.Helper() + ctx := context.Background() + store, sender, factory := new(MockConfigStore), new(MockEmailSender), new(MockProviderFactory) + prov, svc := new(MockProvider), new(MockServiceClient) + store.SavePurchaseExecutionFn = func(context.Context, *config.PurchaseExecution) error { return nil } + store.On("SavePurchaseHistory", ctx, mock.Anything).Return(nil) + factory.On("CreateAndValidateProvider", mock.Anything, providerName, mock.Anything).Return(prov, nil) + prov.On("GetServiceClient", mock.Anything, mock.Anything, mock.Anything).Return(svc, nil) + var mu sync.Mutex + tokens := map[string]string{} + svc.On("PurchaseCommitment", mock.Anything, mock.AnythingOfType("common.Recommendation"), mock.AnythingOfType("common.PurchaseOptions")). + Run(func(a mock.Arguments) { + mu.Lock() + tokens[a.Get(1).(common.Recommendation).ResourceType] = a.Get(2).(common.PurchaseOptions).IdempotencyToken + mu.Unlock() + }).Return(common.PurchaseResult{Success: true, CommitmentID: "c"}, nil) + m := &Manager{config: store, email: sender, providerFactory: factory} + _, _, errs, err := m.processPurchaseRecommendations(ctx, exec, &config.PurchasePlan{Name: "p"}, "acct", (*provider.ProviderConfig)(nil)) + require.NoError(t, err) + require.Empty(t, errs) + return tokens +} + +func sortedAttachedExec(provider, service string) *config.PurchaseExecution { + recs := []config.RecommendationRecord{ + {ID: "rec-a", Provider: provider, Service: service, ResourceType: "type-a", Region: "r", Count: 1, Term: 1, UpfrontCost: 1, Selected: true}, + {ID: "rec-b", Provider: provider, Service: service, ResourceType: "type-b", Region: "r", Count: 2, Term: 1, UpfrontCost: 2, Selected: true}, + } + return &config.PurchaseExecution{ExecutionID: "exec-1", IdempotencyKey: "lineage-1", Recommendations: recs} +} + +// S11, crash after the provider call: the reaper and re-drive run the same +// persisted (sorted) recommendations again, and every recommendation receives +// DeriveIdempotencyToken(lineage, index) both times, so a provider that dedupes +// on the token buys nothing twice. The token is also what reaches each +// provider's client (its use there is proven in the pinned cloud-commitments-go +// module; see the PR body). +func TestRedriveReusesDerivedTokensForEveryProvider(t *testing.T) { + for _, tc := range []struct { + provider, service string + }{{"aws", "rds"}, {"aws", "savings-plans"}, {"azure", "compute"}, {"gcp", "compute"}} { + t.Run(tc.provider+"/"+tc.service, func(t *testing.T) { + first := tokenCapture(t, tc.provider, sortedAttachedExec(tc.provider, tc.service)) + redrive := tokenCapture(t, tc.provider, sortedAttachedExec(tc.provider, tc.service)) + + want := map[string]string{ + "type-a": common.DeriveIdempotencyToken("lineage-1", 0), + "type-b": common.DeriveIdempotencyToken("lineage-1", 1), + } + assert.Equal(t, want, first) + assert.Equal(t, first, redrive, "a re-drive reuses the identical tokens") + assert.NotEqual(t, first["type-a"], first["type-b"], "each recommendation gets its own token") + }) + } +} + +// Providers whose purchase API has no usable duplicate guard are refused on +// re-drive instead of bought again (the double-buy list in the PR body). +func TestRedriveRefusedWhereNoProviderGuardExists(t *testing.T) { + for _, rec := range []config.RecommendationRecord{ + {Provider: "azure", Service: "savingsplans"}, + {Provider: "aws", Service: "ec2"}, + {Provider: "aws", Service: "redshift"}, + {Provider: "oracle", Service: "compute"}, + } { + assert.NotEmpty(t, recRedriveRefusalReason(rec), "%s/%s", rec.Provider, rec.Service) + } + for _, rec := range []config.RecommendationRecord{ + {Provider: "aws", Service: "rds"}, + {Provider: "azure", Service: "compute"}, + {Provider: "gcp", Service: "compute"}, + } { + assert.Empty(t, recRedriveRefusalReason(rec), "%s/%s", rec.Provider, rec.Service) + } +}