548 lines
16 KiB
Go
548 lines
16 KiB
Go
package main
|
|
|
|
// docker_test.go — spawner pipeline against a fake docker REST API and a
|
|
// fake git binary (PATH shim recording invocations).
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"fmt"
|
|
"io"
|
|
"net/http"
|
|
"net/http/httptest"
|
|
"os"
|
|
"path/filepath"
|
|
"strings"
|
|
"sync"
|
|
"testing"
|
|
"time"
|
|
)
|
|
|
|
// --- fake docker daemon (just enough REST for the spawner pipeline) ---
|
|
|
|
type dockerCall struct {
|
|
Method string
|
|
Path string
|
|
Query string
|
|
Body string
|
|
}
|
|
|
|
func (c dockerCall) String() string { return c.Method + " " + c.Path + "?" + c.Query }
|
|
|
|
// recordedCreate is the parsed POST /containers/create request body.
|
|
type recordedCreate struct {
|
|
Name string
|
|
Image string `json:"Image"`
|
|
Env []string `json:"Env"`
|
|
Labels map[string]string `json:"Labels"`
|
|
Cmd []string `json:"Cmd"`
|
|
WorkingDir string `json:"WorkingDir"`
|
|
HostConfig struct {
|
|
Binds []string `json:"Binds"`
|
|
NetworkMode string `json:"NetworkMode"`
|
|
Init *bool `json:"Init"`
|
|
} `json:"HostConfig"`
|
|
}
|
|
|
|
type fakeDocker struct {
|
|
mu sync.Mutex
|
|
|
|
calls []dockerCall
|
|
images int // legacy: >0 serves the default worker image
|
|
imageTags map[string]bool // when set, /images/json serves exactly these
|
|
volume map[string]bool // existing volumes
|
|
nextID int
|
|
create []recordedCreate
|
|
archives []int // sizes of accepted CopyToContainer tar streams
|
|
|
|
containers map[string]string // name-or-id → container id
|
|
running map[string]bool // container id → running
|
|
|
|
failBuild bool
|
|
failBuildHTTP bool
|
|
failCreate bool
|
|
failStart bool
|
|
failStop bool
|
|
failWait bool
|
|
failVolumeCreate bool
|
|
failVolumeDelete bool
|
|
failImages bool
|
|
failVolumeInspect bool
|
|
failArchive bool
|
|
archiveHang bool
|
|
waitHang bool
|
|
|
|
unknown []string
|
|
}
|
|
|
|
func newFakeDocker() *fakeDocker {
|
|
return &fakeDocker{images: 1, volume: map[string]bool{},
|
|
containers: map[string]string{}, running: map[string]bool{}}
|
|
}
|
|
|
|
// containerID resolves a name-or-id path token to the tracked container id.
|
|
func (f *fakeDocker) containerID(token string) (string, bool) {
|
|
f.mu.Lock()
|
|
defer f.mu.Unlock()
|
|
id, ok := f.containers[token]
|
|
return id, ok
|
|
}
|
|
|
|
// isRunning reports the running flag of a tracked container id.
|
|
func (f *fakeDocker) isRunning(id string) bool {
|
|
f.mu.Lock()
|
|
defer f.mu.Unlock()
|
|
return f.running[id]
|
|
}
|
|
|
|
func (f *fakeDocker) server(t *testing.T) *httptest.Server {
|
|
t.Helper()
|
|
ts := httptest.NewServer(f)
|
|
t.Cleanup(ts.Close)
|
|
return ts
|
|
}
|
|
|
|
// record appends a call; the /vX.Y version prefix negotiated by the SDK is
|
|
// stripped so assertions are version-agnostic.
|
|
// imageListJSON builds the GET /images/json entry list (joined with commas)
|
|
// honoring the reference filter. Legacy mode (imageTags nil): images>0 means
|
|
// exactly the default worker image is present.
|
|
func (f *fakeDocker) imageListJSON(filtersJSON string) string {
|
|
matches := func(ref string) bool {
|
|
if f.imageTags == nil {
|
|
return f.images > 0 && ref == imageRefWorker
|
|
}
|
|
return f.imageTags[ref]
|
|
}
|
|
var refs []string
|
|
if filtersJSON != "" {
|
|
var flt struct {
|
|
Reference map[string]bool `json:"reference"`
|
|
}
|
|
if err := json.Unmarshal([]byte(filtersJSON), &flt); err != nil {
|
|
return ""
|
|
}
|
|
for ref, want := range flt.Reference {
|
|
if want {
|
|
refs = append(refs, ref)
|
|
}
|
|
}
|
|
} else if f.imageTags != nil {
|
|
for ref := range f.imageTags {
|
|
refs = append(refs, ref)
|
|
}
|
|
}
|
|
entries := make([]string, 0, len(refs))
|
|
for _, ref := range refs {
|
|
if matches(ref) {
|
|
entries = append(entries, fmt.Sprintf(`{"Id":"sha256:abc","RepoTags":[%q]}`, ref))
|
|
}
|
|
}
|
|
return strings.Join(entries, ",")
|
|
}
|
|
|
|
func (f *fakeDocker) record(r *http.Request, body string) dockerCall {
|
|
path := r.URL.Path
|
|
if strings.HasPrefix(path, "/v") {
|
|
if i := strings.Index(path[1:], "/"); i >= 0 {
|
|
path = path[i+1:]
|
|
}
|
|
}
|
|
c := dockerCall{Method: r.Method, Path: path, Query: r.URL.RawQuery, Body: body}
|
|
f.mu.Lock()
|
|
f.calls = append(f.calls, c)
|
|
f.mu.Unlock()
|
|
return c
|
|
}
|
|
|
|
func (f *fakeDocker) hasCall(method, pathPrefix string) bool {
|
|
f.mu.Lock()
|
|
defer f.mu.Unlock()
|
|
for _, c := range f.calls {
|
|
if c.Method == method && strings.HasPrefix(c.Path, pathPrefix) {
|
|
return true
|
|
}
|
|
}
|
|
return false
|
|
}
|
|
|
|
func (f *fakeDocker) hasCallSuffix(method, suffix string) bool {
|
|
f.mu.Lock()
|
|
defer f.mu.Unlock()
|
|
for _, c := range f.calls {
|
|
if c.Method == method && strings.HasSuffix(c.Path, suffix) {
|
|
return true
|
|
}
|
|
}
|
|
return false
|
|
}
|
|
|
|
func (f *fakeDocker) countCalls(method, pathPrefix string) int {
|
|
f.mu.Lock()
|
|
defer f.mu.Unlock()
|
|
n := 0
|
|
for _, c := range f.calls {
|
|
if c.Method == method && strings.HasPrefix(c.Path, pathPrefix) {
|
|
n++
|
|
}
|
|
}
|
|
return n
|
|
}
|
|
|
|
func (f *fakeDocker) createsByName(prefix string) []recordedCreate {
|
|
f.mu.Lock()
|
|
defer f.mu.Unlock()
|
|
var out []recordedCreate
|
|
for _, c := range f.create {
|
|
if strings.HasPrefix(c.Name, prefix) {
|
|
out = append(out, c)
|
|
}
|
|
}
|
|
return out
|
|
}
|
|
|
|
func (f *fakeDocker) volumeExists(name string) bool {
|
|
f.mu.Lock()
|
|
defer f.mu.Unlock()
|
|
return f.volume[name]
|
|
}
|
|
|
|
func (f *fakeDocker) assertNoUnknown(t *testing.T) {
|
|
t.Helper()
|
|
f.mu.Lock()
|
|
defer f.mu.Unlock()
|
|
if len(f.unknown) > 0 {
|
|
t.Fatalf("fake docker got unhandled requests: %v", f.unknown)
|
|
}
|
|
}
|
|
|
|
func writeJSONNow(w http.ResponseWriter, status int, v string) {
|
|
w.Header().Set("Content-Type", "application/json")
|
|
w.WriteHeader(status)
|
|
_, _ = io.WriteString(w, v)
|
|
}
|
|
|
|
func (f *fakeDocker) ServeHTTP(w http.ResponseWriter, r *http.Request) {
|
|
body, _ := io.ReadAll(r.Body)
|
|
call := f.record(r, string(body))
|
|
switch {
|
|
case call.Path == "/_ping":
|
|
w.Header().Set("Api-Version", "1.44")
|
|
w.WriteHeader(http.StatusOK)
|
|
case call.Method == http.MethodGet && call.Path == "/images/json":
|
|
|
|
if f.failImages {
|
|
writeJSONNow(w, http.StatusInternalServerError, `{"message":"images endpoint broken"}`)
|
|
return
|
|
}
|
|
writeJSONNow(w, http.StatusOK, "["+f.imageListJSON(r.URL.Query().Get("filters"))+"]")
|
|
case call.Method == http.MethodPost && call.Path == "/build":
|
|
if f.failBuildHTTP {
|
|
writeJSONNow(w, http.StatusInternalServerError, `{"message":"build endpoint broken"}`)
|
|
return
|
|
}
|
|
if f.failBuild {
|
|
writeJSONNow(w, http.StatusOK, `{"errorDetail":{"message":"build exploded"},"error":"build exploded"}`+"\n")
|
|
return
|
|
}
|
|
writeJSONNow(w, http.StatusOK, `{"stream":"Successfully tagged lvmh-worker:latest"}`+"\n")
|
|
case call.Method == http.MethodPost && call.Path == "/volumes/create":
|
|
var req struct {
|
|
Name string `json:"Name"`
|
|
}
|
|
_ = json.Unmarshal(body, &req)
|
|
f.mu.Lock()
|
|
f.volume[req.Name] = true
|
|
f.mu.Unlock()
|
|
if f.failVolumeCreate {
|
|
writeJSONNow(w, http.StatusInternalServerError, `{"message":"volume create failed"}`)
|
|
return
|
|
}
|
|
writeJSONNow(w, http.StatusCreated, fmt.Sprintf(`{"Name":%q,"CreatedAt":"2024-01-01T00:00:00Z"}`, req.Name))
|
|
case call.Method == http.MethodGet && strings.HasPrefix(call.Path, "/volumes/"):
|
|
name := strings.TrimPrefix(call.Path, "/volumes/")
|
|
f.mu.Lock()
|
|
exists := f.volume[name]
|
|
f.mu.Unlock()
|
|
if f.failVolumeInspect {
|
|
writeJSONNow(w, http.StatusInternalServerError, `{"message":"transient docker error"}`)
|
|
return
|
|
}
|
|
if exists {
|
|
writeJSONNow(w, http.StatusOK, fmt.Sprintf(`{"Name":%q,"CreatedAt":"2024-01-01T00:00:00Z"}`, name))
|
|
return
|
|
}
|
|
writeJSONNow(w, http.StatusNotFound, `{"message":"no such volume"}`)
|
|
case call.Method == http.MethodDelete && strings.HasPrefix(call.Path, "/volumes/"):
|
|
if f.failVolumeDelete {
|
|
writeJSONNow(w, http.StatusInternalServerError, `{"message":"volume delete failed"}`)
|
|
return
|
|
}
|
|
name := strings.TrimPrefix(call.Path, "/volumes/")
|
|
f.mu.Lock()
|
|
delete(f.volume, name)
|
|
f.mu.Unlock()
|
|
w.WriteHeader(http.StatusNoContent)
|
|
case call.Method == http.MethodPost && call.Path == "/containers/create":
|
|
var rc recordedCreate
|
|
if err := json.Unmarshal(body, &rc); err != nil {
|
|
writeJSONNow(w, http.StatusBadRequest, `{"message":"bad create body"}`)
|
|
return
|
|
}
|
|
rc.Name = r.URL.Query().Get("name")
|
|
if f.failCreate {
|
|
writeJSONNow(w, http.StatusInternalServerError, `{"message":"create failed"}`)
|
|
return
|
|
}
|
|
f.mu.Lock()
|
|
f.nextID++
|
|
id := fmt.Sprintf("cid-%d", f.nextID)
|
|
f.create = append(f.create, rc)
|
|
f.containers[id] = id
|
|
if rc.Name != "" {
|
|
f.containers[rc.Name] = id
|
|
}
|
|
f.running[id] = false
|
|
f.mu.Unlock()
|
|
writeJSONNow(w, http.StatusCreated, fmt.Sprintf(`{"Id":%q,"Warnings":null}`, id))
|
|
case call.Method == http.MethodGet && strings.HasPrefix(call.Path, "/containers/") && strings.HasSuffix(call.Path, "/json"):
|
|
token := strings.TrimSuffix(strings.TrimPrefix(call.Path, "/containers/"), "/json")
|
|
f.mu.Lock()
|
|
id, ok := f.containers[token]
|
|
running := ok && f.running[id]
|
|
f.mu.Unlock()
|
|
if !ok {
|
|
writeJSONNow(w, http.StatusNotFound, fmt.Sprintf(`{"message":"No such container: %s"}`, token))
|
|
return
|
|
}
|
|
writeJSONNow(w, http.StatusOK, fmt.Sprintf(`{"Id":%q,"State":{"Running":%t}}`, id, running))
|
|
case call.Method == http.MethodPost && strings.HasSuffix(call.Path, "/start"):
|
|
if f.failStart {
|
|
writeJSONNow(w, http.StatusInternalServerError, `{"message":"start failed"}`)
|
|
return
|
|
}
|
|
token := strings.TrimSuffix(strings.TrimPrefix(call.Path, "/containers/"), "/start")
|
|
f.mu.Lock()
|
|
if id, ok := f.containers[token]; ok {
|
|
f.running[id] = true
|
|
}
|
|
f.mu.Unlock()
|
|
w.WriteHeader(http.StatusNoContent)
|
|
case call.Method == http.MethodPost && strings.HasSuffix(call.Path, "/stop"):
|
|
if f.failStop {
|
|
writeJSONNow(w, http.StatusInternalServerError, `{"message":"stop failed"}`)
|
|
return
|
|
}
|
|
token := strings.TrimSuffix(strings.TrimPrefix(call.Path, "/containers/"), "/stop")
|
|
f.mu.Lock()
|
|
if id, ok := f.containers[token]; ok {
|
|
f.running[id] = false
|
|
}
|
|
f.mu.Unlock()
|
|
w.WriteHeader(http.StatusNoContent)
|
|
case call.Method == http.MethodPost && strings.HasSuffix(call.Path, "/wait"):
|
|
if f.failWait {
|
|
writeJSONNow(w, http.StatusInternalServerError, `{"message":"wait failed"}`)
|
|
return
|
|
}
|
|
if f.waitHang {
|
|
<-r.Context().Done() // hold the connection until the client gives up
|
|
return
|
|
}
|
|
writeJSONNow(w, http.StatusOK, `{"StatusCode":0,"Error":null}`)
|
|
case call.Method == http.MethodPut && strings.HasSuffix(call.Path, "/archive"):
|
|
// CopyToContainer: accept the tar stream and record its size.
|
|
if f.archiveHang {
|
|
<-r.Context().Done()
|
|
return
|
|
}
|
|
if f.failArchive {
|
|
writeJSONNow(w, http.StatusInternalServerError, `{"message":"archive failed"}`)
|
|
return
|
|
}
|
|
f.mu.Lock()
|
|
f.archives = append(f.archives, len(body))
|
|
f.mu.Unlock()
|
|
w.WriteHeader(http.StatusOK)
|
|
case call.Method == http.MethodDelete && strings.HasPrefix(call.Path, "/containers/"):
|
|
token := strings.TrimPrefix(call.Path, "/containers/")
|
|
f.mu.Lock()
|
|
if id, ok := f.containers[token]; ok {
|
|
delete(f.containers, token)
|
|
delete(f.containers, id)
|
|
delete(f.running, id)
|
|
}
|
|
f.mu.Unlock()
|
|
w.WriteHeader(http.StatusNoContent)
|
|
default:
|
|
f.mu.Lock()
|
|
f.unknown = append(f.unknown, call.String())
|
|
f.mu.Unlock()
|
|
writeJSONNow(w, http.StatusNotFound, `{"message":"not implemented: `+call.String()+`"}`)
|
|
}
|
|
}
|
|
|
|
// --- fake git (PATH shim): logs invocations, optionally fails ---
|
|
|
|
const (
|
|
fakeGitModeOK string = "ok"
|
|
fakeGitModeFail string = "fail"
|
|
fakeGitModeNoisy string = "noisy"
|
|
fakeGitModeSilent string = "silent"
|
|
fakeGitModeHang string = "hang"
|
|
)
|
|
|
|
// useFakeGit puts a shim `git` first on PATH. Modes: ok (exit 0), fail
|
|
// (stderr + exit 1), noisy (600-byte stderr + exit 1, for truncation tests),
|
|
// hang (sleeps; only ctx cancellation can end it).
|
|
func useFakeGit(t *testing.T, mode string) string {
|
|
t.Helper()
|
|
dir := t.TempDir()
|
|
logPath := filepath.Join(t.TempDir(), "git-calls.log")
|
|
script := "#!/bin/sh\n" +
|
|
"printf '%s\\n' \"$*\" >> \"" + logPath + "\"\n" +
|
|
"case \"$FAKE_GIT_MODE\" in\n" +
|
|
" fail) echo 'fatal: repository not found'; exit 1;;\n" +
|
|
" noisy) printf '%s\\n' \"" + strings.Repeat("x", 600) + "\"; exit 1;;\n" +
|
|
" silent) exit 1;;\n" +
|
|
" hang) exec sleep 30;;\n" +
|
|
"esac\n" +
|
|
"# branch probe: emit the configured HEAD name\n" +
|
|
"if [ \"$1\" = '-C' ] && [ \"$3\" = rev-parse ]; then printf '%s' \"$FAKE_GIT_HEAD\"; exit 0; fi\n" +
|
|
"# clone: create a real worktree so seeding (tarDir) has files to copy\n" +
|
|
"# (clone may be $1 or $3, depending on leading -c auth args)\n" +
|
|
"case \" $* \" in *\" clone \"*) d=$(eval \"echo \\${$#}\"); mkdir -p \"$d\" && printf 'fake-repo\\n' > \"$d/README.md\";; esac\n" +
|
|
"exit 0\n"
|
|
if err := os.WriteFile(filepath.Join(dir, "git"), []byte(script), 0o755); err != nil {
|
|
t.Fatalf("write fake git: %v", err)
|
|
}
|
|
t.Setenv("PATH", dir+":"+os.Getenv("PATH"))
|
|
t.Setenv("FAKE_GIT_MODE", mode)
|
|
return logPath
|
|
}
|
|
|
|
func readGitLog(t *testing.T, path string) []string {
|
|
t.Helper()
|
|
b, err := os.ReadFile(path)
|
|
if err != nil {
|
|
if os.IsNotExist(err) {
|
|
return nil
|
|
}
|
|
t.Fatalf("read git log: %v", err)
|
|
}
|
|
var out []string
|
|
for _, line := range strings.Split(strings.TrimSpace(string(b)), "\n") {
|
|
if line != "" {
|
|
out = append(out, line)
|
|
}
|
|
}
|
|
return out
|
|
}
|
|
|
|
// --- spawner construction wired at the fake daemon ---
|
|
|
|
func newTestSpawner(t *testing.T, f *fakeDocker) (*Spawner, *Store) {
|
|
t.Helper()
|
|
ts := f.server(t)
|
|
t.Setenv("DOCKER_HOST", "tcp://"+ts.Listener.Addr().String())
|
|
|
|
buildCtx := t.TempDir()
|
|
dockerfile := filepath.Join(buildCtx, "docker", "worker.Dockerfile")
|
|
if err := os.MkdirAll(filepath.Dir(dockerfile), 0o755); err != nil {
|
|
t.Fatalf("mkdir dockerfile dir: %v", err)
|
|
}
|
|
if err := os.WriteFile(dockerfile, []byte("FROM scratch\n"), 0o644); err != nil {
|
|
t.Fatalf("write dockerfile: %v", err)
|
|
}
|
|
if err := os.WriteFile(filepath.Join(buildCtx, "app.txt"), []byte("app"), 0o644); err != nil {
|
|
t.Fatalf("write app.txt: %v", err)
|
|
}
|
|
t.Setenv(envWorkerDockerfile, dockerfile)
|
|
t.Setenv(envWorkerContext, buildCtx)
|
|
t.Setenv(envRepoDir, t.TempDir())
|
|
t.Setenv(envProviderAPIKey, "key-123")
|
|
|
|
daemonToken = testToken
|
|
store := openTestStore(t)
|
|
hub := NewHub(store)
|
|
sp, err := NewSpawner(context.Background(), store, hub, "https://gitlab.example/")
|
|
if err != nil {
|
|
t.Fatalf("NewSpawner: %v", err)
|
|
}
|
|
return sp, store
|
|
}
|
|
|
|
// waitJobState polls JobsSnapshot until the job for sessionID reaches want.
|
|
func waitJobState(t *testing.T, sp *Spawner, sessionID, want string) SpawnJob {
|
|
t.Helper()
|
|
var job SpawnJob
|
|
waitFor(t, 5*time.Second, func() bool {
|
|
for _, j := range sp.JobsSnapshot() {
|
|
if j.SessionID == sessionID {
|
|
job = j
|
|
return j.State == want
|
|
}
|
|
}
|
|
return false
|
|
})
|
|
return job
|
|
}
|
|
|
|
func TestSpawnerSecretsBinds(t *testing.T) {
|
|
useFakeGit(t, fakeGitModeOK)
|
|
f := newFakeDocker()
|
|
sp, _ := newTestSpawner(t, f)
|
|
sec := t.TempDir()
|
|
t.Setenv(envSecretsDir, sec)
|
|
|
|
res, err := sp.Start(context.Background(), "group/project", "", "", false)
|
|
if err != nil {
|
|
t.Fatalf("Start: %v", err)
|
|
}
|
|
waitJobState(t, sp, res.SessionID, stateRunning)
|
|
creates := f.createsByName("lvmh-agent-")
|
|
if len(creates) != 1 {
|
|
t.Fatalf("creates = %d", len(creates))
|
|
}
|
|
binds := creates[0].HostConfig.Binds
|
|
ssh, gc := false, false
|
|
for _, b := range binds {
|
|
if b == sec+":/root/.ssh:ro" {
|
|
ssh = true
|
|
}
|
|
if b == sec+"/gitconfig:/root/.gitconfig:ro" {
|
|
gc = true
|
|
}
|
|
}
|
|
if !ssh || !gc {
|
|
t.Fatalf("secrets binds missing: %v", binds)
|
|
}
|
|
}
|
|
|
|
func TestWorkerBindsCaches(t *testing.T) {
|
|
useFakeGit(t, fakeGitModeOK)
|
|
f := newFakeDocker()
|
|
sp, _ := newTestSpawner(t, f)
|
|
t.Setenv(envPlaywrightCacheDir, "/host/pw")
|
|
t.Setenv(envCloakCacheDir, "/host/cb")
|
|
|
|
res, err := sp.Start(context.Background(), "group/project", "", "", false)
|
|
if err != nil {
|
|
t.Fatalf("Start: %v", err)
|
|
}
|
|
waitJobState(t, sp, res.SessionID, stateRunning)
|
|
creates := f.createsByName("lvmh-agent-")
|
|
binds := creates[0].HostConfig.Binds
|
|
for _, want := range []string{"/host/pw:/pw-browsers:ro", "/host/cb:/cloakbrowser-cache:ro"} {
|
|
ok := false
|
|
for _, b := range binds {
|
|
if b == want {
|
|
ok = true
|
|
}
|
|
}
|
|
if !ok {
|
|
t.Fatalf("bind %s missing from %v", want, binds)
|
|
}
|
|
}
|
|
}
|