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.
171func (s *stream) next(t *testing.T, d time.Duration) *pb.Event {
172 t.Helper()
173 select {
174 case ev, ok := <-s.events:
175 if !ok {
176 t.Fatalf("the stream ended: %v", s.err)
177 }
178 return ev
179 case <-time.After(d):
180 t.Fatalf("no event within %v", d)
181 }
182 return nil
183}
184
185// readBlob reads a whole GetBlob stream and returns its messages.
186func readBlob(ctx context.Context, c *clients, req *pb.GetBlobRequest) ([]*pb.GetBlobResponse, error) {
187 st, err := c.git.GetBlob(ctx, connect.NewRequest(req))
188 if err != nil {
189 return nil, err
190 }
191 defer func() { _ = st.Close() }()
192 var msgs []*pb.GetBlobResponse
193 for st.Receive() {
194 msgs = append(msgs, st.Msg())
195 }
196 if err := st.Err(); err != nil && !errors.Is(err, io.EOF) {
197 return msgs, err
198 }
199 return msgs, nil
200}