story/internal/mockprovider/server_test.go

package mockprovider

import (
	"bytes"
	"io"
	"log/slog"
	"net/http"
	"net/http/httptest"
	"strings"
	"sync/atomic"
	"testing"
	"time"

	"data-terminal-cloud/internal/voiceprovider"
)

func TestStartIsIdempotentAndEmitsDuplicateSignedCallbacks(t *testing.T) {
	secret := "01234567890123456789012345678901"
	var callbacks atomic.Int32
	var stoppedCallbacks atomic.Int32
	var transcriptCallbacks atomic.Int32
	callbackServer := httptest.NewServer(http.HandlerFunc(func(writer http.ResponseWriter, request *http.Request) {
		body, _ := io.ReadAll(request.Body)
		if err := voiceprovider.VerifyCallback([]byte(secret), request.Header.Get(voiceprovider.HeaderCallbackTimestamp), request.Header.Get(voiceprovider.HeaderCallbackSignature), body, time.Now(), time.Minute); err != nil {
			t.Errorf("callback signature: %v", err)
		}
		callbacks.Add(1)
		if bytes.Contains(body, []byte(`"voice_chat.stopped"`)) {
			stoppedCallbacks.Add(1)
		}
		if bytes.Contains(body, []byte(`"voice_chat.transcript.final"`)) {
			transcriptCallbacks.Add(1)
		}
		writer.WriteHeader(http.StatusNoContent)
	}))
	defer callbackServer.Close()
	server, err := New(Options{
		APIKey: "mock-provider-key", CallbackURL: callbackServer.URL, CallbackSecret: secret,
		StartDelay: time.Millisecond, DuplicateCallbacks: 2,
	}, slog.Default())
	if err != nil {
		t.Fatal(err)
	}
	httpServer := httptest.NewServer(server.Handler())
	defer httpServer.Close()
	client, _ := voiceprovider.NewHTTPClient(httpServer.URL, "mock-provider-key", time.Second)
	input := voiceprovider.StartRequest{TaskRef: "task-1", RoomID: "room-1", DeviceUserID: "device-1", AgentUserID: "agent-1", ConversationID: "conversation-1", EngineMode: "s2s_hybrid"}
	for index := 0; index < 2; index++ {
		response, err := client.Start(t.Context(), input)
		if err != nil || response.TaskRef != input.TaskRef {
			t.Fatalf("start %d response=%#v err=%v", index, response, err)
		}
	}
	deadline := time.Now().Add(time.Second)
	for callbacks.Load() < 2 && time.Now().Before(deadline) {
		time.Sleep(time.Millisecond)
	}
	if callbacks.Load() != 2 {
		t.Fatalf("callbacks = %d", callbacks.Load())
	}
	stopInput := voiceprovider.StopRequest{TaskRef: input.TaskRef, ConversationID: input.ConversationID, Reason: "handset_on_hook"}
	for index := 0; index < 2; index++ {
		response, err := client.Stop(t.Context(), stopInput)
		if err != nil || response.TaskRef != input.TaskRef {
			t.Fatalf("stop %d response=%#v err=%v", index, response, err)
		}
	}
	deadline = time.Now().Add(time.Second)
	for stoppedCallbacks.Load() < 2 && time.Now().Before(deadline) {
		time.Sleep(time.Millisecond)
	}
	if stoppedCallbacks.Load() != 2 || transcriptCallbacks.Load() != 2 || callbacks.Load() != 6 {
		t.Fatalf("callbacks=%d transcript=%d stopped=%d", callbacks.Load(), transcriptCallbacks.Load(), stoppedCallbacks.Load())
	}
	request, _ := http.NewRequest(http.MethodPost, httpServer.URL+"/mock/v1/voice-chats:start", strings.NewReader(`{}`))
	response, err := http.DefaultClient.Do(request)
	if err != nil {
		t.Fatal(err)
	}
	response.Body.Close()
	if response.StatusCode != http.StatusUnauthorized {
		t.Fatalf("unauthorized status = %d", response.StatusCode)
	}
	availability, _ := http.NewRequest(http.MethodPost, httpServer.URL+"/mock/v1/admin/availability", strings.NewReader(`{"available":false}`))
	availability.Header.Set("Content-Type", "application/json")
	availability.Header.Set("X-Mock-Provider-Key", "mock-provider-key")
	response, err = http.DefaultClient.Do(availability)
	if err != nil {
		t.Fatal(err)
	}
	response.Body.Close()
	if response.StatusCode != http.StatusOK {
		t.Fatalf("availability status = %d", response.StatusCode)
	}
	_, err = client.Start(t.Context(), voiceprovider.StartRequest{TaskRef: "task-unavailable", RoomID: "room-2", DeviceUserID: "device-2", AgentUserID: "agent-2", ConversationID: "conversation-2", EngineMode: "cascade"})
	if !voiceprovider.IsTemporary(err) {
		t.Fatalf("unavailable error = %v", err)
	}
	stateRequest, _ := http.NewRequest(http.MethodGet, httpServer.URL+"/mock/v1/admin/state", nil)
	stateRequest.Header.Set("X-Mock-Provider-Key", "mock-provider-key")
	response, err = http.DefaultClient.Do(stateRequest)
	if err != nil {
		t.Fatal(err)
	}
	stateBody, _ := io.ReadAll(response.Body)
	response.Body.Close()
	if response.StatusCode != http.StatusOK || !bytes.Contains(stateBody, []byte(`"available":false`)) || !bytes.Contains(stateBody, []byte(`"start_calls":3`)) {
		t.Fatalf("state status=%d body=%s", response.StatusCode, stateBody)
	}
}

