felhom-agent/internal/reconcile/engine_test.go

package reconcile

import (
	"context"
	"errors"
	"path/filepath"
	"sync"
	"testing"

	"gitea.dooplex.hu/admin/felhom-agent/internal/hub"
	"gitea.dooplex.hu/admin/felhom-agent/internal/proxmox"
)

// fakeAPI is a configurable GuestAPI for engine tests: it records mutating calls and
// returns canned UPIDs (""=synchronous, non-empty=async) and WaitTask verdicts.
type fakeAPI struct {
	mu  sync.Mutex
	lxc []proxmox.Guest
	cfg map[int]proxmox.GuestConfig

	startUPID, stopUPID, setUPID, resizeUPID string
	startErr, stopErr, setErr, resizeErr     error
	restoreUPID, destroyUPID                 string
	restoreErr, destroyErr                   error
	// status maps vmid -> the Guest returned by GuestStatus (default running if absent).
	status map[int]proxmox.Guest
	// waitFunc maps a UPID to a (status, err); default = OK. Mirrors the real client,
	// which errors on a non-OK exitstatus.
	waitFunc func(upid string) (proxmox.TaskStatus, error)
	// statusFunc backs TaskStatusOnce (crash recovery); default = stopped/OK.
	statusFunc func(upid string) (proxmox.TaskStatus, error)
	// logTailFunc backs TaskLogTail (restore-test start-warning surfacing); default = empty.
	logTailFunc func(upid string) ([]string, error)
	// setFunc, when set, backs SetConfig (drives the F4 lock-500-then-200 test).
	setFunc func(vmid int, params map[string]string) (string, error)
	// restoreHook, when set, fires inside RestoreLXC (used to assert the owning journal entry
	// is written BEFORE the restore — crash-safety ordering).
	restoreHook func()
	// restoreFunc, when set, backs RestoreLXC per-call (drives the F2 band-advance tests:
	// per-vmid "already exists" vs success). Takes precedence over restoreUPID/restoreErr.
	restoreFunc func(opts proxmox.RestoreLXCOptions) (string, error)

	starts   []int
	stops    []int
	sets     []setCall
	resizes  []resizeCall
	restores []proxmox.RestoreLXCOptions
	destroys []int
	waits    []string
	waitOpts []proxmox.WaitOptions // parallel to waits: the options each WaitTask was called with
	listErr  error

	// poolAdds records (pool, vmid) for each PoolAddVMID; poolAddErr backs the failure path.
	poolAdds   []poolAddCall
	poolAddErr error

	// extractCfg/extractErr back ExtractArchiveConfig (GL-5 DR rootfs sizing); extracts records
	// the requested volumes. Empty extractCfg with nil extractErr → a canonical 8G-rootfs config.
	extractCfg string
	extractErr error
	extracts   []string
}

type poolAddCall struct {
	pool string
	vmid int
}

func (f *fakeAPI) PoolAddVMID(_ context.Context, pool string, vmid int) error {
	f.mu.Lock()
	f.poolAdds = append(f.poolAdds, poolAddCall{pool: pool, vmid: vmid})
	err := f.poolAddErr
	f.mu.Unlock()
	return err
}

type resizeCall struct {
	vmid       int
	disk, size string
}

// ExtractArchiveConfig returns extractCfg/extractErr; with neither set it returns a canonical
// minimal archive config (rootfs size=8G) so DR-mode tests that don't care about the rootfs
// override don't have to stage one.
func (f *fakeAPI) ExtractArchiveConfig(_ context.Context, volume string) (string, error) {
	f.mu.Lock()
	f.extracts = append(f.extracts, volume)
	cfg, err := f.extractCfg, f.extractErr
	f.mu.Unlock()
	if err != nil {
		return "", err
	}
	if cfg == "" {
		cfg = "hostname: fake\nrootfs: local-lvm:vm-0-disk-0,size=8G\n"
	}
	return cfg, nil
}

func (f *fakeAPI) RestoreLXC(_ context.Context, opts proxmox.RestoreLXCOptions) (string, error) {
	if f.restoreHook != nil {
		f.restoreHook()
	}
	f.mu.Lock()
	f.restores = append(f.restores, opts)
	fn := f.restoreFunc
	f.mu.Unlock()
	if fn != nil {
		return fn(opts)
	}
	return f.restoreUPID, f.restoreErr
}

func (f *fakeAPI) DestroyLXC(_ context.Context, vmid int) (string, error) {
	f.mu.Lock()
	f.destroys = append(f.destroys, vmid)
	f.mu.Unlock()
	return f.destroyUPID, f.destroyErr
}

