Cyrene_v2/internal/runtime/task/task_test.go

package task

import (
	"context"
	"errors"
	"io"
	"log/slog"
	"testing"
	"time"

	"git.yeij.top/AskaEth/Cyrene_v2/internal/bus"
	"git.yeij.top/AskaEth/Cyrene_v2/internal/config"
	"git.yeij.top/AskaEth/Cyrene_v2/internal/logger"
	"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/exec"
)

// mockProvider is a configurable exec.Provider. It mocks the interface, not a
// concrete type, per the test conventions.
type mockProvider struct {
	execute func(ctx context.Context, req exec.ExecRequest) (*exec.ExecResult, error)
}

func (m mockProvider) Execute(ctx context.Context, req exec.ExecRequest) (*exec.ExecResult, error) {
	if m.execute == nil {
		return &exec.ExecResult{}, nil
	}
	return m.execute(ctx, req)
}

// mockRun returns a provider that sleeps (unless sleep==0) and then returns the
// given output or error. It honors ctx cancellation so Abort releases it.
func mockRun(out string, err error, sleep time.Duration) mockProvider {
	return mockProvider{execute: func(ctx context.Context, req exec.ExecRequest) (*exec.ExecResult, error) {
		if sleep > 0 {
			select {
			case <-ctx.Done():
				return nil, ctx.Err()
			case <-time.After(sleep):
			}
		}
		if err != nil {
			return nil, err
		}
		return &exec.ExecResult{Stdout: out, ExitCode: 0}, nil
	}}
}

func discardLogger() *slog.Logger {
	return logger.New(logger.Options{Level: "error", Out: io.Discard})
}

func newTestManager(t *testing.T, p exec.Provider) *Manager {
	t.Helper()
	m, err := New(&config.Config{}, p, discardLogger())
	if err != nil {
		t.Fatalf("New: %v", err)
	}
	t.Cleanup(func() { _ = m.Close() })
	return m
}

func subscribeDone(t *testing.T, m *Manager) (<-chan DoneEvent, func()) {
	t.Helper()
	ch := make(chan DoneEvent, 16)
	unsub, err := m.Bus().Subscribe(context.Background(), "task.done", func(_ context.Context, e bus.Event) error {
		if de, ok := e.Payload.(DoneEvent); ok {
			ch <- de
		}
		return nil
	})
	if err != nil {
		t.Fatalf("subscribe task.done: %v", err)
	}
	return ch, unsub
}

// waitTerminal polls a task until it reaches a terminal status or the timeout.
func waitTerminal(tk *Task, timeout time.Duration) (Snapshot, bool) {
	deadline := time.Now().Add(timeout)
	for time.Now().Before(deadline) {
		if s := tk.Snapshot(); s.Status.isTerminal() {
			return s, true
		}
		time.Sleep(5 * time.Millisecond)
	}
	return tk.Snapshot(), false
}

func waitActive(t *testing.T, m *Manager, want int, timeout time.Duration) {
	t.Helper()
	deadline := time.Now().Add(timeout)
	for time.Now().Before(deadline) {
		if m.Active() == want {
			return
		}
		time.Sleep(5 * time.Millisecond)
	}
	t.Fatalf("Active()=%d, want %d", m.Active(), want)
}

// TestManager_New_Validates checks the constructor's nil guards.
func TestManager_New_Validates(t *testing.T) {
	t.Parallel()
	if _, err := New(&config.Config{}, nil, discardLogger()); !errors.Is(err, ErrNilProvider) {
		t.Errorf("nil provider: err=%v, want ErrNilProvider", err)
	}
	if _, err := New(&config.Config{}, mockProvider{}, nil); !errors.Is(err, ErrNilLogger) {
		t.Errorf("nil logger: err=%v, want ErrNilLogger", err)
	}
}

// TestManager_BackgroundTask_RealTimeOutput verifies a background task returns
// immediately, that Output() is readable (empty while running) during the run
// and populated after, and that ReadOutput returns the final output.
func TestManager_BackgroundTask_RealTimeOutput(t *testing.T) {
	t.Parallel()
	m := newTestManager(t, mockRun("hello world", nil, 400*time.Millisecond))

	tk, err := m.Start(context.Background(), TaskSpec{Command: []string{"echo", "hi"}, Background: true})
	if err != nil {
		t.Fatalf("Start: %v", err)
	}
	if tk.Kind != KindBackground {
		t.Errorf("Kind=%s, want background", tk.Kind)
	}
	if got := tk.Output().Stdout; got != "" {
		t.Errorf("output during run=%q, want empty (still running)", got)
	}

	out, err := tk.ReadOutput(context.Background())
	if err != nil {
		t.Fatalf("ReadOutput: %v", err)
	}
	if out.Stdout != "hello world" {
		t.Errorf("Stdout=%q, want hello world", out.Stdout)
	}
	if snap := tk.Snapshot(); snap.Status != StatusDone {
		t.Errorf("Status=%s, want done", snap.Status)
	}
}

