orca/internal/cli/job_dispatch_test.go

package cli

import (
	"bytes"
	"context"
	"os"
	"path/filepath"
	"strings"
	"sync"
	"testing"
	"time"

	"git.cloudinit.dev/coreci/orca/internal/certpaths"
	"git.cloudinit.dev/coreci/orca/internal/jobspec"
	"git.cloudinit.dev/coreci/orca/internal/model"
	"git.cloudinit.dev/coreci/orca/internal/store"
)

// mockDispatchTransport is a test double for jobDispatchTransport. It
// records calls and returns configured errors. The zero value succeeds
// for every call.
type mockDispatchTransport struct {
	mu          sync.Mutex
	writeCalls  []mockDispatchWriteCall
	execCalls   []mockDispatchExecCall
	writeErr    error // returned by WriteFile (simulates C-44 push failure)
	execErr     error
	closeCalled bool
}

type mockDispatchWriteCall struct {
	Peer    string
	Path    string
	Content string
	Mode    os.FileMode
}

type mockDispatchExecCall struct {
	Peer string
	Cmd  string
}

func (m *mockDispatchTransport) WriteFile(ctx context.Context, peer, path string, content []byte, mode os.FileMode) error {
	m.mu.Lock()
	defer m.mu.Unlock()
	m.writeCalls = append(m.writeCalls, mockDispatchWriteCall{Peer: peer, Path: path, Content: string(content), Mode: mode})
	return m.writeErr
}

func (m *mockDispatchTransport) Exec(ctx context.Context, peer, cmd string) ([]byte, error) {
	m.mu.Lock()
	defer m.mu.Unlock()
	m.execCalls = append(m.execCalls, mockDispatchExecCall{Peer: peer, Cmd: cmd})
	return nil, m.execErr
}

func (m *mockDispatchTransport) Close() error {
	m.mu.Lock()
	defer m.mu.Unlock()
	m.closeCalled = true
	return nil
}

// insertRemoteNode registers a ready remote (non-localhost) node in the
// test DB so the scheduler sees it as a candidate.
func insertRemoteNode(t *testing.T, name, addr string) {
	t.Helper()
	db, err := store.Open(certpaths.DBPath())
	if err != nil {
		t.Fatalf("open db: %v", err)
	}
	defer db.Close()
	repo := store.NewNodeRepo(db)
	if err := repo.Insert(context.Background(), &model.Node{
		ID:       name,
		Name:     name,
		Address:  addr,
		State:    model.NodeStateReady,
		JoinedAt: time.Now().UTC(),
		LastSeen: time.Now().UTC(),
		Kind:     string(model.NodeKindLinux),
		OS:       "linux",
	}); err != nil {
		t.Fatalf("insert node %s: %v", name, err)
	}
}

// writeJobMDSpec writes a Markdown jobspec to a temp file and returns
// the path.
func writeJobMDSpec(t *testing.T, content string) string {
	t.Helper()
	dir := t.TempDir()
	p := filepath.Join(dir, "spec.md")
	if err := os.WriteFile(p, []byte(content), 0o644); err != nil {
		t.Fatalf("write spec: %v", err)
	}
	return p
}

const mdJobTrue = "---\n" +
	"kind: Job\n" +
	"name: true-job\n" +
	"runtime:\n" +
	"  one_of: process\n" +
	"  command: /bin/true\n" +
	"---\n# True\n\nRuns /bin/true.\n"

// TestREQ151_LocalFallbackNoRemoteNodes (T13): `job run` with no remote
// nodes registered (only localhost or none) runs locally via the
// executor. The output says "Job complete" (local), not "deployed".
func TestREQ151_LocalFallbackNoRemoteNodes(t *testing.T) {
	_, cleanup := initTestEnv(t)
	defer cleanup()
	resetRootFlags(t)
	spec := writeJobMDSpec(t, mdJobTrue)
	var buf bytes.Buffer
	rootCmd.SetOut(&buf)
	rootCmd.SetErr(&buf)
	rootCmd.SetArgs([]string{"job", "run", spec})
	if err := rootCmd.Execute(); err != nil {
		t.Fatalf("job run local fallback: %v\n%s", err, buf.String())
	}
	out := buf.String()
	if !strings.Contains(out, "Job complete") {
		t.Errorf("expected local 'Job complete' output, got: %s", out)
	}
	if strings.Contains(out, "deployed") {
		t.Errorf("did not expect 'deployed' for local fallback, got: %s", out)
	}
}