func (f *fakeAPI) GuestStatus(_ context.Context, vmid int) (proxmox.Guest, error) {
	f.mu.Lock()
	defer f.mu.Unlock()
	if g, ok := f.status[vmid]; ok {
		return g, nil
	}
	return proxmox.Guest{VMID: vmid, Status: "running"}, nil
}

func (f *fakeAPI) TaskStatusOnce(_ context.Context, upid string) (proxmox.TaskStatus, error) {
	if f.statusFunc != nil {
		return f.statusFunc(upid)
	}
	return proxmox.TaskStatus{UPID: upid, Status: "stopped", ExitStatus: "OK"}, nil
}

func (f *fakeAPI) TaskLogTail(_ context.Context, upid string, _ int) ([]string, error) {
	if f.logTailFunc != nil {
		return f.logTailFunc(upid)
	}
	return nil, nil
}

type setCall struct {
	vmid   int
	params map[string]string
}

func (f *fakeAPI) ListLXC(context.Context) ([]proxmox.Guest, error) {
	if f.listErr != nil {
		return nil, f.listErr
	}
	return f.lxc, nil
}

func (f *fakeAPI) GuestConfig(_ context.Context, vmid int) (proxmox.GuestConfig, error) {
	c, ok := f.cfg[vmid]
	if !ok {
		return proxmox.GuestConfig{}, errors.New("no config")
	}
	return c, nil
}

func (f *fakeAPI) Start(_ context.Context, vmid int) (string, error) {
	f.mu.Lock()
	f.starts = append(f.starts, vmid)
	f.mu.Unlock()
	return f.startUPID, f.startErr
}

func (f *fakeAPI) Stop(_ context.Context, vmid int) (string, error) {
	f.mu.Lock()
	f.stops = append(f.stops, vmid)
	f.mu.Unlock()
	return f.stopUPID, f.stopErr
}

func (f *fakeAPI) SetConfig(_ context.Context, vmid int, params map[string]string) (string, error) {
	f.mu.Lock()
	f.sets = append(f.sets, setCall{vmid, params})
	fn := f.setFunc
	f.mu.Unlock()
	if fn != nil {
		return fn(vmid, params)
	}
	return f.setUPID, f.setErr
}

func (f *fakeAPI) ResizeLXC(_ context.Context, vmid int, disk, size string) (string, error) {
	f.mu.Lock()
	f.resizes = append(f.resizes, resizeCall{vmid, disk, size})
	f.mu.Unlock()
	return f.resizeUPID, f.resizeErr
}

func (f *fakeAPI) WaitTask(_ context.Context, upid string, opts proxmox.WaitOptions) (proxmox.TaskStatus, error) {
	f.mu.Lock()
	f.waits = append(f.waits, upid)
	f.waitOpts = append(f.waitOpts, opts)
	f.mu.Unlock()
	if f.waitFunc != nil {
		return f.waitFunc(upid)
	}
	return proxmox.TaskStatus{Status: "stopped", ExitStatus: "OK"}, nil
}

func newEngine(t *testing.T, api GuestAPI, provider DesiredProvider) (*Engine, *Journal, *Queue) {
	t.Helper()
	jp := filepath.Join(t.TempDir(), "journal.log")
	j, err := OpenJournal(jp)
	if err != nil {
		t.Fatalf("OpenJournal: %v", err)
	}
	t.Cleanup(func() { j.Close() })
	q := NewQueue()
	t.Cleanup(q.Close)
	e := NewEngine(EngineOptions{API: api, Queue: q, Journal: j, Provider: provider})
	return e, j, q
}

func TestEngine_EmptyProviderNoMutations(t *testing.T) {
	api := &fakeAPI{
		lxc: []proxmox.Guest{{VMID: 100, Status: "running"}},
		cfg: map[int]proxmox.GuestConfig{100: {Cores: 2}},
	}
	e, _, _ := newEngine(t, api, EmptyProvider{})
	res, err := e.Reconcile(context.Background())
	if err != nil {
		t.Fatalf("Reconcile: %v", err)
	}
	if res.Planned != 0 || res.Executed != 0 {
		t.Errorf("EmptyProvider should plan nothing, got %+v", res)
	}
	if len(api.starts)+len(api.stops)+len(api.sets) != 0 {
		t.Errorf("EmptyProvider mutated Proxmox: starts=%v stops=%v sets=%v", api.starts, api.stops, api.sets)
	}
}

