1// SPDX-License-Identifier: AGPL-3.0-or-later
2
3package app
4
5import (
6 "context"
7 "errors"
8 "fmt"
9 "io"
10 "net/http"
11 "os"
12 "path/filepath"
13 "sync/atomic"
14 "testing"
15 "time"
16
17 "connectrpc.com/connect"
18 "github.com/stretchr/testify/require"
19
20 pb "bnema.dev/ermite-api/gen/ermite"
21 "bnema.dev/ermite-api/gen/ermite/ermiteconnect"
22 "bnema.dev/ermite/internal/domain"
23 "bnema.dev/ermite/internal/ports"
24)
25
26// clients are the generated Connect clients of one caller.
27type clients struct {
28 repo ermiteconnect.RepoServiceClient
29 git ermiteconnect.GitServiceClient
30 token ermiteconnect.TokenServiceClient
31 event ermiteconnect.EventServiceClient
32 server ermiteconnect.ServerServiceClient
33 mr ermiteconnect.MergeRequestServiceClient
34
35 session ermiteconnect.SessionServiceClient
36 runner ermiteconnect.RunnerServiceClient
37 run ermiteconnect.RunServiceClient
38}
39
40// bearerTransport adds the Authorization header of a secret, if any.
41type bearerTransport struct {
42 secret string
43 // freshClient, when set, gives every request its own client address
44 // through X-Forwarded-For (the harness trusts the loopback proxy), so that
45 // many failed calls do not block the one client of the test.
46 freshClient *atomic.Uint32
47 next http.RoundTripper
48}
49
50func (b bearerTransport) RoundTrip(r *http.Request) (*http.Response, error) {
51 r = r.Clone(r.Context())
52 if b.secret != "" {
53 r.Header.Set("Authorization", "Bearer "+b.secret)
54 }
55 if b.freshClient != nil {
56 n := b.freshClient.Add(1)
57 r.Header.Set("X-Forwarded-For", fmt.Sprintf("198.51.100.%d", n%250+1))
58 }
59 return b.next.RoundTrip(r)
60}
61
62// newClients builds the clients of the API at base for a secret ("" is
63// anonymous), speaking the Connect protocol with the given options.
64func newClients(base, secret string, opts ...connect.ClientOption) *clients {
65 hc := &http.Client{Transport: bearerTransport{secret: secret, next: http.DefaultTransport}}
66 return newClientsOn(hc, base, opts...)
67}
68
69// newDisposableClients is newClients for a caller whose failures must not
70// count against the others: every call comes from a new client address.
71func newDisposableClients(base, secret string) *clients {
72 hc := &http.Client{Transport: bearerTransport{secret: secret, freshClient: new(atomic.Uint32), next: http.DefaultTransport}}
73 return newClientsOn(hc, base)
74}
75
76func newClientsOn(hc connect.HTTPClient, base string, opts ...connect.ClientOption) *clients {
77 return &clients{
78 repo: ermiteconnect.NewRepoServiceClient(hc, base, opts...),
79 git: ermiteconnect.NewGitServiceClient(hc, base, opts...),
80 token: ermiteconnect.NewTokenServiceClient(hc, base, opts...),
81 event: ermiteconnect.NewEventServiceClient(hc, base, opts...),
82 server: ermiteconnect.NewServerServiceClient(hc, base, opts...),
83 mr: ermiteconnect.NewMergeRequestServiceClient(hc, base, opts...),
84
85 session: ermiteconnect.NewSessionServiceClient(hc, base, opts...),
86 runner: ermiteconnect.NewRunnerServiceClient(hc, base, opts...),
87 run: ermiteconnect.NewRunServiceClient(hc, base, opts...),
88 }
89}
90
91// api returns the clients of a caller of the harness' server.
92func (h *harness) api(secret string, opts ...connect.ClientOption) *clients {
93 return newClients(h.url("/-/api"), secret, opts...)
94}
95
96// codeName names the Connect code of err, "ok" for nil.
97func codeName(err error) string {
98 if err == nil {
99 return "ok"
100 }
101 var ce *connect.Error
102 if errors.As(err, &ce) {
103 return ce.Code().String()
104 }
105 return "non-connect: " + err.Error()
106}
107
108// seed creates a repository through the use case as root and pushes one
109// commit per file to its main branch with git.
110func (h *harness) seed(name string, vis domain.Visibility, files ...string) {
111 h.t.Helper()
112 _, err := h.forge.Repos.Create(h.ctx, domain.Root(), ports.CreateRepoParams{Name: h.name(name), Visibility: vis})
113 require.NoError(h.t, err)
114 if len(files) == 0 {
115 return
116 }
117 wt := h.workTree("main", files...)
118 h.mustGit(wt, "push", "-q", h.urlAuth(adminToken, "/"+name+".git"), "main")
119}
120
121// writeFile commits a file with the given bytes to a work tree.
122func (h *harness) writeFile(wt, path string, data []byte) {
123 h.t.Helper()
124 full := filepath.Join(wt, path)
125 require.NoError(h.t, os.MkdirAll(filepath.Dir(full), 0o755))
126 require.NoError(h.t, os.WriteFile(full, data, 0o644))
127 h.mustGit(wt, "add", path)
128 h.mustGit(wt, "commit", "-q", "-m", "add "+path)
129}
130
131// stream is a running Watch: its events, and how it ended.
132type stream struct {
133 events chan *pb.Event
134 done chan struct{}
135 err error
136 cancel context.CancelFunc
137}
138
139// watch starts a Watch and collects its events. The stream is cancelled when
140// the test ends.
141func (h *harness) watch(c *clients, resume string, topics ...string) *stream {
142 h.t.Helper()
143 ctx, cancel := context.WithCancel(h.ctx)
144 h.t.Cleanup(cancel)
145 st, err := c.event.Watch(ctx, connect.NewRequest(&pb.WatchRequest{Topics: topics, ResumeAfter: resume}))
146 require.NoError(h.t, err)
147 s := &stream{events: make(chan *pb.Event, 256), done: make(chan struct{}), cancel: cancel}
148 go func() {
149 defer close(s.done)
150 defer close(s.events)
151 for st.Receive() {
152 s.events <- st.Msg().GetEvent()
153 }
154 s.err = st.Err()
155 _ = st.Close()
156 }()
157 return s
158}
159
160// next returns the next event or fails after d.
161func (s *stream) next(t *testing.T, d time.Duration) *pb.Event {
162 t.Helper()
163 select {
164 case ev, ok := <-s.events:
165 if !ok {
166 t.Fatalf("the stream ended: %v", s.err)
167 }
168 return ev
169 case <-time.After(d):
170 t.Fatalf("no event within %v", d)
171 }
172 return nil
173}
174
175// readBlob reads a whole GetBlob stream and returns its messages.
176func readBlob(ctx context.Context, c *clients, req *pb.GetBlobRequest) ([]*pb.GetBlobResponse, error) {
177 st, err := c.git.GetBlob(ctx, connect.NewRequest(req))
178 if err != nil {
179 return nil, err
180 }
181 defer func() { _ = st.Close() }()
182 var msgs []*pb.GetBlobResponse
183 for st.Receive() {
184 msgs = append(msgs, st.Msg())
185 }
186 if err := st.Err(); err != nil && !errors.Is(err, io.EOF) {
187 return msgs, err
188 }
189 return msgs, nil
190}