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