galaxy-game/integration/internal/harness/process.go

package harness

import (
	"bytes"
	"context"
	"errors"
	"fmt"
	"io"
	"net"
	"net/http"
	"os"
	"os/exec"
	"strings"
	"sync"
	"syscall"
	"testing"
	"time"
)

const (
	defaultStartupWait  = 10 * time.Second
	defaultPollInterval = 25 * time.Millisecond
	defaultStopWait     = 5 * time.Second
)

// Process represents one long-lived external service process started by an
// integration suite.
type Process struct {
	name string
	cmd  *exec.Cmd

	logsMu sync.Mutex
	logs   bytes.Buffer

	doneCh  chan struct{}
	waitErr error
}

// StartProcess starts binaryPath with envOverrides and registers cleanup that
// stops the process and prints captured logs on failed tests.
func StartProcess(t testing.TB, name string, binaryPath string, envOverrides map[string]string) *Process {
	t.Helper()

	cmd := exec.Command(binaryPath)
	cmd.Env = mergeEnvironment(os.Environ(), envOverrides)

	process := &Process{
		name:   name,
		cmd:    cmd,
		doneCh: make(chan struct{}),
	}
	cmd.Stdout = process.logWriter()
	cmd.Stderr = process.logWriter()

	if err := cmd.Start(); err != nil {
		t.Fatalf("start %s: %v", name, err)
	}

	go func() {
		process.waitErr = cmd.Wait()
		close(process.doneCh)
	}()

	t.Cleanup(func() {
		process.Stop(t)
		if t.Failed() {
			t.Logf("%s logs:\n%s", name, process.Logs())
		}
	})

	return process
}

// Stop asks the process to terminate gracefully and waits for completion.
func (p *Process) Stop(t testing.TB) {
	t.Helper()

	if p == nil {
		return
	}

	select {
	case <-p.doneCh:
		err := p.waitErr
		if err != nil && !isExpectedProcessExit(err) {
			t.Errorf("%s exited unexpectedly: %v", p.name, err)
		}
		return
	default:
	}

	if p.cmd.Process != nil {
		_ = p.cmd.Process.Signal(syscall.SIGTERM)
	}

	select {
	case <-p.doneCh:
		err := p.waitErr
		if err != nil && !isExpectedProcessExit(err) {
			t.Errorf("%s exited unexpectedly: %v", p.name, err)
		}
	case <-time.After(defaultStopWait):
		if p.cmd.Process != nil {
			_ = p.cmd.Process.Kill()
		}
		<-p.doneCh
		err := p.waitErr
		if err != nil && !isExpectedProcessExit(err) {
			t.Errorf("%s exited unexpectedly: %v", p.name, err)
		}
	}
}

// Logs returns the captured combined stdout/stderr output of the process.
func (p *Process) Logs() string {
	if p == nil {
		return ""
	}

	p.logsMu.Lock()
	defer p.logsMu.Unlock()
	return p.logs.String()
}

// FreeTCPAddress reserves one ephemeral loopback TCP address and releases it
// immediately so a service process can bind to it.
func FreeTCPAddress(t testing.TB) string {
	t.Helper()

	listener, err := net.Listen("tcp", "127.0.0.1:0")
	if err != nil {
		t.Fatalf("reserve free TCP address: %v", err)
	}

	addr := listener.Addr().String()
	if err := listener.Close(); err != nil {
		t.Fatalf("release reserved TCP address: %v", err)
	}

	return addr
}

// WaitForHTTPStatus waits until url responds with wantStatus or fails when the
// backing process exits early.
func WaitForHTTPStatus(t testing.TB, process *Process, url string, wantStatus int) {
	t.Helper()

	client := &http.Client{
		Timeout: 250 * time.Millisecond,
		Transport: &http.Transport{
			DisableKeepAlives: true,
		},
	}
	defer client.CloseIdleConnections()

	ctx, cancel := context.WithTimeout(context.Background(), defaultStartupWait)
	defer cancel()

	ticker := time.NewTicker(defaultPollInterval)
	defer ticker.Stop()

	for {
		if err := processErr(process); err != nil {
			t.Fatalf("%s exited before %s became ready: %v\n%s", process.name, url, err, process.Logs())
		}

		request, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil)
		if err != nil {
			t.Fatalf("build readiness request for %s: %v", url, err)
		}

		response, err := client.Do(request)
		if err == nil {
			_, _ = io.Copy(io.Discard, response.Body)
			response.Body.Close()
			if response.StatusCode == wantStatus {
				return
			}
		}

		select {
		case <-ctx.Done():
			t.Fatalf("wait for %s status %d: %v\n%s", url, wantStatus, ctx.Err(), process.Logs())
		case <-ticker.C:
		}
	}
}

// WaitForTCP waits until address accepts TCP connections or fails when the
// backing process exits early.
func WaitForTCP(t testing.TB, process *Process, address string) {
	t.Helper()

	ctx, cancel := context.WithTimeout(context.Background(), defaultStartupWait)
	defer cancel()

	ticker := time.NewTicker(defaultPollInterval)
	defer ticker.Stop()

	for {
		if err := processErr(process); err != nil {
			t.Fatalf("%s exited before %s became reachable: %v\n%s", process.name, address, err, process.Logs())
		}

		conn, err := net.DialTimeout("tcp", address, 100*time.Millisecond)
		if err == nil {
			_ = conn.Close()
			return
		}

		select {
		case <-ctx.Done():
			t.Fatalf("wait for %s TCP readiness: %v\n%s", address, ctx.Err(), process.Logs())
		case <-ticker.C:
		}
	}
}

func (p *Process) logWriter() io.Writer {
	return writerFunc(func(data []byte) (int, error) {
		p.logsMu.Lock()
		defer p.logsMu.Unlock()
		return p.logs.Write(data)
	})
}

func mergeEnvironment(base []string, overrides map[string]string) []string {
	values := make(map[string]string, len(base)+len(overrides))
	for _, entry := range base {
		name, value, ok := strings.Cut(entry, "=")
		if ok {
			values[name] = value
		}
	}
	for name, value := range overrides {
		values[name] = value
	}

	merged := make([]string, 0, len(values))
	for name, value := range values {
		merged = append(merged, fmt.Sprintf("%s=%s", name, value))
	}
	return merged
}

func processErr(process *Process) error {
	if process == nil {
		return errors.New("nil process")
	}

	select {
	case <-process.doneCh:
		return process.waitErr
	default:
		return nil
	}
}

func isExpectedProcessExit(err error) bool {
	if err == nil {
		return true
	}

	var exitErr *exec.ExitError
	if !errors.As(err, &exitErr) {
		return false
	}

	return exitErr.ExitCode() == -1
}

type writerFunc func([]byte) (int, error)

func (f writerFunc) Write(data []byte) (int, error) {
	return f(data)
}