488 lines
17 KiB
Go
488 lines
17 KiB
Go
//go:build integration
|
|
|
|
package main
|
|
|
|
import (
|
|
"context"
|
|
"crypto/sha256"
|
|
"crypto/x509"
|
|
"encoding/hex"
|
|
"encoding/json"
|
|
"encoding/pem"
|
|
"errors"
|
|
"fmt"
|
|
"io"
|
|
"io/fs"
|
|
"net"
|
|
"net/http"
|
|
"net/http/httptest"
|
|
"os"
|
|
"path/filepath"
|
|
"strings"
|
|
"sync/atomic"
|
|
"testing"
|
|
"time"
|
|
|
|
agentpb "git.ipao.vip/rogee/go-sip/gen/agent"
|
|
"git.ipao.vip/rogee/go-sip/internal/mq"
|
|
"git.ipao.vip/rogee/go-sip/internal/rpc"
|
|
"git.ipao.vip/rogee/go-sip/internal/tenant"
|
|
amqp "github.com/rabbitmq/amqp091-go"
|
|
"google.golang.org/grpc"
|
|
"google.golang.org/grpc/credentials"
|
|
)
|
|
|
|
func TestDispatcherCommandStartsWithIsolatedMQHTTPAndAgent(t *testing.T) {
|
|
brokerURL, adminURL := os.Getenv("RABBITMQ_URL"), os.Getenv("RABBITMQ_PROVISIONER_URL")
|
|
if brokerURL == "" || adminURL == "" {
|
|
t.Skip("requires the isolated RabbitMQ mock provisioner")
|
|
}
|
|
const dispatcherID = "c046b893-8628-4589-ae50-619d049248a6"
|
|
adminConnection, err := amqp.Dial(adminURL)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
defer adminConnection.Close()
|
|
admin, err := adminConnection.Channel()
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
defer admin.Close()
|
|
for _, exchange := range []string{mq.CommandsExchange, mq.ResultsExchange, mq.DeadLetterExchange} {
|
|
if err := admin.ExchangeDeclare(exchange, "topic", true, false, false, false, nil); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
}
|
|
control, err := tenant.ControlRoute(dispatcherID)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
result, err := tenant.ResultRoute(dispatcherID)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if _, err := admin.QueueDeclare(control.Queue, true, false, false, false, nil); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
defer func() { _, _ = admin.QueueDelete(control.Queue, false, false, false) }()
|
|
if err := admin.QueueBind(control.Queue, control.BindingKey, control.Exchange, false, nil); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if _, err := admin.QueueDeclare(result.Queue, true, false, false, false, nil); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
defer func() { _, _ = admin.QueueDelete(result.Queue, false, false, false) }()
|
|
if err := admin.QueueBind(result.Queue, result.BindingKey, result.Exchange, false, nil); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
defer func() { _ = admin.QueueUnbind(result.Queue, result.BindingKey, result.Exchange, nil) }()
|
|
taskRoute, err := tenant.TaskRoute(dispatcherID, "task-asr")
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if _, err := admin.QueueDeclare(taskRoute.Queue, true, false, false, false, nil); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
defer func() { _, _ = admin.QueueDelete(taskRoute.Queue, false, false, false) }()
|
|
if err := admin.QueueBind(taskRoute.Queue, taskRoute.BindingKey, taskRoute.Exchange, false, nil); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
for _, queue := range []string{control.Queue, result.Queue, taskRoute.Queue} {
|
|
if _, err := admin.QueuePurge(queue, false); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
}
|
|
|
|
root := t.TempDir()
|
|
reserved, err := net.Listen("tcp", "127.0.0.1:0")
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
listen := reserved.Addr().String()
|
|
_ = reserved.Close()
|
|
ca, agentCert, agentKey, dispatcherCert, dispatcherKey, dispatcherLeaf := localCommandCertificates(t)
|
|
var putCount, putBytes atomic.Int64
|
|
var putChecksum atomic.Value
|
|
oss := httptest.NewTLSServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
|
if r.Method != http.MethodPut || !strings.HasPrefix(r.URL.Path, "/mock-bucket/approved/") {
|
|
http.Error(w, "unexpected Mock OSS request", http.StatusBadRequest)
|
|
return
|
|
}
|
|
checksum := sha256.New()
|
|
size, err := io.Copy(checksum, r.Body)
|
|
if err != nil || size < 44 {
|
|
http.Error(w, "invalid Mock WAV upload", http.StatusBadRequest)
|
|
return
|
|
}
|
|
putBytes.Store(size)
|
|
putChecksum.Store(hex.EncodeToString(checksum.Sum(nil)))
|
|
putCount.Add(1)
|
|
w.WriteHeader(http.StatusOK)
|
|
}))
|
|
defer oss.Close()
|
|
ca = append(ca, pem.EncodeToMemory(&pem.Block{Type: "CERTIFICATE", Bytes: oss.Certificate().Raw})...)
|
|
agentBlock, _ := pem.Decode(agentCert)
|
|
if agentBlock == nil {
|
|
t.Fatal("isolated Agent certificate is invalid")
|
|
}
|
|
agentLeaf, err := x509.ParseCertificate(agentBlock.Bytes)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
agentTLS, err := rpc.NewServerTLSConfig(ca, agentCert, agentKey)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
for filename, content := range map[string][]byte{
|
|
filepath.Join(root, "ca.pem"): ca,
|
|
filepath.Join(root, "agent.pem"): agentCert,
|
|
filepath.Join(root, "agent.key"): agentKey,
|
|
} {
|
|
if err := os.WriteFile(filename, content, 0600); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
}
|
|
settings, scenario := currentAgentSetupFixture(t)
|
|
settings.DispatcherID = dispatcherID
|
|
settings.DispatcherEndpoint = listen
|
|
settings.CAFile = filepath.Join(root, "ca.pem")
|
|
settings.CertFile = filepath.Join(root, "agent.pem")
|
|
settings.KeyFile = filepath.Join(root, "agent.key")
|
|
settings.SessionPath = filepath.Join(root, "agent-session.json")
|
|
settings.RecoveryRoot = filepath.Join(root, "agent-recovery")
|
|
settings.PeerFingerprints = map[string]struct{}{rpc.CertificateFingerprint(dispatcherLeaf): {}}
|
|
if err := os.MkdirAll(settings.RecoveryRoot, 0700); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
dispatcherTLS, err := rpc.NewClientTLSConfig(ca, agentCert, agentKey, settings.DispatcherServerName)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
agentConnection, err := grpc.NewClient(listen, grpc.WithTransportCredentials(credentials.NewTLS(dispatcherTLS)))
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
defer agentConnection.Close()
|
|
agentContext, stopAgent := context.WithCancel(context.Background())
|
|
defer stopAgent()
|
|
agentServer, err := newAgentServer(agentContext, settings, scenario, map[string]int64{"trunk-mock": 8}, agentpb.NewAgentControlServiceClient(agentConnection))
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
agentListener, err := net.Listen("tcp", "127.0.0.1:0")
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
agentGRPC := grpc.NewServer(grpc.Creds(credentials.NewTLS(agentTLS)))
|
|
agentpb.RegisterAgentControlServiceServer(agentGRPC, agentServer)
|
|
go func() { _ = agentGRPC.Serve(agentListener) }()
|
|
defer agentGRPC.Stop()
|
|
|
|
examples := map[string][]byte{}
|
|
for _, name := range []string{"config-read-sip.json", "task-discovery-page.json", "task-discovery-end.json", "config-read-task-asr.json", "config-read-providers.json", "config-read-quota.json"} {
|
|
body, err := os.ReadFile(filepath.Join("..", "..", "contracts", "local", "examples", name))
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
examples[name] = body
|
|
}
|
|
weekly := map[string]any{}
|
|
for _, day := range []string{"monday", "tuesday", "wednesday", "thursday", "friday", "saturday", "sunday"} {
|
|
weekly[day] = []map[string]string{{"start": "00:00", "end": "24:00"}}
|
|
}
|
|
var sipConfig map[string]any
|
|
if err := json.Unmarshal(examples["config-read-sip.json"], &sipConfig); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
trunk := sipConfig["trunks"].([]any)[0].(map[string]any)
|
|
trunk["transport"] = "udp"
|
|
trunk["auth_mode"] = "ip"
|
|
trunk["registration_required"] = false
|
|
trunk["max_concurrent_calls"] = 3
|
|
trunk["schedule"].(map[string]any)["weekly_windows"] = weekly
|
|
if examples["config-read-sip.json"], err = json.Marshal(sipConfig); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
var taskConfig map[string]any
|
|
if err := json.Unmarshal(examples["config-read-task-asr.json"], &taskConfig); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
taskSchedule := taskConfig["schedule"].(map[string]any)
|
|
taskSchedule["starts_at"] = "2000-01-01T00:00:00+08:00"
|
|
taskSchedule["weekly_windows"] = weekly
|
|
if examples["config-read-task-asr.json"], err = json.Marshal(taskConfig); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
var sipReads, discoveryReads, taskReads, providerReads, quotaReads atomic.Int64
|
|
saas := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
|
if r.Header.Get("X-DISPATCHER-id") != dispatcherID || r.Header.Get("X-DISPATCHER-SECRET-KEY") != "isolated-secret" {
|
|
http.Error(w, "unapproved Dispatcher", http.StatusForbidden)
|
|
return
|
|
}
|
|
w.Header().Set("Content-Type", "application/json")
|
|
switch r.URL.Path {
|
|
case "/internal/v1/dispatcher/sip":
|
|
sipReads.Add(1)
|
|
_, _ = w.Write(examples["config-read-sip.json"])
|
|
case "/internal/v1/dispatcher/tasks":
|
|
discoveryReads.Add(1)
|
|
switch r.URL.Query().Get("after") {
|
|
case "":
|
|
_, _ = w.Write(examples["task-discovery-page.json"])
|
|
case "opaque-page-token-1", "opaque-end-token":
|
|
_, _ = w.Write(examples["task-discovery-end.json"])
|
|
default:
|
|
http.Error(w, "unexpected Mock cursor", http.StatusBadRequest)
|
|
}
|
|
case "/internal/v1/dispatcher/task/task-asr":
|
|
taskReads.Add(1)
|
|
_, _ = w.Write(examples["config-read-task-asr.json"])
|
|
case "/internal/v1/dispatcher/ai-providers":
|
|
providerReads.Add(1)
|
|
_, _ = w.Write(examples["config-read-providers.json"])
|
|
case "/internal/v1/dispatcher/tenant/1001/quota":
|
|
quotaReads.Add(1)
|
|
_, _ = w.Write(examples["config-read-quota.json"])
|
|
default:
|
|
http.Error(w, "unexpected Mock read", http.StatusNotFound)
|
|
}
|
|
}))
|
|
defer saas.Close()
|
|
|
|
agentInventory := filepath.Join(root, "agent-endpoints.json")
|
|
ossFile := filepath.Join(root, "oss.json")
|
|
for filename, body := range map[string][]byte{
|
|
agentInventory: []byte(fmt.Sprintf(`[{"agent_id":"agent-mock","cell_id":"cell-mock","address":%q,"server_name":"agent.local"}]`, agentListener.Addr().String())),
|
|
ossFile: []byte(fmt.Sprintf(`{"dispatcher_id":%q,"oss":{"endpoint":%q,"region":"cn-mock","bucket":"mock-bucket","object_prefix":"approved","access_key_id_env":"MOCK_OSS_KEY_ID","access_key_secret_env":"MOCK_OSS_KEY_SECRET","max_asset_bytes":1048576}}`, dispatcherID, oss.URL)),
|
|
filepath.Join(root, "ca.pem"): ca,
|
|
filepath.Join(root, "dispatcher.pem"): dispatcherCert,
|
|
filepath.Join(root, "dispatcher.key"): dispatcherKey,
|
|
} {
|
|
if err := os.WriteFile(filename, body, 0600); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
}
|
|
database := filepath.Join(root, "dispatcher.sqlite")
|
|
for name, value := range map[string]string{
|
|
"DISPATCHER_ID": dispatcherID, "DISPATCHER_SECRET_KEY": "isolated-secret",
|
|
"SAAS_BASE_URL": saas.URL, "RABBITMQ_URL": brokerURL,
|
|
"DISPATCHER_SQLITE_PATH": database, "DISPATCHER_GRPC_LISTEN": listen,
|
|
"DISPATCHER_AGENT_ENDPOINTS_FILE": agentInventory, "DISPATCHER_OSS_CONFIG_FILE": ossFile,
|
|
"MTLS_CA_FILE": filepath.Join(root, "ca.pem"),
|
|
"MTLS_CERT_FILE": filepath.Join(root, "dispatcher.pem"),
|
|
"MTLS_KEY_FILE": filepath.Join(root, "dispatcher.key"),
|
|
"MTLS_PEER_CERT_FINGERPRINTS": rpc.CertificateFingerprint(agentLeaf),
|
|
"MOCK_OSS_KEY_ID": "isolated-id", "MOCK_OSS_KEY_SECRET": "isolated-secret",
|
|
} {
|
|
t.Setenv(name, value)
|
|
}
|
|
process, cancel := context.WithCancel(context.Background())
|
|
command := newRootCommand()
|
|
command.SetContext(process)
|
|
command.SetOut(io.Discard)
|
|
command.SetErr(io.Discard)
|
|
command.SetArgs([]string{"dispatcher", "--mode", "mock"})
|
|
finished := make(chan error, 1)
|
|
go func() { finished <- command.Execute() }()
|
|
deadline := time.After(8 * time.Second)
|
|
for sipReads.Load() < 1 || discoveryReads.Load() < 2 || taskReads.Load() == 0 || providerReads.Load() == 0 || quotaReads.Load() == 0 {
|
|
select {
|
|
case err := <-finished:
|
|
cancel()
|
|
t.Fatalf("isolated Dispatcher exited before the five HTTP resources were read: %v", err)
|
|
case <-deadline:
|
|
cancel()
|
|
t.Fatal("isolated Dispatcher did not read SIP, discovery, task, providers, and tenant quota")
|
|
case <-time.After(20 * time.Millisecond):
|
|
}
|
|
}
|
|
for {
|
|
queue, err := admin.QueueInspect(taskRoute.Queue)
|
|
if err != nil {
|
|
cancel()
|
|
t.Fatalf("assigned task queue cannot be inspected: %v", err)
|
|
}
|
|
if queue.Consumers > 0 {
|
|
break
|
|
}
|
|
select {
|
|
case err := <-finished:
|
|
cancel()
|
|
t.Fatalf("isolated Dispatcher exited before consuming its task queue: %v", err)
|
|
case <-deadline:
|
|
cancel()
|
|
t.Fatal("isolated Dispatcher did not consume its SaaS-provisioned task queue")
|
|
case <-time.After(20 * time.Millisecond):
|
|
}
|
|
}
|
|
if _, err := os.Stat(database); err != nil {
|
|
cancel()
|
|
t.Fatalf("successful isolated bootstrap did not retain SQLite: %v", err)
|
|
}
|
|
select {
|
|
case err := <-finished:
|
|
cancel()
|
|
t.Fatalf("isolated Dispatcher stopped despite a live process: %v", err)
|
|
default:
|
|
}
|
|
commandBody, err := os.ReadFile(filepath.Join("..", "..", "contracts", "local", "examples", "mq-execute.json"))
|
|
if err != nil {
|
|
cancel()
|
|
t.Fatal(err)
|
|
}
|
|
if err := admin.PublishWithContext(process, taskRoute.Exchange, taskRoute.BindingKey, true, false, amqp.Publishing{ContentType: "application/json", DeliveryMode: amqp.Persistent, Body: commandBody}); err != nil {
|
|
cancel()
|
|
t.Fatal(err)
|
|
}
|
|
resultDeadline := time.After(12 * time.Second)
|
|
for {
|
|
message, ok, err := admin.Get(result.Queue, false)
|
|
if err != nil {
|
|
cancel()
|
|
t.Fatalf("cannot read isolated result queue: %v", err)
|
|
}
|
|
if ok {
|
|
var event struct {
|
|
EventType string `json:"event_type"`
|
|
Payload struct {
|
|
Outcome string `json:"outcome"`
|
|
Recording struct {
|
|
Status string `json:"status"`
|
|
Bucket string `json:"bucket"`
|
|
Format string `json:"format"`
|
|
SizeBytes int64 `json:"size_bytes"`
|
|
ChecksumSHA256 string `json:"checksum_sha256"`
|
|
} `json:"recording"`
|
|
} `json:"payload"`
|
|
}
|
|
if err := json.Unmarshal(message.Body, &event); err != nil {
|
|
cancel()
|
|
t.Fatalf("isolated result is invalid JSON: %v", err)
|
|
}
|
|
if err := message.Ack(false); err != nil {
|
|
cancel()
|
|
t.Fatalf("isolated result cannot be ACKed: %v", err)
|
|
}
|
|
if event.EventType == "call.execute.result" {
|
|
if event.Payload.Outcome != "answered" || event.Payload.Recording.Status != "uploaded" || event.Payload.Recording.Bucket != "mock-bucket" {
|
|
cancel()
|
|
t.Fatalf("isolated call did not produce an uploaded result: outcome=%q recording=%q bucket=%q", event.Payload.Outcome, event.Payload.Recording.Status, event.Payload.Recording.Bucket)
|
|
}
|
|
if event.Payload.Recording.Format != "wav" || event.Payload.Recording.SizeBytes != putBytes.Load() || event.Payload.Recording.ChecksumSHA256 != putChecksum.Load().(string) {
|
|
cancel()
|
|
t.Fatalf("isolated result WAV metadata does not match the single Mock PUT: size=%d format=%q", event.Payload.Recording.SizeBytes, event.Payload.Recording.Format)
|
|
}
|
|
break
|
|
}
|
|
}
|
|
select {
|
|
case err := <-finished:
|
|
cancel()
|
|
t.Fatalf("isolated Dispatcher exited before the final result: %v", err)
|
|
case <-resultDeadline:
|
|
cancel()
|
|
t.Fatal("isolated call did not produce a final result")
|
|
case <-time.After(25 * time.Millisecond):
|
|
}
|
|
}
|
|
if putCount.Load() != 1 {
|
|
cancel()
|
|
t.Fatalf("isolated call uploaded %d times; expected exactly one PUT", putCount.Load())
|
|
}
|
|
if err := filepath.WalkDir(settings.RecoveryRoot, func(_ string, entry fs.DirEntry, walkErr error) error {
|
|
if walkErr != nil {
|
|
return walkErr
|
|
}
|
|
if !entry.IsDir() {
|
|
return fmt.Errorf("successful upload persisted recovery file %s", entry.Name())
|
|
}
|
|
return nil
|
|
}); err != nil {
|
|
cancel()
|
|
t.Fatal(err)
|
|
}
|
|
cancel()
|
|
select {
|
|
case err := <-finished:
|
|
if err != nil && !errors.Is(err, context.Canceled) {
|
|
t.Fatalf("isolated Dispatcher shutdown failed: %v", err)
|
|
}
|
|
case <-time.After(5 * time.Second):
|
|
t.Fatal("isolated Dispatcher did not stop after cancellation")
|
|
}
|
|
|
|
previousSIP, previousDiscovery := sipReads.Load(), discoveryReads.Load()
|
|
restartContext, restartCancel := context.WithCancel(context.Background())
|
|
defer restartCancel()
|
|
restartCommand := newRootCommand()
|
|
restartCommand.SetContext(restartContext)
|
|
restartCommand.SetOut(io.Discard)
|
|
restartCommand.SetErr(io.Discard)
|
|
restartCommand.SetArgs([]string{"dispatcher", "--mode", "mock"})
|
|
restarted := make(chan error, 1)
|
|
go func() { restarted <- restartCommand.Execute() }()
|
|
restartDeadline := time.After(8 * time.Second)
|
|
for sipReads.Load() <= previousSIP || discoveryReads.Load() <= previousDiscovery {
|
|
select {
|
|
case err := <-restarted:
|
|
t.Fatalf("isolated Dispatcher could not reopen its original SQLite and session: %v", err)
|
|
case <-restartDeadline:
|
|
t.Fatal("isolated Dispatcher did not restart its approved HTTP snapshot")
|
|
case <-time.After(20 * time.Millisecond):
|
|
}
|
|
}
|
|
for {
|
|
queue, err := admin.QueueInspect(taskRoute.Queue)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if queue.Consumers > 0 {
|
|
break
|
|
}
|
|
select {
|
|
case err := <-restarted:
|
|
t.Fatalf("isolated Dispatcher stopped before restoring its task queue: %v", err)
|
|
case <-restartDeadline:
|
|
t.Fatal("isolated Dispatcher did not restore its task consumer")
|
|
case <-time.After(20 * time.Millisecond):
|
|
}
|
|
}
|
|
if err := admin.PublishWithContext(restartContext, taskRoute.Exchange, taskRoute.BindingKey, true, false, amqp.Publishing{ContentType: "application/json", DeliveryMode: amqp.Persistent, Body: commandBody}); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
for {
|
|
queue, err := admin.QueueInspect(taskRoute.Queue)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if queue.Messages == 0 {
|
|
break
|
|
}
|
|
select {
|
|
case err := <-restarted:
|
|
t.Fatalf("isolated Dispatcher stopped before acknowledging the duplicate call: %v", err)
|
|
case <-time.After(3 * time.Second):
|
|
t.Fatal("duplicate call was not consumed after restart")
|
|
case <-time.After(20 * time.Millisecond):
|
|
}
|
|
}
|
|
time.Sleep(250 * time.Millisecond)
|
|
if putCount.Load() != 1 {
|
|
t.Fatalf("restart repeated an already uploaded call: PUT count=%d", putCount.Load())
|
|
}
|
|
select {
|
|
case err := <-restarted:
|
|
t.Fatalf("isolated Dispatcher stopped after duplicate delivery: %v", err)
|
|
default:
|
|
}
|
|
restartCancel()
|
|
select {
|
|
case err := <-restarted:
|
|
if err != nil && !errors.Is(err, context.Canceled) {
|
|
t.Fatalf("restarted Dispatcher shutdown failed: %v", err)
|
|
}
|
|
case <-time.After(5 * time.Second):
|
|
t.Fatal("restarted Dispatcher did not stop after cancellation")
|
|
}
|
|
}
|