| 1 | package console |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "database/sql" |
| 6 | "encoding/json" |
| 7 | "errors" |
| 8 | "testing" |
| 9 | "time" |
| 10 | |
| 11 | "github.com/carlosframework/platform/internal/hostops" |
| 12 | "github.com/carlosframework/platform/internal/release" |
| 13 | ) |
| 14 | |
| 15 | func TestClaimPurgeStageSettlesOnlyObservedEffects(t *testing.T) { |
| 16 | transient := errors.New("transient read") |
| 17 | for _, tc := range []struct { |
| 18 | name string |
| 19 | write bool |
| 20 | callErr error |
| 21 | state, result string |
| 22 | }{ |
| 23 | {"read failure", false, transient, "complete", "not-applied"}, |
| 24 | {"pending before effect", false, hostops.ErrPending, "complete", "not-applied"}, |
| 25 | {"pending after joined effect", true, hostops.ErrPending, "complete", "joined"}, |
| 26 | {"ambiguous effect", true, transient, "uncertain", ""}, |
| 27 | {"pending joined with transport error", true, errors.Join(hostops.ErrPending, transient), "uncertain", ""}, |
| 28 | {"nested not-applied after earlier effect", true, errors.Join(hostops.ErrNotApplied, transient), "uncertain", ""}, |
| 29 | } { |
| 30 | t.Run(tc.name, func(t *testing.T) { |
| 31 | cx := newClaimCase(t) |
| 32 | cx.purgeWorld(t) |
| 33 | p := cx.srv.cfg.ClaimPurge |
| 34 | id, err := p.Request(t.Context(), "bab", "titogo", "stage-test", "owner", claimFxOwner) |
| 35 | if id == "" || err != nil && !errors.Is(err, hostops.ErrPending) { |
| 36 | t.Fatalf("request: %s %v", id, err) |
| 37 | } |
| 38 | a, err := hostops.ReadCanonicalAdmission(t.Context(), cx.srv.cfg.Releases, "bab", "titogo") |
| 39 | if err != nil { |
| 40 | t.Fatal(err) |
| 41 | } |
| 42 | h, err := cx.srv.cfg.ClaimAuthority.Control.ResumeExclusive(t.Context(), "bab", "titogo", a.Value.AppID, id, hostops.OpPurging, cx.srv.claimOwner) |
| 43 | if err != nil { |
| 44 | t.Fatal(err) |
| 45 | } |
| 46 | call := func(ctx context.Context) error { |
| 47 | if tc.write { |
| 48 | if _, err := cx.srv.cfg.Releases.Put(ctx, "apps/bab/titogo/stage-test.json", []byte(`{}`), ""); err != nil { |
| 49 | t.Fatal(err) |
| 50 | } |
| 51 | } |
| 52 | return tc.callErr |
| 53 | } |
| 54 | err = p.stage(t.Context(), h, "witness", call) |
| 55 | if !errors.Is(err, tc.callErr) { |
| 56 | t.Fatalf("lost error: %v", err) |
| 57 | } |
| 58 | a, err = hostops.ReadCanonicalAdmission(t.Context(), cx.srv.cfg.Releases, "bab", "titogo") |
| 59 | if err != nil { |
| 60 | t.Fatal(err) |
| 61 | } |
| 62 | found := false |
| 63 | for _, e := range a.Value.Effects { |
| 64 | if e.Resource == "purge-stage:witness" { |
| 65 | found = true |
| 66 | if e.State != tc.state || e.ResultReference != tc.result { |
| 67 | t.Fatalf("settlement=%+v, want %s/%s", e, tc.state, tc.result) |
| 68 | } |
| 69 | } |
| 70 | } |
| 71 | if !found { |
| 72 | t.Fatal("no stage effect") |
| 73 | } |
| 74 | called := false |
| 75 | err = p.stage(t.Context(), h, "witness", func(context.Context) error { called = true; return nil }) |
| 76 | if tc.state == "complete" { |
| 77 | if err != nil || !called { |
| 78 | t.Fatalf("settled stage cannot retry: %v %v", called, err) |
| 79 | } |
| 80 | } else if err == nil || called { |
| 81 | t.Fatalf("uncertain stage retried: %v %v", called, err) |
| 82 | } |
| 83 | }) |
| 84 | } |
| 85 | } |
| 86 | |
| 87 | func TestClaimAutomaticPurgeRemovesUntrashedProjection(t *testing.T) { |
| 88 | cx := newClaimCase(t) |
| 89 | cx.purgeWorld(t) |
| 90 | ctx := t.Context() |
| 91 | a, err := hostops.ReadCanonicalAdmission(ctx, cx.rel, "bab", "titogo") |
| 92 | if err != nil { |
| 93 | t.Fatal(err) |
| 94 | } |
| 95 | scope := claimSQLScope{AccountID: OpsAccountID, App: "titogo", AppID: a.Value.AppID, Mode: claimModeEffect} |
| 96 | if err = cx.st.withClaimSQL(gateCtx(t), []claimSQLScope{scope}, allowAllClaimSQL, func(st *Store) error { |
| 97 | _, err := st.db.Exec(`UPDATE apps SET deleted_at=0 WHERE account_id=? AND app='titogo'`, OpsAccountID) |
| 98 | return err |
| 99 | }); err != nil { |
| 100 | t.Fatal(err) |
| 101 | } |
| 102 | key := release.AppTombstoneKey("bab", "titogo") |
| 103 | body, etag, err := cx.rel.Get(ctx, key) |
| 104 | if err != nil { |
| 105 | t.Fatal(err) |
| 106 | } |
| 107 | var tomb release.AppTombstone |
| 108 | if err = json.Unmarshal(body, &tomb); err != nil { |
| 109 | t.Fatal(err) |
| 110 | } |
| 111 | tomb.DeletedAt = time.Now().Add(-release.TrashCooloff - time.Hour) |
| 112 | body, err = json.Marshal(tomb) |
| 113 | if err != nil { |
| 114 | t.Fatal(err) |
| 115 | } |
| 116 | if _, err = cx.rel.Put(ctx, key, body, etag); err != nil { |
| 117 | t.Fatal(err) |
| 118 | } |
| 119 | p := cx.srv.cfg.ClaimPurge |
| 120 | id, err := p.Request(ctx, "bab", "titogo", "auto-projection", "automatic", "auto-purge") |
| 121 | if id == "" || err != nil && !errors.Is(err, hostops.ErrPending) { |
| 122 | t.Fatalf("request %s: %v", id, err) |
| 123 | } |
| 124 | cx.publishPurgeCensus(t, id) |
| 125 | for i := 0; i < 12 && cx.ownsPurge(t, "bab", "titogo", id); i++ { |
| 126 | if err = p.Resume(ctx, "bab", "titogo", id, a.Value.AppID); err != nil && !errors.Is(err, hostops.ErrPending) { |
| 127 | t.Fatal(err) |
| 128 | } |
| 129 | } |
| 130 | if cx.ownsPurge(t, "bab", "titogo", id) { |
| 131 | t.Fatal("automatic purge did not finish") |
| 132 | } |
| 133 | if _, err = cx.st.ClaimAppID(OpsAccountID, "titogo"); !errors.Is(err, sql.ErrNoRows) { |
| 134 | t.Fatalf("old row survived: %v", err) |
| 135 | } |
| 136 | } |
| 137 | |
amadan