75 lines
3.9 KiB
Go
75 lines
3.9 KiB
Go
package rpc
|
|
|
|
import (
|
|
"context"
|
|
"os"
|
|
"path/filepath"
|
|
"testing"
|
|
"time"
|
|
|
|
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/testfixture"
|
|
"google.golang.org/protobuf/proto"
|
|
)
|
|
|
|
func TestControlReceiptSurvivesRestartAndNewSession(t *testing.T) {
|
|
now := time.Date(2026, 9, 18, 1, 0, 0, 0, time.UTC)
|
|
path := filepath.Join(t.TempDir(), "session.json")
|
|
start := func(generation uint64) *Server {
|
|
bootID := "boot-1"
|
|
if generation > 1 {
|
|
bootID = "boot-2"
|
|
}
|
|
s := NewServer(ServerOptions{Mode: "mock", Now: func() time.Time { return now }, StatePath: path, Status: &agentv1.AgentStatus{AgentId: "agent-1", CellId: "cell-1", BootId: bootID}})
|
|
activationMeta := testMeta("activate", "", 0)
|
|
activationMeta.BootId = bootID
|
|
_, err := s.ActivateAgent(context.Background(), &agentv1.ActivateAgentRequest{Meta: activationMeta, Binding: &agentv1.AgentBinding{AgentId: "agent-1", CellId: "cell-1", ExpectedBootId: bootID, DispatcherEpoch: "epoch-1", SessionGeneration: generation}, ActivationOperationId: "activate"})
|
|
require.NoError(t, err)
|
|
return s
|
|
}
|
|
s := start(1)
|
|
raw, err := testfixture.Execute()
|
|
require.NoError(t, err)
|
|
envelope, payload, err := contract.DecodeExecute(raw)
|
|
require.NoError(t, err)
|
|
binding := &agentv1.ExecutionBinding{TenantId: envelope.TenantID, TenantKey: envelope.TenantKey, ExecutionId: payload.ExecutionID, TaskId: payload.TaskID, TaskItemId: payload.TaskItemID, TaskRevision: payload.TaskRevision, AgentVersionId: payload.AgentVersionID}
|
|
_, err = s.Execute(context.Background(), &agentv1.ExecuteRequest{Meta: testMeta("execute", "execute-key", 1), Binding: binding, CallExecuteJson: raw})
|
|
require.NoError(t, err)
|
|
request := &agentv1.ApplyTaskControlRequest{Meta: testMeta("pause", "pause-key", 1), Binding: binding, Action: agentv1.ControlAction_CONTROL_ACTION_PAUSE, ActiveCallPolicy: agentv1.ActiveCallPolicy_ACTIVE_CALL_POLICY_DRAIN, ExpectedTaskRevision: binding.TaskRevision}
|
|
applied, err := s.ApplyTaskControl(context.Background(), request)
|
|
require.NoError(t, err)
|
|
require.Equal(t, agentv1.ResultCode_RESULT_CODE_APPLIED, applied.Receipt.Result)
|
|
recovered := start(2)
|
|
request.Meta.SessionGeneration = 2
|
|
request.Meta.BootId = "boot-2"
|
|
freshMeta := func(operation, key string) *agentv1.RequestMeta {
|
|
meta := testMeta(operation, key, 2)
|
|
meta.BootId = "boot-2"
|
|
return meta
|
|
}
|
|
replayed, err := recovered.ApplyTaskControl(context.Background(), request)
|
|
require.NoError(t, err)
|
|
if !proto.Equal(applied, replayed) {
|
|
t.Fatal("lost original control receipt after restart")
|
|
}
|
|
snapshot, err := recovered.QueryExecution(context.Background(), &agentv1.QueryExecutionRequest{Meta: freshMeta("query", "query-key"), Binding: binding})
|
|
require.NoError(t, err)
|
|
if snapshot.Snapshot == nil || !snapshot.Snapshot.Unknown || snapshot.Snapshot.Binding.TaskRevision != applied.AppliedTaskRevision {
|
|
t.Fatal("recovered active work must remain unknown with applied control revision")
|
|
}
|
|
retry, err := recovered.Execute(context.Background(), &agentv1.ExecuteRequest{Meta: freshMeta("new-execute", "new-key"), Binding: binding, CallExecuteJson: raw})
|
|
require.NoError(t, err)
|
|
if retry.Receipt.Result == agentv1.ResultCode_RESULT_CODE_ACCEPTED {
|
|
t.Fatal("restarted execution was prepared again")
|
|
}
|
|
}
|
|
|
|
func TestCorruptExecutionJournalPreventsActivation(t *testing.T) {
|
|
path := filepath.Join(t.TempDir(), "session.json")
|
|
require.NoError(t, os.WriteFile(path+".executions", []byte(`{"version":999}`), 0600))
|
|
s := NewServer(ServerOptions{Mode: "mock", StatePath: path, Status: &agentv1.AgentStatus{AgentId: "agent-1", CellId: "cell-1", BootId: "boot-1"}})
|
|
_, err := s.ActivateAgent(context.Background(), &agentv1.ActivateAgentRequest{Meta: testMeta("activate", "", 0), Binding: &agentv1.AgentBinding{AgentId: "agent-1", CellId: "cell-1", ExpectedBootId: "boot-1", DispatcherEpoch: "epoch-1", SessionGeneration: 1}, ActivationOperationId: "activate"})
|
|
require.Error(t, err)
|
|
}
|