// TestREQ151_RemoteNodeScheduledAndPushed (T7): `job run` with a remote
// node registered invokes the scheduler and SSH-pushes the unit. The
// mock transport records the write and the output says "deployed".
func TestREQ151_RemoteNodeScheduledAndPushed(t *testing.T) {
	_, cleanup := initTestEnv(t)
	defer cleanup()
	insertRemoteNode(t, "worker-1", "10.0.0.5:8443")
	resetRootFlags(t)

	mock := &mockDispatchTransport{}
	prev := jobDispatchTransportOverride
	jobDispatchTransportOverride = mock
	defer func() { jobDispatchTransportOverride = prev }()

	spec := writeJobMDSpec(t, mdJobTrue)
	var buf bytes.Buffer
	rootCmd.SetOut(&buf)
	rootCmd.SetErr(&buf)
	rootCmd.SetArgs([]string{"job", "run", spec})
	if err := rootCmd.Execute(); err != nil {
		t.Fatalf("job run remote: %v\n%s", err, buf.String())
	}
	out := buf.String()
	if !strings.Contains(out, "deployed to worker-1") {
		t.Errorf("expected 'deployed to worker-1', got: %s", out)
	}
	if len(mock.writeCalls) == 0 {
		t.Errorf("expected SSH-push write calls, got 0")
	}
	// The unit path should be the orca-v1 systemd unit.
	wrote := false
	for _, c := range mock.writeCalls {
		if strings.HasSuffix(c.Path, "orca-v1-true-job.service") {
			wrote = true
			if !strings.Contains(c.Content, "ExecStart=/bin/true") {
				t.Errorf("unit content missing ExecStart:\n%s", c.Content)
			}
		}
	}
	if !wrote {
		t.Errorf("no write to orca-v1-true-job.service; calls=%+v", mock.writeCalls)
	}
}

// TestREQ151_C44_PushFailureReturnsError (T14, binding condition
// C-44): when the scheduler selects a remote node but SSH-push fails,
// `job run` returns an error. It does NOT silently fall back to local
// execution.
func TestREQ151_C44_PushFailureReturnsError(t *testing.T) {
	_, cleanup := initTestEnv(t)
	defer cleanup()
	insertRemoteNode(t, "worker-1", "10.0.0.5:8443")
	resetRootFlags(t)

	mock := &mockDispatchTransport{writeErr: errMockPush}
	prev := jobDispatchTransportOverride
	jobDispatchTransportOverride = mock
	defer func() { jobDispatchTransportOverride = prev }()

	spec := writeJobMDSpec(t, mdJobTrue)
	var buf bytes.Buffer
	rootCmd.SetOut(&buf)
	rootCmd.SetErr(&buf)
	rootCmd.SetArgs([]string{"job", "run", spec})
	err := rootCmd.Execute()
	if err == nil {
		t.Fatal("expected error for SSH-push failure (C-44), got nil")
	}
	out := buf.String()
	// Must NOT have fallen back to local execution.
	if strings.Contains(out, "Job complete") {
		t.Errorf("C-44 violation: silently fell back to local exec on push failure:\n%s", out)
	}
	if !strings.Contains(err.Error(), "push") {
		t.Errorf("error should mention push failure, got: %v", err)
	}
}

// TestREQ151_TargetOverridesScheduler (T6): --target pins to the named
// node, bypassing the scheduler bin-packing.
func TestREQ151_TargetOverridesScheduler(t *testing.T) {
	_, cleanup := initTestEnv(t)
	defer cleanup()
	// Register two remote nodes; --target forces the specific one
	// even if the scheduler would prefer the other.
	insertRemoteNode(t, "worker-1", "10.0.0.5:8443")
	insertRemoteNode(t, "worker-2", "10.0.0.6:8443")
	resetRootFlags(t)

	mock := &mockDispatchTransport{}
	prev := jobDispatchTransportOverride
	jobDispatchTransportOverride = mock
	defer func() { jobDispatchTransportOverride = prev }()

	spec := writeJobMDSpec(t, mdJobTrue)
	var buf bytes.Buffer
	rootCmd.SetOut(&buf)
	rootCmd.SetErr(&buf)
	rootCmd.SetArgs([]string{"job", "run", spec, "--target", "worker-2"})
	if err := rootCmd.Execute(); err != nil {
		t.Fatalf("job run --target: %v\n%s", err, buf.String())
	}
	out := buf.String()
	if !strings.Contains(out, "deployed to worker-2") {
		t.Errorf("expected --target to pin worker-2, got: %s", out)
	}
	// The push must go to worker-2's SSH peer (10.0.0.6:22).
	if len(mock.writeCalls) == 0 {
		t.Fatalf("expected SSH-push write calls, got 0")
	}
	for _, c := range mock.writeCalls {
		if !strings.HasPrefix(c.Peer, "10.0.0.6:") {
			t.Errorf("push peer = %q, want 10.0.0.6:* (worker-2)", c.Peer)
		}
	}
}

