package tentacle

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"os/exec"
	"path/filepath"
	"testing"
)

func TestPythonPeerUsesSameContract(t *testing.T) {
	python, err := exec.LookPath("python3")
	if err != nil {
		t.Skip("Python 3 not installed")
	}
	fixture, err := filepath.Abs("testdata/peer.py")
	if err != nil {
		t.Fatal(err)
	}
	manager := test_manager(t, DefaultLimits(), nil)
	session, err := manager.Start(context.Background(), Spec{ID: "python", Program: []string{python, "-I", "-u", fixture}})
	if err != nil {
		t.Fatal(err)
	}
	wait_signal(t, session.Ready())
	if err := session.Activate(); err != nil {
		t.Fatal(err)
	}
	arguments := json.RawMessage(`{"text":"é猫😀é","enabled":false,"count":0}`)
	pending, err := session.Invoke(context.Background(), "echo", arguments)
	if err != nil {
		t.Fatal(err)
	}
	result := wait_result(t, pending)
	var value struct {
		Text    string
		Enabled bool
		Count   int
	}
	if result.Err != nil || json.Unmarshal(result.Value, &value) != nil || value.Text != "é猫😀é" || value.Enabled || value.Count != 0 {
		t.Fatalf("Python round trip: %+v, error %v", value, result.Err)
	}
	if result.Session != session.Identity() {
		t.Fatal("wrong session")
	}
}

func TestManyTentaclesHaveAggregateBoundAndIndependentProgress(t *testing.T) {
	manager := test_manager(t, DefaultLimits(), nil)
	sessions := make([]*Session, 0, 16)
	// Keep one healthy peer alongside fifteen peers that never reply.
	healthy := start_active(t, manager, "healthy", "echo")
	for index := range 15 {
		sessions = append(sessions, start_active(t, manager, fmt.Sprintf("blocked-%d", index), "timeout"))
	}
	for index := range 63 {
		if _, err := sessions[index%len(sessions)].Invoke(context.Background(), "echo", nil); err != nil {
			t.Fatal(err)
		}
	}
	result, err := healthy.Invoke(context.Background(), "echo", json.RawMessage(`{"healthy":true}`))
	if err != nil {
		t.Fatal(err)
	}
	if reply := wait_result(t, result); reply.Err != nil || string(reply.Value) != `{"healthy":true}` {
		t.Fatalf("healthy peer stalled: %+v", reply)
	}
	if _, err := sessions[0].Invoke(context.Background(), "echo", nil); err != nil {
		t.Fatal(err)
	}
	if _, err := healthy.Invoke(context.Background(), "echo", nil); !errors.Is(err, ErrBusy) {
		t.Fatal("missing aggregate backpressure", err)
	}
	if stats := manager.Stats(); stats.Pending != 64 || stats.Processes != 16 {
		t.Fatalf("%+v", stats)
	}
	for _, session := range sessions {
		session.Disable()
	}
	result, err = healthy.Invoke(context.Background(), "echo", nil)
	if err != nil {
		t.Fatal(err)
	}
	if reply := wait_result(t, result); reply.Err != nil {
		t.Fatal(reply.Err)
	}
}

func BenchmarkCommandRoundTrip(b *testing.B) {
	for _, count := range []int{1, 8, 16} {
		b.Run(fmt.Sprintf("processes-%d", count), func(b *testing.B) {
			manager := test_manager(b, DefaultLimits(), nil)
			sessions := make([]*Session, count)
			for index := range count {
				sessions[index] = start_active(b, manager, fmt.Sprintf("peer-%d", index), "echo")
			}
			arguments := json.RawMessage(`{"text":"sample"}`)
			b.ReportAllocs()
			b.ResetTimer()
			for index := range b.N {
				pending, err := sessions[index%count].Invoke(context.Background(), "echo", arguments)
				if err != nil {
					b.Fatal(err)
				}
				if result := <-pending; result.Err != nil {
					b.Fatal(result.Err)
				}
			}
			b.StopTimer()
		})
	}
}