amadan

carlos / platform Public

Clone 
git clone https://amadan.net/carlos/platform

Plain git — no account needed to clone.

Download

Download this file

1package console
2
3import (
4 "context"
5 "errors"
6 "fmt"
7 "net/http"
8 "os"
9 "testing"
10 "time"
11
12 "github.com/carlosframework/platform/internal/dnsrec"
13 "github.com/carlosframework/platform/internal/hostops"
14 "github.com/carlosframework/platform/internal/mail"
15 "github.com/carlosframework/platform/internal/objstore"
16 "github.com/carlosframework/platform/internal/release"
17)
18
19func TestClaimPurgeStoreInventoryRefusesBeforeEffects(t *testing.T) {
20 for _, tc := range []struct{ name, key, body string }{
21 {"malformed key", "stores/bad/nested.json", `{"schema":1,"app":"titogo"}`},
22 {"unsupported schema", "stores/prod.json", `{"schema":2,"app":"titogo","environment":"prod"}`},
23 {"identity mismatch", "stores/prod.json", `{"schema":1,"app":"other","environment":"prod"}`},
24 {"default unsupported schema", "store.json", `{"schema":2,"app":"titogo"}`},
25 {"default identity mismatch", "store.json", `{"schema":1,"app":"other"}`},
26 } {
27 for _, stage := range []string{"request", "teardown"} {
28 t.Run(tc.name+"/"+stage, func(t *testing.T) {
29 cx := newClaimCase(t)
30 cx.purgeWorld(t)
31 p := cx.srv.cfg.ClaimPurge
32 ctx := t.Context()
33 const id = "strict-store-purge"
34 appID := cx.appIDs[claimAppRef{Account: "bab", App: "titogo"}]
35 var h *hostops.OperationHandle
36 if stage == "teardown" {
37 got, err := p.Request(ctx, "bab", "titogo", id, "owner", claimFxOwner)
38 if got != id || err != nil && !errors.Is(err, hostops.ErrPending) {
39 t.Fatalf("request: %s %v", got, err)
40 }
41 var err2 error
42 h, err2 = p.Authority.Control.ResumeExclusive(ctx, "bab", "titogo", appID, id, hostops.OpPurging, p.Owner)
43 if err2 != nil {
44 t.Fatal(err2)
45 }
46 j, _, err := hostops.ReadPurgeJournal(ctx, cx.rel, "bab", "titogo", id, appID)
47 if err != nil {
48 t.Fatal(err)
49 }
50 j.Phase = "claimed" // Stage fixture: host barriers are tested separately.
51 key, _ := hostops.PurgeKey("bab", "titogo")
52 putMailJSON(t, cx.rel, key, j)
53 }
54 key := release.AppPrefix("bab", "titogo") + tc.key
55 if _, err := cx.rel.Put(ctx, key, []byte(tc.body), ""); err != nil {
56 t.Fatal(err)
57 }
58 mark := cx.rec.mark()
59 var err error
60 if stage == "request" {
61 _, err = p.Request(ctx, "bab", "titogo", id, "owner", claimFxOwner)
62 journal, _ := hostops.PurgeKey("bab", "titogo")
63 if _, _, e := cx.rel.Get(ctx, journal); !errors.Is(e, os.ErrNotExist) {
64 t.Errorf("invalid inventory reserved purge: %v", e)
65 }
66 } else {
67 err = p.stage(ctx, h, "teardown", func(ctx context.Context) error {
68 scope := mail.ClaimMailScope{Account: "bab", App: "titogo", AppID: appID, Operation: id}
69 return mail.WithClaimMailResource(mail.WithClaimMailScope(ctx, scope), cx.srv.claimMailStore(), func(held context.Context) error {
70 return p.teardown(held, h, "bab", "titogo", id, appID)
71 })
72 })
73 }
74 if err == nil || errors.Is(err, hostops.ErrPending) {
75 t.Errorf("invalid inventory was not refused: %v", err)
76 }
77 for _, e := range cx.rec.since(mark) {
78 if !e.Control && (e.Unscoped || len(e.Scopes) > 0) {
79 t.Errorf("effect before validation: %v", e)
80 }
81 }
82 if _, _, e := cx.rel.Get(ctx, key); e != nil {
83 t.Errorf("invalid estate record removed: %v", e)
84 }
85 })
86 }
87 }
88}
89
90func TestClaimStoreReadersValidateEnabledMode(t *testing.T) {
91 cx := newClaimCase(t)
92 for _, tc := range []struct{ name, body string }{
93 {"schema", `{"schema":2,"app":"titogo"}`},
94 {"identity", `{"schema":1,"app":"other"}`},
95 } {
96 t.Run(tc.name, func(t *testing.T) {
97 key := objstore.Key("bab", "titogo")
98 _, tag, _ := cx.rel.Get(t.Context(), key)
99 if _, err := cx.rel.Put(t.Context(), key, []byte(tc.body), tag); err != nil {
100 t.Fatal(err)
101 }
102 if _, _, err := objstore.GetIn(t.Context(), cx.srv.cfg.Releases, "bab", "titogo", ""); err == nil {
103 t.Fatal("enabled reader accepted invalid record")
104 }
105 if _, _, err := objstore.GetIn(t.Context(), cx.rel, "bab", "titogo", ""); err != nil {
106 t.Fatalf("legacy reader changed: %v", err)
107 }
108 mark := cx.rec.mark()
109 if err := objstore.TeardownIn(t.Context(), cx.srv.cfg.Releases, cx.srv.cfg.StoreProvisioner, nil, "bab", "titogo", ""); err == nil {
110 t.Fatal("teardown accepted invalid record")
111 }
112 for _, e := range cx.rec.since(mark) {
113 if !e.Control {
114 t.Errorf("teardown effect: %v", e)
115 }
116 }
117 })
118 }
119}
120
121func TestClaimBackendLegacyDomainClaims(t *testing.T) {
122 for _, schema := range []int{0, 1, -1, 2} {
123 for _, bad := range []string{"", "domain", "account", "app"} {
124 t.Run(string(rune('a'+schema+1))+"/"+bad, func(t *testing.T) {
125 raw := release.NewDirStore(t.TempDir())
126 ctx := mailGuardPermits(t, raw, hostops.AppScope{Account: "bab", App: "titogo"})
127 w := claimTestWriter(t, raw)
128 lease, err := w.AcquireMailGate(ctx, hostops.MailGateOwner{Console: "c", Enrollment: "e", Controller: "k"}, hostops.MailGateScope{})
129 if err != nil {
130 t.Fatal(err)
131 }
132 ctx = hostops.WithMailGateLease(ctx, lease)
133 const domain = "legacy.example.com"
134 c := mail.DomainClaim{Schema: schema, Domain: domain, Account: "bab", App: "titogo", CreatedAt: time.Now()}
135 switch bad {
136 case "domain":
137 c.Domain = "other.example.com"
138 case "account":
139 c.Account = ""
140 case "app":
141 c.App = "../other"
142 }
143 putMailJSON(t, raw, mail.DomainClaimKey(domain), c)
144 mem := mail.NewMemProvisioner("1")
145 dns := dnsrec.NewMemWriter()
146 g := backendGuard{data: raw, control: w}
147 p := newClaimMailProvisioner(mem, g)
148 d := newClaimDNSWriter(dns, g)
149 _, sesErr := p.EnsureIdentity(ctx, "eu-west-1", domain, "", false)
150 dnsErr := d.EnsureRecord(ctx, dnsrec.Record{Name: "_dmarc." + domain, Type: "TXT", Values: []string{"v=DMARC1; p=none"}})
151 valid := (schema == 0 || schema == 1) && bad == ""
152 if valid {
153 if sesErr != nil || dnsErr != nil {
154 t.Fatalf("legacy supported claim: SES=%v DNS=%v", sesErr, dnsErr)
155 }
156 } else {
157 if sesErr == nil || dnsErr == nil {
158 t.Fatalf("invalid claim accepted: SES=%v DNS=%v", sesErr, dnsErr)
159 }
160 if len(mem.Identities) != 0 || len(dns.Records) != 0 {
161 t.Fatal("invalid claim reached backend")
162 }
163 }
164 })
165 }
166 }
167}
168
169// The same worker fixture checks actual HTTP delivery, not just its SQL receipt.
170func TestClaimMailPushCountsOutboundEffects(t *testing.T) {
171 cx := newClaimCase(t)
172 door := claimFixtures["worker StartMailEventPush"].door
173 door.setup(t, cx)
174 ref := door.apps[0]
175 restore := cx.foreign(t, ref)
176 w := cx.measure(t, door)
177 if problems := claimForeignProblems(ref, door.apps, w); len(problems) > 0 {
178 t.Errorf("foreign delivery: %v", problems)
179 }
180 restore()
181 w = cx.measure(t, door)
182 sends := 0
183 for _, e := range w.events {
184 if e.Surface == "http" && e.Method == http.MethodPost {
185 sends++
186 }
187 }
188 if sends != 2 {
189 t.Fatalf("open worker sent %d requests, want 2", sends)
190 }
191}
192
193type claimDeliveryTransport struct {
194 call func(*http.Request) (*http.Response, error)
195}
196
197func (tr claimDeliveryTransport) RoundTrip(r *http.Request) (*http.Response, error) {
198 return tr.call(r)
199}
200
201func TestClaimMailPushHoldsAdmissionThroughDelivery(t *testing.T) {
202 for _, ambiguous := range []bool{false, true} {
203 t.Run(fmt.Sprint(ambiguous), func(t *testing.T) {
204 cx := newClaimCase(t)
205 door := claimFixtures["worker StartMailEventPush"].door
206 door.setup(t, cx)
207 appID := cx.appIDs[door.apps[0]]
208 calls := 0
209 old := mailPushClient.Transport
210 mailPushClient.Transport = claimDeliveryTransport{call: func(r *http.Request) (*http.Response, error) {
211 calls++
212 if err := hostops.RequireAppMutation(r.Context(), cx.rel, "bab", "titogo", appID); err != nil {
213 t.Errorf("outbound call lacks permit: %v", err)
214 }
215 a, err := hostops.ReadCanonicalAdmission(r.Context(), cx.rel, "bab", "titogo")
216 if err != nil || a.Value.Op == nil || a.Value.Op.Intent["action"] != "mail-event-delivery" {
217 t.Errorf("delivery admission: %+v %v", a.Value, err)
218 }
219 _, err = cx.srv.cfg.ClaimAuthority.Control.BeginExclusive(r.Context(), "bab", "titogo", appID, hostops.AppOperation{ID: "competing", Kind: hostops.OpPurging, AppID: appID, Phase: "reserved"}, claimTestOwner)
220 if err == nil {
221 t.Error("purge entered while HTTP delivery was in flight")
222 }
223 if ambiguous {
224 return nil, errors.New("lost response")
225 }
226 return old.RoundTrip(r)
227 }}
228 t.Cleanup(func() { mailPushClient.Transport = old })
229 if err := cx.srv.mailPushPass(t.Context(), time.Now); err != nil {
230 t.Fatal(err)
231 }
232 a, err := hostops.ReadCanonicalAdmission(t.Context(), cx.rel, "bab", "titogo")
233 if err != nil {
234 t.Fatal(err)
235 }
236 if ambiguous {
237 if calls != 1 || a.Value.Op == nil || a.Value.State != "closed" {
238 t.Fatalf("ambiguous call released: calls=%d admission=%+v", calls, a.Value)
239 }
240 } else {
241 if calls != 2 || a.Value.Op != nil || a.Value.State != "open" {
242 t.Fatalf("joined delivery unsettled: calls=%d admission=%+v", calls, a.Value)
243 }
244 rows, err := cx.st.dueMailEventDeliveriesFor(t.Context(), "bab", "titogo", time.Now(), mailPushPerApp)
245 if err != nil || len(rows) != 0 {
246 t.Fatalf("delivery did not settle SQL: %v %v", rows, err)
247 }
248 }
249 })
250 }
251}
252
253type claimStoreInventoryFault struct {
254 release.Store
255 list []string
256 listErr, errorRead error
257}
258
259func (s claimStoreInventoryFault) List(context.Context, string) ([]string, error) {
260 return s.list, s.listErr
261}
262func (s claimStoreInventoryFault) Get(ctx context.Context, key string) ([]byte, string, error) {
263 if s.errorRead != nil {
264 return nil, "", s.errorRead
265 }
266 return s.Store.Get(ctx, key)
267}
268func TestClaimStoreInventoryCompleteListing(t *testing.T) {
269 raw := release.NewDirStore(t.TempDir())
270 key := objstore.KeyIn("bab", "titogo", "prod")
271 putMailJSON(t, raw, key, objstore.Record{Schema: 1, App: "titogo", Environment: "prod"})
272 sentinel := errors.New("partial listing")
273 for _, tc := range []struct {
274 name string
275 keys []string
276 listErr, readErr error
277 want bool
278 }{
279 {"valid duplicate", []string{key, key}, nil, nil, true},
280 {"partial", []string{key}, sentinel, nil, false},
281 {"missing listed", []string{key}, nil, os.ErrNotExist, false},
282 {"failed read", []string{key}, nil, sentinel, false},
283 {"foreign prefix", []string{"apps/zed/other/stores/prod.json"}, nil, nil, false},
284 {"stray", []string{key, "apps/bab/titogo/stores/notes.txt"}, nil, nil, false},
285 } {
286 t.Run(tc.name, func(t *testing.T) {
287 s := claimStoreInventoryFault{Store: raw, list: tc.keys, listErr: tc.listErr, errorRead: tc.readErr}
288 got, err := objstore.StrictStoreEnvironments(t.Context(), s, "bab", "titogo")
289 if tc.want {
290 if err != nil || len(got) != 1 || got[0] != "prod" {
291 t.Fatalf("%v %v", got, err)
292 }
293 } else if err == nil || got != nil {
294 t.Fatalf("incomplete inventory accepted: %v %v", got, err)
295 }
296 })
297 }
298}
299