// TestREQ151_SchedulerNoFittingNodeErrors (C-44): a remote node is
// registered but the workload's runtime/constraint excludes it; the
// scheduler returns an error (no local fallback).
func TestREQ151_SchedulerNoFittingNodeErrors(t *testing.T) {
	_, cleanup := initTestEnv(t)
	defer cleanup()
	insertRemoteNode(t, "worker-1", "10.0.0.5:8443")
	resetRootFlags(t)

	// A wasm workload cannot fit a process-only node.
	spec := writeJobMDSpec(t, "---\nkind: Job\nname: wjob\nruntime:\n  one_of: wasm\n  command: /bin/true\n---\nbody\n")
	var buf bytes.Buffer
	rootCmd.SetOut(&buf)
	rootCmd.SetErr(&buf)
	rootCmd.SetArgs([]string{"job", "run", spec})
	err := rootCmd.Execute()
	if err == nil {
		t.Fatal("expected error for no-fitting node, got nil")
	}
	out := buf.String()
	if strings.Contains(out, "Job complete") {
		t.Errorf("C-44 violation: fell back to local exec when no node fit:\n%s", out)
	}
}

// errMockPush is the sentinel returned by the mock transport on push
// failure.
var errMockPush = &mockPushError{}

type mockPushError struct{}

func (e *mockPushError) Error() string { return "mock push failure" }

// TestREQ151_VerifySystemdUnitSkipsWhenNoSystemdAnalyse ensures the
// T9 verify step is a no-op (not an error) when systemd-analyze is not
// on PATH (common on dev/macOS test boxes).
func TestREQ151_VerifySystemdUnitSkipsWhenNoSystemdAnalyse(t *testing.T) {
	// Save PATH and strip systemd-analyze if present. Most CI/dev
	// boxes don't have it; if they do, we remove it from PATH for
	// this test by pointing PATH at an empty dir.
	dir := t.TempDir()
	t.Setenv("PATH", dir)
	err := verifySystemdUnit(context.Background(), "/etc/systemd/system/foo.service", "[Service]\nExecStart=/bin/true\n")
	if err != nil {
		t.Errorf("verifySystemdUnit should skip when systemd-analyze missing, got: %v", err)
	}
}

// TestREQ151_NodeToNodeInfoProjection verifies the projection from
// model.Node + capacity into scheduler.NodeInfo.
func TestREQ151_NodeToNodeInfoProjection(t *testing.T) {
	n := &model.Node{
		ID:      "n1",
		Name:    "worker-1",
		Address: "10.0.0.5:8443",
		Kind:    string(model.NodeKindLinux),
		Metadata: map[string]string{
			"tags": "ssd,fast",
		},
	}
	cap := &store.NodeCapacity{NodeID: "n1", CPUMillicores: 4000, MemoryMiB: 8192}
	ni := nodeToNodeInfo(n, cap)
	if ni.Hostname != "worker-1" {
		t.Errorf("Hostname = %q, want worker-1", ni.Hostname)
	}
	if ni.Kind != "linux" {
		t.Errorf("Kind = %q, want linux", ni.Kind)
	}
	if ni.FreeCPU != 4000 || ni.FreeMem != 8192 {
		t.Errorf("FreeCPU=%d FreeMem=%d, want 4000/8192", ni.FreeCPU, ni.FreeMem)
	}
	if len(ni.Tags) != 2 || ni.Tags[0] != "ssd" || ni.Tags[1] != "fast" {
		t.Errorf("Tags = %v, want [ssd fast]", ni.Tags)
	}

	// Proxmox node.
	pn := &model.Node{Name: "pve-1", Address: "10.0.0.9:8443", Kind: string(model.NodeKindProxmox)}
	pni := nodeToNodeInfo(pn, nil)
	if pni.Kind != "proxmox" {
		t.Errorf("Kind = %q, want proxmox", pni.Kind)
	}
	found := false
	for _, r := range pni.Runtimes {
		if r == "proxmox" {
			found = true
		}
	}
	if !found {
		t.Errorf("proxmox node missing 'proxmox' runtime: %v", pni.Runtimes)
	}
}