func TestServerBuildsAllDeidentifiedReplaySamples(t *testing.T) {
	server, err := New(Options{
		APIKey: "mock-provider-key", CallbackURL: "http://127.0.0.1:1", CallbackSecret: "01234567890123456789012345678901",
		StartDelay: 0, DuplicateCallbacks: 1,
	}, slog.Default())
	if err != nil {
		t.Fatal(err)
	}
	request := voiceprovider.StartRequest{TaskRef: "sample-task", ConversationID: "sample-conversation"}
	for _, sample := range []string{"started", "final_transcript", "agent_answering", "interrupted", "usage", "error", "stopped"} {
		event, err := server.sampleEvent(request, sample)
		if err != nil {
			t.Fatalf("sample %s: %v", sample, err)
		}
		if event.TaskRef != request.TaskRef || event.ConversationID != request.ConversationID || event.EventID == "" || event.EventType == "" || event.Status == "" {
			t.Fatalf("sample %s event=%#v", sample, event)
		}
		if sample == "interrupted" && (len(event.Turns) != 1 || !event.Turns[0].Interrupted) {
			t.Fatalf("interrupted sample=%#v", event)
		}
		if sample == "usage" && len(event.Usage) != 1 {
			t.Fatalf("usage sample=%#v", event)
		}
	}
	if _, err := server.sampleEvent(request, "unknown"); err == nil {
		t.Fatal("unknown replay sample was accepted")
	}
}

func TestAdminReplayDeliversAnIdenticalSampleMultipleTimes(t *testing.T) {
	secret := "01234567890123456789012345678901"
	var callbacks atomic.Int32
	callbackServer := httptest.NewServer(http.HandlerFunc(func(writer http.ResponseWriter, request *http.Request) {
		body, _ := io.ReadAll(request.Body)
		if err := voiceprovider.VerifyCallback([]byte(secret), request.Header.Get(voiceprovider.HeaderCallbackTimestamp), request.Header.Get(voiceprovider.HeaderCallbackSignature), body, time.Now(), time.Minute); err != nil {
			t.Errorf("callback signature: %v", err)
		}
		callbacks.Add(1)
		writer.WriteHeader(http.StatusNoContent)
	}))
	defer callbackServer.Close()
	server, err := New(Options{APIKey: "mock-provider-key", CallbackURL: callbackServer.URL, CallbackSecret: secret, StartDelay: time.Hour, DuplicateCallbacks: 1}, slog.Default())
	if err != nil {
		t.Fatal(err)
	}
	httpServer := httptest.NewServer(server.Handler())
	defer httpServer.Close()
	client, _ := voiceprovider.NewHTTPClient(httpServer.URL, "mock-provider-key", time.Second)
	input := voiceprovider.StartRequest{TaskRef: "replay-task", RoomID: "room", DeviceUserID: "device", AgentUserID: "agent", ConversationID: "conversation", EngineMode: "s2s_pure"}
	if _, err := client.Start(t.Context(), input); err != nil {
		t.Fatal(err)
	}
	replay, _ := http.NewRequest(http.MethodPost, httpServer.URL+"/mock/v1/admin/replay", strings.NewReader(`{"task_ref":"replay-task","sample":"interrupted","duplicate_count":2}`))
	replay.Header.Set("Content-Type", "application/json")
	replay.Header.Set("X-Mock-Provider-Key", "mock-provider-key")
	response, err := http.DefaultClient.Do(replay)
	if err != nil {
		t.Fatal(err)
	}
	response.Body.Close()
	if response.StatusCode != http.StatusAccepted || callbacks.Load() != 2 {
		t.Fatalf("status=%d callbacks=%d", response.StatusCode, callbacks.Load())
	}
}