86 lines
2.7 KiB
Go
86 lines
2.7 KiB
Go
package store
|
|
|
|
import (
|
|
"testing"
|
|
"time"
|
|
|
|
"git.ipao.vip/rogee/go-sip/contracts"
|
|
agentv1 "git.ipao.vip/rogee/go-sip/gen/agent/v1"
|
|
"git.ipao.vip/rogee/go-sip/internal/contract"
|
|
"git.ipao.vip/rogee/go-sip/internal/tenant"
|
|
)
|
|
|
|
func TestControlAcknowledgementCommitsFinalReceiptOnce(t *testing.T) {
|
|
s, err := Open(":memory:")
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
defer s.Close()
|
|
if err := s.BindDispatcherID(identityA); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
s.now = func() time.Time { return time.Date(2026, 9, 21, 0, 0, 1, 0, time.UTC) }
|
|
base := "upstream/" + contract.MQSourceCommit + "/examples/"
|
|
raw, err := contracts.Files.ReadFile(base + "call-execute.json")
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
envelope, payload, err := contract.DecodeExecute(raw)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
route, _ := tenant.NewDispatcherRoute(identityA, envelope.TenantKey)
|
|
if _, err := s.IngestCommand(raw, route.InboundKey); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if err := s.SetQuota("global", 1); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if err := s.Reserve("reservation-a", payload.ExecutionID, envelope.TenantKey, []string{"global"}); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
binding := &agentv1.ExecutionBinding{TenantId: envelope.TenantID, TenantKey: envelope.TenantKey, TaskId: payload.TaskID, TaskItemId: payload.TaskItemID, TaskRevision: payload.TaskRevision, ExecutionId: payload.ExecutionID, AgentVersionId: payload.AgentVersionID, RoutePolicyId: payload.RoutePolicyID, CallerProfileId: payload.CallerProfileID}
|
|
if err := s.BindExecutionAgent("agent-a", binding); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
control, err := contracts.Files.ReadFile(base + "task-control.json")
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
original, _, err := s.HandleTaskControl(control, route.InboundKey)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
targets, err := s.PendingTaskControls(10)
|
|
if err != nil || len(targets) != 1 {
|
|
t.Fatalf("targets: %v %v", targets, err)
|
|
}
|
|
if err := s.CompleteTaskControl(targets[0], 1); err == nil {
|
|
t.Fatal("unapplied revision accepted")
|
|
}
|
|
if err := s.CompleteTaskControl(targets[0], 2); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if err := s.CompleteTaskControl(targets[0], 2); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
final, duplicate, err := s.HandleTaskControl(control, route.InboundKey)
|
|
if err != nil || !duplicate || final == original {
|
|
t.Fatalf("final receipt missing: %v", err)
|
|
}
|
|
pending, err := s.PendingTaskControls(10)
|
|
if err != nil || len(pending) != 0 {
|
|
t.Fatal("applied control remained pending")
|
|
}
|
|
var revision, count int
|
|
if err := s.DB().QueryRow(`SELECT task_revision FROM tasks`).Scan(&revision); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if err := s.DB().QueryRow(`SELECT COUNT(*) FROM outbox`).Scan(&count); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if revision != 2 || count != 3 {
|
|
t.Fatalf("revision=%d outbox=%d", revision, count)
|
|
}
|
|
}
|