mesh-catalog/modules/mesh-build-go/cmd/mesh-build-go/integration_level_test.go

package main

import (
	"context"
	"encoding/json"
	"net/url"
	"os"
	"path/filepath"
	"strings"
	"sync"
	"testing"
	"time"

	"git.novox.be/novox/mesh-sdk/go/delivery"
	"git.novox.be/novox/mesh-sdk/go/forgekey"
)

// Issue 1486: the integration level, with the store and the bus faked at the process boundary: the commands the level
// runs are recorded, and a clock the commands advance stands in for the time they would take.

type integFake struct {
	mu       sync.Mutex
	calls    []cmdRecord
	clock    time.Time
	postgres string // what `postgres --version` prints
	captured string // the go test -json output of the run
	// advance is how long a phase takes on the fake clock: compile, start (of the store), test.
	advance map[string]time.Duration
	busURL  string
	busDir  string
	busDown bool
	runDirs []string // run directories seen when the store was started, to be gone at the end
	panicOn string   // a phase that panics
}

func newIntegFake(t *testing.T, captured string) *integFake {
	t.Helper()
	raw, err := os.ReadFile(filepath.Join("testdata", "captured", captured+".json"))
	if err != nil {
		t.Fatal(err)
	}
	return &integFake{clock: time.Date(2026, 10, 20, 12, 0, 0, 0, time.UTC), postgres: "postgres (PostgreSQL) 18.6",
		captured: string(raw), advance: map[string]time.Duration{}, busURL: "nats://127.0.0.1:45222"}
}

func (f *integFake) now() time.Time { f.mu.Lock(); defer f.mu.Unlock(); return f.clock }

func (f *integFake) tick(phase string) {
	f.mu.Lock()
	f.clock = f.clock.Add(f.advance[phase])
	f.mu.Unlock()
	if f.panicOn == phase {
		panic("the " + phase + " phase panicked")
	}
}

func (f *integFake) run(ctx context.Context, env []string, name string, args ...string) forgekey.Ran {
	line := name + " " + strings.Join(args, " ")
	f.mu.Lock()
	f.calls = append(f.calls, cmdRecord{env: env, line: line})
	f.mu.Unlock()
	switch {
	case name == "postgres":
		return forgekey.Ran{Stdout: f.postgres + "\n"}
	case name == "git" && len(args) > 2 && args[2] == "ls-files":
		return forgekey.Ran{Stdout: "go.mod\n"}
	case name == "go" && strings.Contains(line, "-run ^$"):
		f.tick("compile")
	case name == "go":
		f.tick("test")
		status := 0
		if strings.Contains(f.captured, `"Action":"fail"`) {
			status = 1
		}
		return forgekey.Ran{Stdout: f.captured, Status: status}
	case name == "pg_ctl" && strings.HasSuffix(line, " start"):
		f.tick("start")
		for i, a := range args {
			if a == "-D" {
				f.runDirs = append(f.runDirs, filepath.Dir(filepath.Dir(args[i+1])))
			}
		}
	}
	return forgekey.Ran{}
}

func (f *integFake) startBus(_ context.Context, dir string) (string, func(), error) {
	f.mu.Lock()
	f.busDir = dir
	f.mu.Unlock()
	return f.busURL, func() { f.mu.Lock(); f.busDown = true; f.mu.Unlock() }, nil
}

func (f *integFake) lines() string {
	f.mu.Lock()
	defer f.mu.Unlock()
	out := []string{}
	for _, c := range f.calls {
		out = append(out, c.line)
	}
	return strings.Join(out, "\n")
}

func integModule(t *testing.T, f *integFake) (*Module, *keptStates) {
	t.Helper()
	states := newStates()
	m := moduleWith(states, "go1.27.1")
	m.Home, m.Work, m.Cache = t.TempDir(), t.TempDir(), t.TempDir()
	m.Run, m.Now, m.StartBus = f.run, f.now, f.startBus
	// These tests judge the level as it runs in its shut-off process (level.go, 1309-S5): the store, the bus and the
	// tests side by side. That the step starts that one process shut off is isolate_test.go's.
	m.Inside = true
	if err := os.MkdirAll(filepath.Join(m.Work, "ws1", "tree"), 0o755); err != nil {
		t.Fatal(err)
	}
	return m, states
}