// TestManager_TaskOutputAndDoneEvents verifies both bus subjects fire, in order,
// with the expected payloads.
func TestManager_TaskOutputAndDoneEvents(t *testing.T) {
	t.Parallel()
	m := newTestManager(t, mockRun("hi out", nil, 0))

	ch := make(chan bus.Event, 16)
	subCtx, cancel := context.WithCancel(context.Background())
	defer cancel()
	unsub, err := m.Bus().Subscribe(subCtx, "task.>", func(_ context.Context, e bus.Event) error {
		ch <- e
		return nil
	})
	if err != nil {
		t.Fatalf("subscribe task.>: %v", err)
	}
	defer unsub()

	tk, err := m.Start(context.Background(), TaskSpec{Command: []string{"echo"}, Background: true})
	if err != nil {
		t.Fatalf("Start: %v", err)
	}

	var gotOutput, gotDone bool
	timeout := time.After(2 * time.Second)
	for !gotDone {
		select {
		case e := <-ch:
			switch e.Subject {
			case "task.output":
				gotOutput = true
				if oe := e.Payload.(OutputEvent); oe.Output.Stdout != "hi out" {
					t.Errorf("output event Stdout=%q", oe.Output.Stdout)
				}
			case "task.done":
				gotDone = true
				de := e.Payload.(DoneEvent)
				if de.ID != tk.ID {
					t.Errorf("done ID=%q, want %q", de.ID, tk.ID)
				}
				if de.Status != StatusDone {
					t.Errorf("done Status=%s, want done", de.Status)
				}
				if de.Output.Stdout != "hi out" {
					t.Errorf("done Stdout=%q, want hi out", de.Output.Stdout)
				}
			}
		case <-timeout:
			t.Fatal("timeout waiting for bus events")
		}
	}
	if !gotOutput {
		t.Errorf("task.output event not received")
	}
}

// TestManager_ForegroundTask_FastCompletes verifies a foreground task that
// finishes inside the threshold is returned already done.
func TestManager_ForegroundTask_FastCompletes(t *testing.T) {
	t.Parallel()
	m := newTestManager(t, mockRun("fast out", nil, 0))

	start := time.Now()
	tk, err := m.Start(context.Background(), TaskSpec{Command: []string{"echo", "fast"}, Background: false})
	if err != nil {
		t.Fatalf("Start: %v", err)
	}
	if elapsed := time.Since(start); elapsed > 1500*time.Millisecond {
		t.Errorf("Start took %v, want < 1.5s", elapsed)
	}
	if tk.Kind != KindForeground {
		t.Errorf("Kind=%s, want foreground", tk.Kind)
	}
	snap := tk.Snapshot()
	if snap.Status != StatusDone {
		t.Errorf("Status=%s, want done", snap.Status)
	}
	if snap.Output.Stdout != "fast out" {
		t.Errorf("Stdout=%q, want fast out", snap.Output.Stdout)
	}
	// The monotonic clock may be too coarse to tick between StartedAt and
	// FinishedAt, so a task that finishes in less than one tick legitimately
	// reads the same stamp for both. Assert the platform-independent invariant
	// instead: the finish time is never before the start time.
	if snap.FinishedAt.Before(snap.StartedAt) {
		t.Errorf("FinishedAt=%v should not be before StartedAt=%v", snap.FinishedAt, snap.StartedAt)
	}
}

// TestManager_ForegroundTask_PromotedAfterThreshold verifies a foreground task
// that outlives the threshold is promoted: Start returns early with Kind=background
// and Status=running, and the task completes in the background.
func TestManager_ForegroundTask_PromotedAfterThreshold(t *testing.T) {
	t.Parallel()
	m := newTestManager(t, mockRun("slow out", nil, 3*time.Second))

	start := time.Now()
	tk, err := m.Start(context.Background(), TaskSpec{Command: []string{"echo", "slow"}, Background: false})
	if err != nil {
		t.Fatalf("Start: %v", err)
	}
	if elapsed := time.Since(start); elapsed >= 3*time.Second {
		t.Errorf("Start waited %v; should promote before completion", elapsed)
	}
	if tk.Kind != KindBackground {
		t.Errorf("Kind=%s, want background after promotion", tk.Kind)
	}
	if tk.Status() != StatusRunning {
		t.Errorf("Status=%s, want running", tk.Status())
	}

	snap, ok := waitTerminal(tk, 4*time.Second)
	if !ok {
		t.Fatalf("task did not finish after promotion, status=%s", tk.Status())
	}
	if snap.Status != StatusDone {
		t.Errorf("Status=%s, want done", snap.Status)
	}
	if snap.Output.Stdout != "slow out" {
		t.Errorf("Stdout=%q, want slow out", snap.Output.Stdout)
	}
}

// TestManager_Foreground_ParentContextCancelAborts verifies a foreground Start is
// aborted with the parent context's error when that context is canceled.
func TestManager_Foreground_ParentContextCancelAborts(t *testing.T) {
	t.Parallel()
	m := newTestManager(t, mockRun("hang", nil, 5*time.Second))

	ctx, cancel := context.WithCancel(context.Background())
	go func() {
		time.Sleep(100 * time.Millisecond)
		cancel()
	}()

	start := time.Now()
	_, err := m.Start(ctx, TaskSpec{Command: []string{"hang"}, Background: false})
	if err == nil {
		t.Fatal("Start: want error on parent context cancel")
	}
	if !errors.Is(err, context.Canceled) {
		t.Errorf("err=%v, want context.Canceled", err)
	}
	if elapsed := time.Since(start); elapsed > 2*time.Second {
		t.Errorf("Start took %v, want < 2s", elapsed)
	}
}