oikos/internal/core/app/execution_test.go

package app

import (
	"context"
	"errors"
	"strings"
	"testing"
	"time"

	"github.com/dtoro/oikos/internal/core/domain"
	"github.com/dtoro/oikos/internal/core/ports"
	"github.com/dtoro/oikos/internal/core/ports/portstest"
)

func newExecDeps(t *testing.T) (*portstest.ExecutionRecorder, *portstest.RecordingExecutor, *ExecutionService) {
	t.Helper()
	store := portstest.NewGovernanceStore()
	store.PlanSessions[sessionOK] = true
	rec := portstest.NewExecutionRecorder()
	exec := &portstest.RecordingExecutor{}
	resolver := &portstest.FakeResolver{Addr: "10.1.1.1", User: "root"}
	svc := NewExecutionService(NewPolicyService(store), exec, resolver, rec)
	return rec, exec, svc
}

func TestExecutionSubmitRefuseRecordsNothing(t *testing.T) {
	rec, exec, svc := newExecDeps(t)
	res := svc.Submit(context.Background(), ExecutionSubmitCmd{
		AgentID: agentID, TargetID: targetID, TargetSlug: "lxc:x",
		Command: "qm list", SessionID: sessionOK, // host-only on lxc → refuse
	})
	if res.Decision.Action != DecisionRefuse {
		t.Fatalf("action = %q, want refuse", res.Decision.Action)
	}
	if len(rec.Created) != 0 || len(rec.Queued) != 0 || len(exec.Calls) != 0 {
		t.Error("refusal must record nothing and execute nothing")
	}
	if res.ExecutionID != "" {
		t.Error("refusal has no execution id")
	}
}

func TestExecutionSubmitAutoRun(t *testing.T) {
	rec, exec, svc := newExecDeps(t)
	exec.Results = []ports.ExecResult{{Output: "Filesystem 512-blocks..."}}

	res := svc.Submit(context.Background(), ExecutionSubmitCmd{
		AgentID: agentID, TargetID: targetID, TargetSlug: "lxc:x",
		Command: "df -h", Purpose: "check disk", SessionID: sessionOK,
	})
	if res.Decision.Action != DecisionAuto || res.Err != nil {
		t.Fatalf("result = %+v err=%v", res.Decision, res.Err)
	}
	if res.Output != "Filesystem 512-blocks..." {
		t.Errorf("output = %q", res.Output)
	}
	if len(exec.Calls) != 1 || exec.Calls[0].Command != "df -h" {
		t.Errorf("executor calls = %+v", exec.Calls)
	}
	if len(rec.Created) != 1 {
		t.Fatalf("created = %d, want 1", len(rec.Created))
	}
	created := rec.Created[0]
	if created.RiskClass != "read_only" || created.Route != "auto-act" {
		t.Errorf("created = %+v", created)
	}
	if !strings.HasPrefix(created.Action, "run:{") {
		t.Errorf("action col = %q", created.Action)
	}
	if len(rec.Running) != 1 || len(rec.Finalized) != 1 {
		t.Errorf("running=%d finalized=%d", len(rec.Running), len(rec.Finalized))
	}
	fin := rec.Finalized[0]
	if fin.Status != "completed" || string(fin.Result) != `{"output":"Filesystem 512-blocks..."}` {
		t.Errorf("finalized = %+v", fin)
	}
	if fin.ExecutionID != res.ExecutionID || rec.Running[0] != res.ExecutionID {
		t.Errorf("ids inconsistent: fin=%s run=%s res=%s", fin.ExecutionID, rec.Running[0], res.ExecutionID)
	}
	if len(rec.Queued) != 0 {
		t.Errorf("auto-run must not queue approvals, got %+v", rec.Queued)
	}
}