func runLevelTool(t *testing.T, m *Module, states *keptStates, level string) (TestResult, delivery.Step) {
	t.Helper()
	for _, tool := range m.Tools(m.StepTools()) {
		if tool.Name != "build_go_test" {
			continue
		}
		if _, err := tool.Run(map[string]any{"workspace": "ws1", "level": level}); err != nil {
			t.Fatal(err)
		}
	}
	st := states.waitEnd(t)
	var res TestResult
	if st.State == delivery.StepPassed {
		if err := json.Unmarshal(st.Result, &res); err != nil {
			t.Fatal(err)
		}
	}
	return res, st
}

// 1486/AC2: the store's start carries listen_addresses=” and a socket directory under the run; the tests are given
// its address and the bus's, and the run's directory is in the workspace's.
func TestTheStoreListensOnASocketUnderTheRunAndNowhereElse(t *testing.T) {
	f := newIntegFake(t, "slow")
	m, states := integModule(t, f)
	res, st := runLevelTool(t, m, states, "integration")
	if st.State != delivery.StepPassed || !res.Passed || res.Level != "integration" {
		t.Fatalf("the step ended %s (%s): %+v", st.State, st.Reason, res)
	}
	var start, initdb string
	for _, c := range f.calls {
		if strings.HasPrefix(c.line, "pg_ctl ") && strings.HasSuffix(c.line, " start") {
			start = c.line
		}
		if strings.HasPrefix(c.line, "initdb ") {
			initdb = c.line
		}
	}
	run := f.runDirs[0]
	if !strings.HasPrefix(run, filepath.Join(m.Work, "ws1", "run-")) {
		t.Errorf("the run directory is %s, not under the workspace's", run)
	}
	_, afterK, _ := strings.Cut(start, "-k ")
	socket, _, _ := strings.Cut(afterK, " ")
	if len(socket)+len(socketName) >= 100 {
		t.Errorf("the socket %q is too long for postgres", socket)
	}
	for _, want := range []string{"listen_addresses=''", "-k " + socket, "fsync=off", "-w"} {
		if !strings.Contains(start, want) {
			t.Errorf("the store's start %q lacks %q", start, want)
		}
	}
	if strings.Contains(start, "listen_addresses='*'") || strings.Contains(start, "-h ") || strings.Contains(start, "port=") {
		t.Errorf("the store listens on TCP: %s", start)
	}
	if !strings.Contains(initdb, "-D "+filepath.Join(run, "pg", "data")) || !strings.Contains(initdb, "--auth=trust") {
		t.Errorf("initdb = %q", initdb)
	}
	var test cmdRecord
	for _, c := range f.calls {
		if strings.HasPrefix(c.line, "go ") && strings.Contains(c.line, "-json") {
			test = c
		}
	}
	if !strings.Contains(test.line, "-tags integration,isolation") { // the bubblewrap proofs run too (the checker's agreement)
		t.Errorf("the run does not ask for the tag: %s", test.line)
	}
	env := map[string]string{}
	for _, w := range test.env {
		k, v, _ := strings.Cut(w, "=")
		env[k] = v
	}
	u, err := url.Parse(env["MESH_TEST_POSTGRES"])
	if err != nil || u.Scheme != "postgres" || u.Query().Get("host") != socket || u.Host != "" {
		t.Errorf("MESH_TEST_POSTGRES = %q", env["MESH_TEST_POSTGRES"])
	}
	if strings.Count(strings.TrimPrefix(env["MESH_TEST_POSTGRES"], "postgres://"), "/") != 1 {
		t.Errorf("MESH_TEST_POSTGRES = %q has a slash beyond the one before the database: the controller's tests cut it at the last", env["MESH_TEST_POSTGRES"])
	}
	if env["MESH_TEST_NATS"] != "nats://127.0.0.1:45222" || f.busDir != filepath.Join(run, "nats") {
		t.Errorf("MESH_TEST_NATS = %q, bus dir %q", env["MESH_TEST_NATS"], f.busDir)
	}
	if res.Store != "postgres 18" || res.Bus != "nats-server "+m.Recipe.NatsServer {
		t.Errorf("the result names %q and %q", res.Store, res.Bus)
	}
}

