Function TestTaskManager

internal/taskmanager/manager_test.go:27–98

(t *testing.T)

func (m *mockLogger) Error(format string, args ...interface{}) {}

func TestTaskManager(t *testing.T) {
	// Setup Mock Server
	state := mockpve.NewMockState()
	r := mux.NewRouter()

	// Register relevant handlers
	r.HandleFunc("/api2/json/nodes/{node}/tasks/{upid}/status", mockpve.HandleTaskStatus(state)).Methods("GET")
	r.HandleFunc("/api2/json/nodes/{node}/tasks/{upid}", mockpve.HandleStopTask(state)).Methods("DELETE")

	// Mock auth endpoint to avoid 401
	r.HandleFunc("/api2/json/access/ticket", func(w http.ResponseWriter, r *http.Request) {
		w.Header().Set("Content-Type", "application/json")
		_, _ = fmt.Fprint(w, `{"data": {"ticket": "dummy", "CSRFPreventionToken": "dummy"}}`)
	})

	server := httptest.NewServer(r)
	defer server.Close()

	// Setup Client
	config := &mockConfig{addr: server.URL}
	opts := api.ClientOption(func(c *api.ClientOptions) {
		c.Logger = &mockLogger{}
		c.Cache = &interfaces.NoOpCache{}
	})

	client, err := api.NewClient(config, opts)
	require.NoError(t, err)

	// Setup TaskManager
	var updated atomic.Bool
	notify := func() {
		updated.Store(true)
	}
	resolver := func(nodeName string) (*api.Client, error) {
		return client, nil
	}
	tm := NewTaskManager(resolver, notify)
	defer tm.Stop()

	// Create a dummy task that is already "running" on the server side
	// We use the state helper to create a task in the mock state
	upid := state.CreateTask("pve", "qmstart", "100", "root@pam")

	// Enqueue a task
	task := &Task{
		TargetVMID: 100,
		TargetNode: "pve",
		Type:       "Start",
		Operation: func() (string, error) {
			return upid, nil
		},
	}

	tm.Enqueue(task)

	// Wait for task to be picked up
	assert.Eventually(t, func() bool {
		t := tm.GetActiveTaskForVM("pve", 100)