func TestEngine_AsyncStartWaitsTask(t *testing.T) {
	api := &fakeAPI{
		lxc:       []proxmox.Guest{{VMID: 100, Status: "stopped"}},
		cfg:       map[int]proxmox.GuestConfig{100: {Cores: 2}},
		startUPID: "UPID:demo:start:100:",
	}
	e, j, _ := newEngine(t, api, StaticProvider{State: desired(DesiredGuest{VMID: 100, Run: RunRunning})})
	res, err := e.Reconcile(context.Background())
	if err != nil {
		t.Fatalf("Reconcile: %v", err)
	}
	if res.Executed != 1 || res.Failed != 0 {
		t.Fatalf("want 1 executed, got %+v", res)
	}
	if len(api.starts) != 1 || api.starts[0] != 100 {
		t.Errorf("expected Start(100), got %v", api.starts)
	}
	if len(api.waits) != 1 {
		t.Errorf("async op must WaitTask, got waits=%v", api.waits)
	}
	if len(j.InFlight()) != 0 {
		t.Errorf("no ops should be in-flight after success: %+v", j.InFlight())
	}
}

func TestEngine_SynchronousSetConfigNoWait(t *testing.T) {
	// Empty UPID = PVE applied synchronously (slice-4 proven for description). Must be
	// treated as success WITHOUT a WaitTask call.
	api := &fakeAPI{
		lxc:     []proxmox.Guest{{VMID: 100, Status: "stopped"}},
		cfg:     map[int]proxmox.GuestConfig{100: {Cores: 2}},
		setUPID: "", // synchronous
	}
	e, _, _ := newEngine(t, api, StaticProvider{State: desired(
		DesiredGuest{VMID: 100, Spec: &hub.GuestSpec{Cores: 4, MemoryBytes: mib(2048)}})})
	res, err := e.Reconcile(context.Background())
	if err != nil {
		t.Fatalf("Reconcile: %v", err)
	}
	if res.Executed != 1 {
		t.Fatalf("want 1 executed, got %+v", res)
	}
	if len(api.sets) != 1 || api.sets[0].params["cores"] != "4" {
		t.Errorf("expected SetConfig cores=4, got %v", api.sets)
	}
	if len(api.waits) != 0 {
		t.Errorf("synchronous op must NOT WaitTask, got waits=%v", api.waits)
	}
}

func TestEngine_WaitTaskFailureCountsFailed(t *testing.T) {
	api := &fakeAPI{
		lxc:       []proxmox.Guest{{VMID: 100, Status: "stopped"}},
		cfg:       map[int]proxmox.GuestConfig{100: {Cores: 2}},
		startUPID: "UPID:demo:start:100:",
		waitFunc: func(string) (proxmox.TaskStatus, error) {
			return proxmox.TaskStatus{Status: "stopped", ExitStatus: "got 403"}, errors.New("task failed: got 403")
		},
	}
	e, j, _ := newEngine(t, api, StaticProvider{State: desired(DesiredGuest{VMID: 100, Run: RunRunning})})
	res, err := e.Reconcile(context.Background())
	if err != nil {
		t.Fatalf("Reconcile (pass): %v", err)
	}
	if res.Failed != 1 || res.Executed != 0 {
		t.Fatalf("want 1 failed, got %+v", res)
	}
	// The failed op is journaled terminal (failed), not left in-flight.
	if len(j.InFlight()) != 0 {
		t.Errorf("failed op should be terminal, in-flight=%+v", j.InFlight())
	}
}

func TestEngine_PostErrorCountsFailed(t *testing.T) {
	api := &fakeAPI{
		lxc:      []proxmox.Guest{{VMID: 100, Status: "stopped"}},
		cfg:      map[int]proxmox.GuestConfig{100: {Cores: 2}},
		startErr: errors.New("connection refused"),
	}
	e, _, _ := newEngine(t, api, StaticProvider{State: desired(DesiredGuest{VMID: 100, Run: RunRunning})})
	res, _ := e.Reconcile(context.Background())
	if res.Failed != 1 {
		t.Fatalf("want 1 failed on POST error, got %+v", res)
	}
	if len(api.waits) != 0 {
		t.Errorf("POST error must not reach WaitTask, got %v", api.waits)
	}
}

func TestEngine_ListErrorIsPassFailure(t *testing.T) {
	api := &fakeAPI{listErr: errors.New("api down")}
	e, _, _ := newEngine(t, api, StaticProvider{State: desired(DesiredGuest{VMID: 100, Run: RunRunning})})
	if _, err := e.Reconcile(context.Background()); err == nil {
		t.Error("expected a pass-level error when actual state can't be read")
	}
}