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