// TestREQ151_SSHPeerFor verifies the SSH peer address derivation.
func TestREQ151_SSHPeerFor(t *testing.T) {
	cases := []struct {
		addr string
		meta map[string]string
		want string
	}{
		{"10.0.0.5:8443", nil, "10.0.0.5:22"},
		{"10.0.0.5:8443", map[string]string{"ssh_port": "2222"}, "10.0.0.5:2222"},
		{"host.example.com:8443", nil, "host.example.com:22"},
	}
	for _, c := range cases {
		n := &model.Node{Address: c.addr, Metadata: c.meta}
		got := sshPeerFor(n)
		if got != c.want {
			t.Errorf("sshPeerFor(%q) = %q, want %q", c.addr, got, c.want)
		}
	}
}

// TestREQ151_DispatchDecisionLocal ensures dispatchDecision returns
// "local" when no remote nodes are registered.
func TestREQ151_DispatchDecisionLocal(t *testing.T) {
	_, cleanup := initTestEnv(t)
	defer cleanup()
	spec := &jobspec.WorkloadSpec{Kind: "Job", Name: "x", Count: 1, Runtime: &jobspec.RuntimeBlock{OneOf: "process", Command: "/bin/true"}}
	res, _, err := dispatchDecision(context.Background(), spec, "")
	if err != nil {
		t.Fatalf("dispatchDecision: %v", err)
	}
	if res.mode != "local" {
		t.Errorf("mode = %q, want local (no remote nodes)", res.mode)
	}
}

// TestREQ151_DispatchDecisionRemote ensures dispatchDecision returns
// "remote" when a remote node is registered and fits.
func TestREQ151_DispatchDecisionRemote(t *testing.T) {
	_, cleanup := initTestEnv(t)
	defer cleanup()
	insertRemoteNode(t, "worker-1", "10.0.0.5:8443")
	spec := &jobspec.WorkloadSpec{Kind: "Job", Name: "x", Count: 1, Runtime: &jobspec.RuntimeBlock{OneOf: "process", Command: "/bin/true"}}
	res, nodes, err := dispatchDecision(context.Background(), spec, "")
	if err != nil {
		t.Fatalf("dispatchDecision: %v", err)
	}
	if res.mode != "remote" {
		t.Errorf("mode = %q, want remote", res.mode)
	}
	if res.node != "worker-1" {
		t.Errorf("node = %q, want worker-1", res.node)
	}
	if _, ok := nodes["worker-1"]; !ok {
		t.Errorf("nodes map missing worker-1")
	}
}

// TestREQ151_DispatchDecisionTarget ensures --target pins to the named
// node even when no other remote nodes exist.
func TestREQ151_DispatchDecisionTarget(t *testing.T) {
	_, cleanup := initTestEnv(t)
	defer cleanup()
	insertRemoteNode(t, "worker-9", "10.0.0.9:8443")
	spec := &jobspec.WorkloadSpec{Kind: "Job", Name: "x", Count: 1, Runtime: &jobspec.RuntimeBlock{OneOf: "process", Command: "/bin/true"}}
	res, _, err := dispatchDecision(context.Background(), spec, "worker-9")
	if err != nil {
		t.Fatalf("dispatchDecision: %v", err)
	}
	if res.mode != "remote" || res.node != "worker-9" {
		t.Errorf("result = %+v, want remote/worker-9", res)
	}
}

// TestREQ151_DispatchDecisionTargetNotFound ensures a bad --target
// returns an error (no fallback).
func TestREQ151_DispatchDecisionTargetNotFound(t *testing.T) {
	_, cleanup := initTestEnv(t)
	defer cleanup()
	insertRemoteNode(t, "worker-1", "10.0.0.5:8443")
	spec := &jobspec.WorkloadSpec{Kind: "Job", Name: "x", Count: 1, Runtime: &jobspec.RuntimeBlock{OneOf: "process", Command: "/bin/true"}}
	_, _, err := dispatchDecision(context.Background(), spec, "no-such-node")
	if err == nil {
		t.Fatal("expected error for unknown --target, got nil")
	}
}