func TestExecutionSubmitAutoRunFailure(t *testing.T) {
	rec, exec, svc := newExecDeps(t)
	exec.Results = []ports.ExecResult{{Output: "boom", Err: errors.New("exit 1")}}

	res := svc.Submit(context.Background(), ExecutionSubmitCmd{
		AgentID: agentID, TargetID: targetID, TargetSlug: "lxc:x",
		Command: "df -h", SessionID: sessionOK,
	})
	if res.Err == nil {
		t.Fatal("expected error surfaced")
	}
	fin := rec.Finalized[0]
	if fin.Status != "failed" || !strings.Contains(string(fin.Result), "exit 1") {
		t.Errorf("finalized = %+v", fin)
	}
}

func TestExecutionSubmitResolverFailureFailsExecution(t *testing.T) {
	rec, exec, _ := newExecDeps(t)
	// A resolver that always errors.
	store := portstest.NewGovernanceStore()
	store.PlanSessions[sessionOK] = true
	failing := &portstest.FakeResolver{Err: errors.New("no IP found")}
	svc := NewExecutionService(NewPolicyService(store), exec, failing, rec)

	res := svc.Submit(context.Background(), ExecutionSubmitCmd{
		AgentID: agentID, TargetID: targetID, TargetSlug: "lxc:x",
		Command: "df -h", SessionID: sessionOK,
	})
	if res.Err == nil {
		t.Fatal("resolver failure must surface")
	}
	if len(exec.Calls) != 0 {
		t.Error("no execution on resolve failure")
	}
	if rec.Finalized[0].Status != "failed" {
		t.Errorf("finalized = %+v, want failed", rec.Finalized[0])
	}
}

func TestExecutionSubmitQueue(t *testing.T) {
	rec, _, svc := newExecDeps(t)
	res := svc.Submit(context.Background(), ExecutionSubmitCmd{
		AgentID: agentID, TargetID: targetID, TargetSlug: "lxc:x",
		Command: "systemctl enable caddy", Purpose: "enable service", SessionID: sessionOK,
	})
	if res.Decision.Action != DecisionQueue {
		t.Fatalf("action = %q, want queue", res.Decision.Action)
	}
	if len(rec.Queued) != 1 {
		t.Fatalf("queued = %d, want 1", len(rec.Queued))
	}
	q := rec.Queued[0]
	if q.RiskClass != "config_mutation" || q.SessionID != sessionOK || q.ExecutionID != res.ExecutionID {
		t.Errorf("queued = %+v", q)
	}
	if len(rec.Finalized) != 0 {
		t.Error("queued submission must not finalize")
	}
}

func TestExecutionSubmitAsync(t *testing.T) {
	rec, exec, _ := newExecDeps(t)
	exec.Results = []ports.ExecResult{{Output: "done"}}

	// Wrap the executor to block until released.
	release := make(chan struct{})
	blocking := &blockingExecutor{inner: exec, release: release}
	// sleep-classified commands are config_mutation; open the assent
	// window so the decision is auto-run, then dispatch async.
	store := portstest.NewGovernanceStore()
	store.PlanSessions[sessionOK] = true
	store.Assent[string(agentID)+"/"+sessionOK] = true
	svc := NewExecutionService(NewPolicyService(store), blocking, &portstest.FakeResolver{Addr: "10.0.0.1", User: "root"}, rec)

	res := svc.Submit(context.Background(), ExecutionSubmitCmd{
		AgentID: agentID, TargetID: targetID, TargetSlug: "lxc:x",
		Command: "sleep 60", SessionID: sessionOK, Async: true,
	})
	if !res.AsyncStarted {
		t.Fatalf("result = %+v, want async started", res)
	}
	if res.ExecutionID == "" {
		t.Fatal("async submission must return the execution id synchronously")
	}
	close(release)
	deadline := time.Now().Add(2 * time.Second)
	for time.Now().Before(deadline) {
		rec.MuLock()
		n := len(rec.Finalized)
		rec.MuUnlock()
		if n == 1 {
			return
		}
		time.Sleep(5 * time.Millisecond)
	}
	t.Fatal("async execution did not finalize")
}

type blockingExecutor struct {
	inner   *portstest.RecordingExecutor
	release chan struct{}
}

func (b *blockingExecutor) Run(ctx context.Context, target ports.Target, command string, opts ports.ExecOpts) ports.ExecResult {
	<-b.release
	return b.inner.Run(ctx, target, command, opts)
}

