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 Complete *ermite.CompleteJobRequest
36 Lines []Line
37 Steps map[int32]*ermite.StepUpdate
38}
39
40// Server is the fake. Create it with New; URL is where to point the runner.
41type Server struct {
42 ermiteconnect.UnimplementedRunnerServiceHandler
43
44 URL string
45 srv *httptest.Server
46
47 mu sync.Mutex
48 queue []*ermite.Job
49 jobs map[string]*jobState
50 assigned map[string]*ermite.Job // by job token
51 // Errors lists contract violations seen (a gap in AppendLog, a request over
52 // the limits, a call after the end): a test fails when it is not empty.
53 Errors []string
54 done chan string
55}
56
57type jobState struct {
58 log *LogStore
59 steps map[int32]*ermite.StepUpdate
60 complete *ermite.CompleteJobRequest
61}
62
63// Mount serves a RunnerService handler where ermite does: under /-/api, so
64// that a client that forgets the prefix fails here as it would against ermite.
65func Mount(path string, h http.Handler) http.Handler {
66 mux := http.NewServeMux()
67 mux.Handle("/-/api"+path, http.StripPrefix("/-/api", h))
68 return mux
69}
70
71// New starts the server; Close stops it.
72func New() *Server {
73 s := &Server{jobs: map[string]*jobState{}, assigned: map[string]*ermite.Job{}, done: make(chan string, 256)}
74 s.srv = httptest.NewServer(Mount(ermiteconnect.NewRunnerServiceHandler(s)))
75 s.URL = s.srv.URL
76 return s
77}
78
79// Close stops the server.
80func (s *Server) Close() { s.srv.Close() }
81
82// Queue adds jobs to hand out, in order.
83func (s *Server) Queue(jobs ...*ermite.Job) {
84 s.mu.Lock()
85 defer s.mu.Unlock()
86 s.queue = append(s.queue, jobs...)
87}
88
89// Violations returns the contract violations seen so far.
90func (s *Server) Violations() []string {
91 s.mu.Lock()
92 defer s.mu.Unlock()
93 return append([]string(nil), s.Errors...)
94}
95
96// Done receives the id of each job when its CompleteJob has been applied.
97func (s *Server) Done() <-chan string { return s.done }
98
99// Result returns what a job reported so far.
100func (s *Server) Result(jobID string) Result {
101 s.mu.Lock()
102 defer s.mu.Unlock()
103 st := s.jobs[jobID]
104 if st == nil {
105 return Result{}
106 }
107 steps := map[int32]*ermite.StepUpdate{}
108 for k, v := range st.steps {
109 steps[k] = v
110 }
111 return Result{Complete: st.complete, Lines: st.log.Lines(), Steps: steps}
112}
113
114func (s *Server) violation(msg string) { s.Errors = append(s.Errors, msg) } // called with s.mu held
115
116func unauth() error { return connect.NewError(connect.CodeUnauthenticated, io.EOF) }
117
118func (s *Server) job(h http.Header, id string) (*jobState, error) {
119 j := s.assigned[strings.TrimPrefix(h.Get("Authorization"), "Bearer ")]
120 if j == nil {
121 return nil, unauth()
122 }
123 if j.JobId != id {
124 return nil, connect.NewError(connect.CodePermissionDenied, io.EOF)
125 }
126 return s.jobs[id], nil
127}
128
129func ended(c ermite.CiConclusion) error {
130 e := connect.NewError(connect.CodeFailedPrecondition, io.EOF)
131 if d, err := connect.NewErrorDetail(&ermite.JobEnded{Conclusion: c}); err == nil {
132 e.AddDetail(d)
133 }
134 return e
135}
136
137// FetchJob hands out the queued jobs.
138func (s *Server) FetchJob(_ context.Context, r *connect.Request[ermite.FetchJobRequest]) (*connect.Response[ermite.FetchJobResponse], error) {
139 s.mu.Lock()
140 defer s.mu.Unlock()
141 if r.Header().Get("Authorization") != "Bearer "+RunnerToken {
142 return nil, unauth()
143 }
144 if len(s.queue) == 0 {
145 return connect.NewResponse(&ermite.FetchJobResponse{}), nil
146 }
147 j := s.queue[0]
148 s.queue = s.queue[1:]
149 s.assigned[j.JobToken] = j
150 s.jobs[j.JobId] = &jobState{log: &LogStore{OnViolation: s.violation}, steps: map[int32]*ermite.StepUpdate{}}
151 return connect.NewResponse(&ermite.FetchJobResponse{Job: j}), nil
152}
153
154// Heartbeat accepts heartbeats of running jobs.
155func (s *Server) Heartbeat(_ context.Context, r *connect.Request[ermite.HeartbeatRequest]) (*connect.Response[ermite.HeartbeatResponse], error) {
156 s.mu.Lock()
157 defer s.mu.Unlock()
158 st, err := s.job(r.Header(), r.Msg.JobId)
159 if err != nil {
160 return nil, err
161 }
162 if st.complete != nil {
163 return nil, ended(st.complete.Conclusion)
164 }
165 return connect.NewResponse(&ermite.HeartbeatResponse{}), nil
166}
167
168// UpdateJob stores step snapshots.
169func (s *Server) UpdateJob(_ context.Context, r *connect.Request[ermite.UpdateJobRequest]) (*connect.Response[ermite.UpdateJobResponse], 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 for _, u := range r.Msg.Steps {
180 if old := st.steps[u.Number]; old != nil && old.State == ermite.StepState_STEP_STATE_COMPLETED {
181 s.violation("UpdateJob: a completed step changed")
182 }
183 st.steps[u.Number] = u
184 }
185 return connect.NewResponse(&ermite.UpdateJobResponse{}), nil
186}
187
188// AppendLog applies the rules of runner.proto through a LogStore.
189func (s *Server) AppendLog(_ context.Context, r *connect.Request[ermite.AppendLogRequest]) (*connect.Response[ermite.AppendLogResponse], error) {
190 s.mu.Lock()
191 defer s.mu.Unlock()
192 st, err := s.job(r.Header(), r.Msg.JobId)
193 if err != nil {
194 return nil, err
195 }
196 if st.complete != nil {
197 s.violation("AppendLog after CompleteJob")
198 return nil, ended(st.complete.Conclusion)
199 }
200 next, err := st.log.Apply(r.Msg)
201 if err != nil {
202 return nil, err
203 }
204 return connect.NewResponse(&ermite.AppendLogResponse{NextOffset: next}), nil
205}
206
207// CompleteJob ends a job, idempotently.
208func (s *Server) CompleteJob(_ context.Context, r *connect.Request[ermite.CompleteJobRequest]) (*connect.Response[ermite.CompleteJobResponse], error) {
209 s.mu.Lock()
210 defer s.mu.Unlock()
211 st, err := s.job(r.Header(), r.Msg.JobId)
212 if err != nil {
213 return nil, err
214 }
215 if st.complete != nil {
216 return connect.NewResponse(&ermite.CompleteJobResponse{Conclusion: st.complete.Conclusion}), nil
217 }
218 st.complete = r.Msg
219 for _, u := range r.Msg.Steps {
220 st.steps[u.Number] = u
221 }
222 s.done <- r.Msg.JobId
223 return connect.NewResponse(&ermite.CompleteJobResponse{Conclusion: r.Msg.Conclusion, Applied: true}), nil
224}
225
226// WatchJob keeps the stream open until the job ends or the client leaves.
227func (s *Server) WatchJob(ctx context.Context, r *connect.Request[ermite.WatchJobRequest], _ *connect.ServerStream[ermite.WatchJobResponse]) error {
228 s.mu.Lock()
229 _, err := s.job(r.Header(), r.Msg.JobId)
230 s.mu.Unlock()
231 if err != nil {
232 return err
233 }
234 <-ctx.Done()
235 return ctx.Err()
236}