lvmh/daemon/docker_test.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 TestWorkerBindsHostModelsJSON(t *testing.T) {
	useFakeGit(t, fakeGitModeOK)
	f := newFakeDocker()
	sp, _ := newTestSpawner(t, f)
	agent := t.TempDir()
	if err := os.WriteFile(filepath.Join(agent, "models.json"), []byte(`{}`), 0o644); err != nil {
		t.Fatal(err)
	}
	t.Setenv(envHostPiAgentDir, agent)

	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
	want := agent + "/models.json:/root/.pi/agent/models.json:ro"
	ok := false
	for _, b := range binds {
		if b == want {
			ok = true
		}
	}
	if !ok {
		t.Fatalf("models.json bind missing from %v", binds)
	}
}

func TestWorkerBindsSpawnPiMesh(t *testing.T) {
	useFakeGit(t, fakeGitModeOK)
	f := newFakeDocker()
	sp, _ := newTestSpawner(t, f)
	runtimeDir := t.TempDir()
	if err := os.MkdirAll(filepath.Join(runtimeDir, "spawn-pi"), 0o755); err != nil {
		t.Fatal(err)
	}
	t.Setenv(envHostPiRuntimeDir, runtimeDir)

	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
	want := filepath.Join(runtimeDir, "spawn-pi") + ":/root/.pi/spawn-pi"
	ok := false
	for _, b := range binds {
		if b == want {
			ok = true
		}
	}
	if !ok {
		t.Fatalf("spawn-pi mesh bind missing from %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)
		}
	}
}