func TestExecutionSubmitSinkFactoryStreaming(t *testing.T) {
	_, exec, svc := newExecDeps(t)
	exec.Results = []ports.ExecResult{{Output: "x"}}

	var streams []string
	flushed := false
	res := svc.Submit(context.Background(), ExecutionSubmitCmd{
		AgentID: agentID, TargetID: targetID, TargetSlug: "lxc:x",
		Command: "df -h", SessionID: sessionOK,
		SinkFactory: func(ctx context.Context, id domain.UUID, corr string) (func(string, []byte), func()) {
			return func(stream string, chunk []byte) { streams = append(streams, stream) },
				func() { flushed = true }
		},
	})
	if res.Err != nil {
		t.Fatalf("submit: %v", res.Err)
	}
	if !flushed {
		t.Error("sink flush must run after execution")
	}
}

func TestExecutionSubmitRecorderErrors(t *testing.T) {
	for _, phase := range []string{"create", "queue", "mark", "finalize"} {
		t.Run(phase, func(t *testing.T) {
			rec, exec, _ := newExecDeps(t)
			rec.ErrStubs[phase] = errors.New("db down")
			store := portstest.NewGovernanceStore()
			store.PlanSessions[sessionOK] = true
			svc := NewExecutionService(NewPolicyService(store), exec, &portstest.FakeResolver{Addr: "10.0.0.1", User: "root"}, rec)

			// create/mark/finalize fail on the auto path; queue fails on
			// the gated path.
			cmd := ExecutionSubmitCmd{
				AgentID: agentID, TargetID: targetID, TargetSlug: "lxc:x",
				Command: "systemctl enable caddy", SessionID: sessionOK,
			}
			if phase == "create" || phase == "mark" || phase == "finalize" {
				cmd.Command = "df -h" // auto-run path
			}
			res := svc.Submit(context.Background(), cmd)
			switch phase {
			case "create", "queue":
				if res.Err == nil {
					t.Errorf("phase %s: expected surfaced error", phase)
				}
			default:
				// mark/finalize failures are logged, not fatal: the
				// execution still reports its outcome.
				if res.Err != nil && phase == "mark" {
					t.Errorf("phase %s: mark failure must not fail submit: %v", phase, res.Err)
				}
			}
		})
	}
}

func TestExecutionFinalizeErrorNotFatal(t *testing.T) {
	rec, exec, _ := newExecDeps(t)
	rec.ErrStubs["finalize"] = errors.New("write failed")
	store := portstest.NewGovernanceStore()
	store.PlanSessions[sessionOK] = true
	svc := NewExecutionService(NewPolicyService(store), exec, &portstest.FakeResolver{Addr: "10.0.0.1", User: "root"}, rec)

	res := svc.Submit(context.Background(), ExecutionSubmitCmd{
		AgentID: agentID, TargetID: targetID, TargetSlug: "lxc:x",
		Command: "df -h", SessionID: sessionOK,
	})
	if res.Err != nil {
		t.Errorf("finalize failure must not fail the submit: %v", res.Err)
	}
	if res.Output != "" {
		t.Errorf("output = %q, want the executed output", res.Output)
	}
}

func TestExecutionSubmitHostHint(t *testing.T) {
	svc, _ := newPolicySvc(t)
	svc.HostHint = func(ctx context.Context, slug string) string { return "host:strong" }
	d := svc.Decide(context.Background(), submit("lxc:dns", "pct list", ""))
	if d.Action != DecisionRefuse || !strings.Contains(d.Message, "host:strong") {
		t.Errorf("decision = %+v, want hint host:strong", d)
	}
	// Hint returning "" falls back to the default.
	svc.HostHint = func(ctx context.Context, slug string) string { return "" }
	d = svc.Decide(context.Background(), submit("lxc:dns", "pvesh get /x", ""))
	if !strings.Contains(d.Message, "host:hubris or host:strong") {
		t.Errorf("default hint missing: %s", d.Message)
	}
}