1// SPDX-License-Identifier: MIT
2
3// Package hosttest is an in-memory RunnerService for tests that run the real
4// host loop against a real sandbox. It enforces the rules of runner.proto that
5// matter to a client: tokens per kind and per job, the offsets of AppendLog,
6// the limits of a request, and the end of a job after CompleteJob.
7package hosttest
8
9import (
10 "context"
11 "io"
12 "net/http"
13 "net/http/httptest"
14 "strings"
15 "sync"
16
17 "connectrpc.com/connect"
18
19 ermite "bnema.dev/ermite-api/gen/ermite"
20 "bnema.dev/ermite-api/gen/ermite/ermiteconnect"
21)
22
23// RunnerToken is the runner token the server accepts.
24const RunnerToken = "ermiterun_test"
25
26// Line is one stored log line.
27type Line struct {
28 Step int32
29 Text string
30 Continued bool
31}
32
33// Result is what a job reported.
34type Result struct {
35 // Annotations are those stored, by step (replace-per-step rule), and
36 // Summaries the Markdown stored by step.
37 Annotations map[int32][]*ermite.Annotation
38 Summaries map[int32]string
39 Complete *ermite.CompleteJobRequest
40 Lines []Line
41 Steps map[int32]*ermite.StepUpdate
42}
43
44// Server is the fake. Create it with New; URL is where to point the runner.
45type Server struct {
46 ermiteconnect.UnimplementedRunnerServiceHandler
47
48 URL string
49 srv *httptest.Server
50
51 mu sync.Mutex
52 queue []*ermite.Job
53 jobs map[string]*jobState
54 assigned map[string]*ermite.Job // by job token
55 // Errors lists contract violations seen (a gap in AppendLog, a request over
56 // the limits, a call after the end): a test fails when it is not empty.
57 Errors []string
58 done chan string
59}
60
61type jobState struct {
62 annots map[int32][]*ermite.Annotation
63 sums map[int32]string
64 log *LogStore
65 steps map[int32]*ermite.StepUpdate
66 complete *ermite.CompleteJobRequest
67}
68
69// Mount serves a RunnerService handler where ermite does: under /-/api, so
70// that a client that forgets the prefix fails here as it would against ermite.
71func Mount(path string, h http.Handler) http.Handler {
72 mux := http.NewServeMux()
73 mux.Handle("/-/api"+path, http.StripPrefix("/-/api", h))
74 return mux
75}
76
77// New starts the server; Close stops it.
78func New() *Server {
79 s := &Server{jobs: map[string]*jobState{}, assigned: map[string]*ermite.Job{}, done: make(chan string, 256)}
80 s.srv = httptest.NewServer(Mount(ermiteconnect.NewRunnerServiceHandler(s)))
81 s.URL = s.srv.URL
82 return s
83}
84
85// Close stops the server.
86func (s *Server) Close() { s.srv.Close() }
87
88// Queue adds jobs to hand out, in order.
89func (s *Server) Queue(jobs ...*ermite.Job) {
90 s.mu.Lock()
91 defer s.mu.Unlock()
92 s.queue = append(s.queue, jobs...)
93}
94
95// Violations returns the contract violations seen so far.
96func (s *Server) Violations() []string {
97 s.mu.Lock()
98 defer s.mu.Unlock()
99 return append([]string(nil), s.Errors...)
100}
101
102// Done receives the id of each job when its CompleteJob has been applied.
103func (s *Server) Done() <-chan string { return s.done }
104
105// Result returns what a job reported so far.
106func (s *Server) Result(jobID string) Result {
107 s.mu.Lock()
108 defer s.mu.Unlock()
109 st := s.jobs[jobID]
110 if st == nil {
111 return Result{}
112 }
113 steps := map[int32]*ermite.StepUpdate{}
114 for k, v := range st.steps {
115 steps[k] = v
116 }
117 anns := map[int32][]*ermite.Annotation{}
118 for k, v := range st.annots {
119 anns[k] = v
120 }
121 sums := map[int32]string{}
122 for k, v := range st.sums {
123 sums[k] = v
124 }
125 return Result{Complete: st.complete, Lines: st.log.Lines(), Steps: steps, Annotations: anns, Summaries: sums}
126}
127
128func (s *Server) violation(msg string) { s.Errors = append(s.Errors, msg) } // called with s.mu held
129
130func unauth() error { return connect.NewError(connect.CodeUnauthenticated, io.EOF) }
131
132func (s *Server) job(h http.Header, id string) (*jobState, error) {
133 j := s.assigned[strings.TrimPrefix(h.Get("Authorization"), "Bearer ")]
134 if j == nil {
135 return nil, unauth()
136 }
137 if j.JobId != id {
138 return nil, connect.NewError(connect.CodePermissionDenied, io.EOF)
139 }
140 return s.jobs[id], nil
141}
142
143func ended(c ermite.CiConclusion) error {
144 e := connect.NewError(connect.CodeFailedPrecondition, io.EOF)
145 if d, err := connect.NewErrorDetail(&ermite.JobEnded{Conclusion: c}); err == nil {
146 e.AddDetail(d)
147 }
148 return e
149}
150
151// FetchJob hands out the queued jobs.
152func (s *Server) FetchJob(_ context.Context, r *connect.Request[ermite.FetchJobRequest]) (*connect.Response[ermite.FetchJobResponse], error) {
153 s.mu.Lock()
154 defer s.mu.Unlock()
155 if r.Header().Get("Authorization") != "Bearer "+RunnerToken {
156 return nil, unauth()
157 }
158 if len(s.queue) == 0 {
159 return connect.NewResponse(&ermite.FetchJobResponse{}), nil
160 }
161 j := s.queue[0]
162 s.queue = s.queue[1:]
163 s.assigned[j.JobToken] = j
164 s.jobs[j.JobId] = &jobState{log: &LogStore{OnViolation: s.violation}, steps: map[int32]*ermite.StepUpdate{}, annots: map[int32][]*ermite.Annotation{}, sums: map[int32]string{}}
165 return connect.NewResponse(&ermite.FetchJobResponse{Job: j}), nil
166}
167
168// Heartbeat accepts heartbeats of running jobs.
169func (s *Server) Heartbeat(_ context.Context, r *connect.Request[ermite.HeartbeatRequest]) (*connect.Response[ermite.HeartbeatResponse], error) {
170 s.mu.Lock()
171 defer s.mu.Unlock()
172 st, err := s.job(r.Header(), r.Msg.JobId)
173 if err != nil {
174 return nil, err
175 }
176 if st.complete != nil {
177 return nil, ended(st.complete.Conclusion)
178 }
179 return connect.NewResponse(&ermite.HeartbeatResponse{}), nil
180}
181
182// UpdateJob stores step snapshots.
183func (s *Server) UpdateJob(_ context.Context, r *connect.Request[ermite.UpdateJobRequest]) (*connect.Response[ermite.UpdateJobResponse], error) {
184 s.mu.Lock()
185 defer s.mu.Unlock()
186 st, err := s.job(r.Header(), r.Msg.JobId)
187 if err != nil {
188 return nil, err
189 }
190 if st.complete != nil {
191 return nil, ended(st.complete.Conclusion)
192 }
193 for _, u := range r.Msg.Steps {
194 if old := st.steps[u.Number]; old != nil && old.State == ermite.StepState_STEP_STATE_COMPLETED {
195 s.violation("UpdateJob: a completed step changed")
196 }
197 st.steps[u.Number] = u
198 }
199 replace(st.annots, r.Msg.Annotations)
200 return connect.NewResponse(&ermite.UpdateJobResponse{}), nil
201}
202
203// AppendLog applies the rules of runner.proto through a LogStore.
204func (s *Server) AppendLog(_ context.Context, r *connect.Request[ermite.AppendLogRequest]) (*connect.Response[ermite.AppendLogResponse], error) {
205 s.mu.Lock()
206 defer s.mu.Unlock()
207 st, err := s.job(r.Header(), r.Msg.JobId)
208 if err != nil {
209 return nil, err
210 }
211 if st.complete != nil {
212 s.violation("AppendLog after CompleteJob")
213 return nil, ended(st.complete.Conclusion)
214 }
215 next, err := st.log.Apply(r.Msg)
216 if err != nil {
217 return nil, err
218 }
219 return connect.NewResponse(&ermite.AppendLogResponse{NextOffset: next}), nil
220}
221
222// CompleteJob ends a job, idempotently.
223func (s *Server) CompleteJob(_ context.Context, r *connect.Request[ermite.CompleteJobRequest]) (*connect.Response[ermite.CompleteJobResponse], error) {
224 s.mu.Lock()
225 defer s.mu.Unlock()
226 st, err := s.job(r.Header(), r.Msg.JobId)
227 if err != nil {
228 return nil, err
229 }
230 if st.complete != nil {
231 return connect.NewResponse(&ermite.CompleteJobResponse{Conclusion: st.complete.Conclusion}), nil
232 }
233 st.complete = r.Msg
234 for _, u := range r.Msg.Steps {
235 st.steps[u.Number] = u
236 }
237 replace(st.annots, r.Msg.Annotations)
238 s.done <- r.Msg.JobId
239 return connect.NewResponse(&ermite.CompleteJobResponse{Conclusion: r.Msg.Conclusion, Applied: true}), nil
240}
241
242// WatchJob keeps the stream open until the job ends or the client leaves.
243func (s *Server) WatchJob(ctx context.Context, r *connect.Request[ermite.WatchJobRequest], _ *connect.ServerStream[ermite.WatchJobResponse]) error {
244 s.mu.Lock()
245 _, err := s.job(r.Header(), r.Msg.JobId)
246 s.mu.Unlock()
247 if err != nil {
248 return err
249 }
250 <-ctx.Done()
251 return ctx.Err()
252}
253
254// replace applies the rule of runner.proto: the annotations of each step the
255// call names replace those stored.
256func replace(stored map[int32][]*ermite.Annotation, anns []*ermite.Annotation) {
257 named := map[int32]bool{}
258 for _, a := range anns {
259 if !named[a.StepNumber] {
260 named[a.StepNumber] = true
261 stored[a.StepNumber] = nil
262 }
263 stored[a.StepNumber] = append(stored[a.StepNumber], a)
264 }
265}
266
267// UploadSummary stores the summary of a step.
268func (s *Server) UploadSummary(_ context.Context, r *connect.Request[ermite.UploadSummaryRequest]) (*connect.Response[ermite.UploadSummaryResponse], error) {
269 s.mu.Lock()
270 defer s.mu.Unlock()
271 st, err := s.job(r.Header(), r.Msg.JobId)
272 if err != nil {
273 return nil, err
274 }
275 if st.complete != nil {
276 return nil, ended(st.complete.Conclusion)
277 }
278 st.sums[r.Msg.StepNumber] = r.Msg.Markdown
279 return connect.NewResponse(&ermite.UploadSummaryResponse{}), nil
280}