// 1486/AC3: a postgres of major 17 refuses the integration level, naming 17 and 18; the unit level still runs.
func TestAnotherPostgresMajorRefusesTheIntegrationLevelOnly(t *testing.T) {
	f := newIntegFake(t, "slow")
	f.postgres = "postgres (PostgreSQL) 17.2"
	m, states := integModule(t, f)
	_, st := runLevelTool(t, m, states, "integration")
	if st.State != delivery.StepRefused || !strings.Contains(st.Reason, "17") || !strings.Contains(st.Reason, "18") {
		t.Fatalf("the integration step ended %s: %q", st.State, st.Reason)
	}
	if strings.Contains(f.lines(), "initdb") || strings.Contains(f.lines(), "-tags") {
		t.Errorf("the refused level ran something:\n%s", f.lines())
	}
	res, st := runLevelTool(t, m, states, "unit")
	if st.State != delivery.StepPassed || !res.Passed || res.Level != "unit" {
		t.Errorf("the unit level: %s %+v", st.State, res)
	}
}

// 1486/AC4: a setup over thirty seconds fails the level naming the setup bound; a run over sixty names the running
// bound and the slowest tests. The clock the commands move stands in for the time they take.
func TestSetupAndRunningTimeAreJudgedApart(t *testing.T) {
	f := newIntegFake(t, "slow")
	f.advance["compile"], f.advance["start"] = 20*time.Second, 15*time.Second
	m, states := integModule(t, f)
	res, _ := runLevelTool(t, m, states, "integration")
	if !res.Passed || !res.OverBound || res.OverWhich != "setup" || res.Setup != 35 || !strings.Contains(res.Why, "setup took 35.00 s, over its setup bound of 30 s") || strings.Contains(res.Why, "slowest tests") {
		t.Errorf("a setup of 35 s: %+v", res)
	}

	g := newIntegFake(t, "slow")
	g.advance["test"] = 61 * time.Second
	m, states = integModule(t, g)
	res, _ = runLevelTool(t, m, states, "integration")
	if !res.Passed || res.OverWhich != "run" || res.Elapsed != 61 || !strings.Contains(res.Why, "took 61.00 s, over its bound of 60 s; the slowest tests: fixture/slow TestTakesAWhile") || strings.Contains(res.Why, "setup") {
		t.Errorf("a run of 61 s: %+v", res)
	}

	h := newIntegFake(t, "slow")
	h.advance["compile"], h.advance["test"] = 29*time.Second, 59*time.Second
	m, states = integModule(t, h)
	if res, _ = runLevelTool(t, m, states, "integration"); !res.Passed || res.OverBound || res.Setup != 29 || res.Elapsed != 59 {
		t.Errorf("29 s of setup and 59 s of running are within both bounds: %+v", res)
	}
}

// 1486/AC5: a run that fails, and a run that panics: the store is stopped, the bus is stopped, the run's directory is
// gone.
func TestAFailedOrPanickedRunStopsTheServersAndRemovesTheRunDirectory(t *testing.T) {
	f := newIntegFake(t, "failing")
	m, states := integModule(t, f)
	res, _ := runLevelTool(t, m, states, "integration")
	if res.Passed {
		t.Fatalf("the failing fixture passed: %+v", res)
	}
	checkCleaned := func(f *integFake, m *Module) {
		t.Helper()
		lines := f.lines()
		if !strings.Contains(lines, "pg_ctl -D "+filepath.Join(f.runDirs[0], "pg", "data")+" -m immediate -w stop") {
			t.Errorf("the store was not stopped:\n%s", lines)
		}
		if strings.LastIndex(lines, " stop") < strings.LastIndex(lines, "-json") {
			t.Errorf("the store was stopped before the tests ended:\n%s", lines)
		}
		if !f.busDown {
			t.Errorf("the bus was not stopped")
		}
		if _, err := os.Stat(f.runDirs[0]); !os.IsNotExist(err) {
			t.Errorf("the run directory %s is still there (%v)", f.runDirs[0], err)
		}
	}
	checkCleaned(f, m)

	g := newIntegFake(t, "failing")
	g.panicOn = "test"
	m, _ = integModule(t, g)
	func() {
		defer func() { recover() }()
		m.integration(context.Background(), "ws1", filepath.Join(m.Work, "ws1", "tree"), nil, "")
	}()
	checkCleaned(g, m)
}

