go-sip/internal/store/current_discovery_test.go

package store

import (
	"path/filepath"
	"testing"

	"git.ipao.vip/rogee/go-sip/internal/configread"
)

func currentDiscoveredTask(status string, revision int64) configread.CurrentDiscoveredTask {
	return configread.CurrentDiscoveredTask{TaskID: "task-asr", TenantID: 1001, TaskRevision: revision, Status: status}
}

func TestCurrentHTTPStopCannotBeUndoneByStaleRunningAndSuppressesInbox(t *testing.T) {
	s, err := OpenCurrent(filepath.Join(t.TempDir(), "state.db"))
	if err != nil {
		t.Fatal(err)
	}
	defer s.Close()
	base := currentStoreSnapshot(t)
	if err := s.ApplyDiscoverySnapshot(currentDispatcherID, []configread.CurrentDiscoveredTask{currentDiscoveredTask("running", 1)}); err != nil {
		t.Fatal(err)
	}
	if err := s.SaveSnapshot(base); err != nil {
		t.Fatal(err)
	}
	if err := s.MarkReadyForSIP(currentDispatcherID, 8); err != nil {
		t.Fatal(err)
	}
	if _, _, err := s.RecordExecute(currentCall("not-admitted-before-http-stop")); err != nil {
		t.Fatal(err)
	}
	if err := s.ApplyDiscoverySnapshot(currentDispatcherID, []configread.CurrentDiscoveredTask{currentDiscoveredTask("stopped", 1)}); err != nil {
		t.Fatal(err)
	}
	if pending, err := s.ListPendingExecute(currentDispatcherID); err != nil || len(pending) != 0 {
		t.Fatalf("HTTP stop retained pending command: %+v %v", pending, err)
	}
	if err := s.ApplyDiscoverySnapshot(currentDispatcherID, []configread.CurrentDiscoveredTask{currentDiscoveredTask("running", 1)}); err != nil {
		t.Fatal(err)
	}
	if err := s.MarkReadyForSIP(currentDispatcherID, 8); err != nil {
		t.Fatal(err)
	}
	if admitted, err := s.CanAdmit(currentDispatcherID, 1001, "task-asr"); err != nil || admitted {
		t.Fatalf("stale HTTP running undid irrevocable stop: %v %v", admitted, err)
	}
	if err := s.PrepareControl(currentDispatcherID, 1001, "task-asr", "resume"); err == nil {
		t.Fatal("stopped task accepted resume")
	}
}

func TestCurrentHTTPPausedRequiresExplicitMQResume(t *testing.T) {
	s, err := OpenCurrent(filepath.Join(t.TempDir(), "state.db"))
	if err != nil {
		t.Fatal(err)
	}
	defer s.Close()
	base := currentStoreSnapshot(t)
	if err := s.ApplyDiscoverySnapshot(currentDispatcherID, []configread.CurrentDiscoveredTask{currentDiscoveredTask("running", 1)}); err != nil {
		t.Fatal(err)
	}
	if err := s.SaveSnapshot(base); err != nil {
		t.Fatal(err)
	}
	if err := s.ApplyDiscoverySnapshot(currentDispatcherID, []configread.CurrentDiscoveredTask{currentDiscoveredTask("paused", 1)}); err != nil {
		t.Fatal(err)
	}
	if err := s.ApplyDiscoverySnapshot(currentDispatcherID, []configread.CurrentDiscoveredTask{currentDiscoveredTask("running", 1)}); err != nil {
		t.Fatal(err)
	}
	if err := s.MarkReadyForSIP(currentDispatcherID, 8); err != nil {
		t.Fatal(err)
	}
	if admitted, err := s.CanAdmit(currentDispatcherID, 1001, "task-asr"); err != nil || admitted {
		t.Fatalf("HTTP running overrode persisted pause: %v %v", admitted, err)
	}
	if err := s.PrepareControl(currentDispatcherID, 1001, "task-asr", "resume"); err != nil {
		t.Fatal(err)
	}
	if err := s.CompleteControl(currentDispatcherID, 1001, "task-asr", "resume", "resume-http-pause"); err != nil {
		t.Fatal(err)
	}
	if admitted, err := s.CanAdmit(currentDispatcherID, 1001, "task-asr"); err != nil || !admitted {
		t.Fatalf("explicit resume failed: %v %v", admitted, err)
	}
}

func TestCurrentDiscoveryPageCommitsAtomicallyAndPreservesControl(t *testing.T) {
	s, err := OpenCurrent(filepath.Join(t.TempDir(), "state.db"))
	if err != nil {
		t.Fatal(err)
	}
	defer s.Close()
	if err := s.ApplyDiscoverySnapshot(currentDispatcherID, []configread.CurrentDiscoveredTask{currentDiscoveredTask("running", 1)}); err != nil {
		t.Fatal(err)
	}
	if err := s.ApplyControl(currentDispatcherID, 1001, "task-asr", "pause"); err != nil {
		t.Fatal(err)
	}
	tasks := []configread.CurrentDiscoveredTask{currentDiscoveredTask("running", 2), {TaskID: "task-second", TenantID: 0, TaskRevision: 1, Status: "running"}}
	if err := s.ApplyDiscoveryPage(currentDispatcherID, tasks); err == nil {
		t.Fatal("invalid page was partly committed")
	}
	var revision int64
	if err := s.db.QueryRow(`SELECT task_revision FROM dispatcher_tasks WHERE task_id='task-asr'`).Scan(&revision); err != nil || revision != 1 {
		t.Fatalf("page committed despite later error: revision=%d err=%v", revision, err)
	}
	tasks[1].TenantID = 1001
	if err := s.ApplyDiscoveryPage(currentDispatcherID, tasks); err != nil {
		t.Fatal(err)
	}
	if err := s.db.QueryRow(`SELECT task_revision FROM dispatcher_tasks WHERE task_id='task-asr'`).Scan(&revision); err != nil || revision != 2 {
		t.Fatalf("valid page not committed: revision=%d err=%v", revision, err)
	}
	var state string
	if err := s.db.QueryRow(`SELECT control_state FROM dispatcher_tasks WHERE task_id='task-asr'`).Scan(&state); err != nil || state != "paused" {
		t.Fatalf("HTTP page overrode MQ pause: %s %v", state, err)
	}
	var count int
	if err := s.db.QueryRow(`SELECT COUNT(*) FROM dispatcher_tasks WHERE task_id='task-second' AND present=1`).Scan(&count); err != nil || count != 1 {
		t.Fatalf("second task of valid page absent: %d %v", count, err)
	}
	assigned, err := s.ListAssignedTasks(currentDispatcherID)
	if err != nil || len(assigned) != 2 || assigned[0].TaskID != "task-asr" || assigned[0].ControlState != "paused" || assigned[1].TaskID != "task-second" {
		t.Fatalf("assigned task queues not discoverable: %+v %v", assigned, err)
	}
}