// A store that does not start ends the step failed, with the server stopped and the directory gone.
func TestAStoreThatDoesNotStartEndsTheStepFailedAndLeavesNothing(t *testing.T) {
	f := newIntegFake(t, "slow")
	m, states := integModule(t, f)
	inner := m.Run
	m.Run = func(ctx context.Context, env []string, name string, args ...string) forgekey.Ran {
		r := inner(ctx, env, name, args...)
		if name == "pg_ctl" && strings.HasSuffix(strings.Join(args, " "), " start") {
			return forgekey.Ran{Status: 1, Stderr: "pg_ctl: could not start server\n"}
		}
		return r
	}
	_, st := runLevelTool(t, m, states, "integration")
	if st.State != delivery.StepFailed || !strings.Contains(st.Reason, "pg_ctl: could not start server") {
		t.Errorf("the step ended %s: %q", st.State, st.Reason)
	}
	if !strings.Contains(f.lines(), " stop") {
		t.Errorf("the half-started store was not stopped:\n%s", f.lines())
	}
	if f.busDir != "" {
		t.Errorf("the bus was started after the store failed")
	}
	if entries, _ := os.ReadDir(filepath.Join(m.Work, "ws1")); len(entries) != 1 || entries[0].Name() != "tree" {
		t.Errorf("the workspace holds %v, want only its tree", entries)
	}
}

func TestThePostgresMajorIsReadFromTheVersionLine(t *testing.T) {
	for line, want := range map[string]int{"postgres (PostgreSQL) 18.6": 18, "postgres (PostgreSQL) 9.6.24": 9, "postgres (PostgreSQL) 17.2 (Arch)": 17} {
		f := &integFake{postgres: line, advance: map[string]time.Duration{}}
		m := &Module{Run: f.run}
		if got, err := m.postgresMajor(context.Background()); err != nil || got != want {
			t.Errorf("%q = %d, %v; want %d", line, got, err, want)
		}
	}
	if _, err := (&Module{Run: (&integFake{postgres: "no", advance: map[string]time.Duration{}}).run}).postgresMajor(context.Background()); err == nil {
		t.Errorf("accepted a line that is not postgres's")
	}
}

// Issue 1486 review: postgres refuses a socket path of 108 bytes and over. A run directory deep in the work directory
// gets a short directory of its own for the socket, removed with the store; a short one keeps it under the run.
func TestTheSocketPathStaysUnderTheLimitForALongRunDirectory(t *testing.T) {
	long := filepath.Join(t.TempDir(), strings.Repeat("deep", 20), "ws1", "run-mvh5uc96142806")
	if err := os.MkdirAll(long, 0o700); err != nil {
		t.Fatal(err)
	}
	dir, short, err := socketDir(long)
	if err != nil || !short || len(dir)+len(socketName) >= 100 {
		t.Fatalf("socketDir(long) = %q, %v, %v", dir, short, err)
	}
	if _, err := os.Stat(dir); err != nil {
		t.Errorf("the short directory was not made: %v", err)
	}
	os.RemoveAll(dir)

	near := t.TempDir()
	if dir, short, err := socketDir(near); err != nil || (len(near) < 50 && (short || !strings.HasPrefix(dir, near))) {
		t.Errorf("socketDir(short) = %q, %v, %v", dir, short, err)
	}

	// Through the store: the short directory is removed when the store is stopped.
	f := newIntegFake(t, "slow")
	m, _ := integModule(t, f)
	_, stop, err := m.startStore(context.Background(), long)
	if err != nil {
		t.Fatal(err)
	}
	var sock string
	for _, c := range f.calls {
		if _, a, ok := strings.Cut(c.line, "-k "); ok && strings.HasPrefix(c.line, "pg_ctl") {
			sock, _, _ = strings.Cut(a, " ")
		}
	}
	if sock == "" || len(sock)+len(socketName) >= 100 {
		t.Fatalf("the store was started with the socket %q", sock)
	}
	stop()
	if _, err := os.Stat(sock); !os.IsNotExist(err) {
		t.Errorf("the short socket directory %s outlived the store (%v)", sock, err)
	}
}