566 lines
17 KiB
Go
566 lines
17 KiB
Go
package tentacle
|
|
|
|
import (
|
|
"bufio"
|
|
"context"
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"os"
|
|
"os/exec"
|
|
"strconv"
|
|
"strings"
|
|
"sync"
|
|
"testing"
|
|
"time"
|
|
|
|
"git.stormux.org/storm/cthulhu/internal/input"
|
|
)
|
|
|
|
func test_manager(t testing.TB, limits Limits, overrides []input.BindingOverride) *Manager {
|
|
t.Helper()
|
|
manager, err := NewManager(limits, overrides)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
t.Cleanup(func() {
|
|
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
|
|
defer cancel()
|
|
if err := manager.Shutdown(ctx); err != nil {
|
|
t.Error(err)
|
|
}
|
|
if stats := manager.Stats(); stats != (Stats{}) {
|
|
t.Errorf("retained resources: %+v", stats)
|
|
}
|
|
})
|
|
return manager
|
|
}
|
|
|
|
func child_spec(t testing.TB, id, mode string) Spec {
|
|
t.Helper()
|
|
executable, err := os.Executable()
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
return Spec{ID: id, Program: []string{executable, "-test.run=^TestTentacleChild$", "--", "tentacle-child:" + mode}}
|
|
}
|
|
|
|
func wait_signal(t testing.TB, signal <-chan struct{}) {
|
|
t.Helper()
|
|
select {
|
|
case <-signal:
|
|
case <-time.After(5 * time.Second):
|
|
t.Fatal("timed out waiting for child lifecycle")
|
|
}
|
|
}
|
|
|
|
func start_ready(t testing.TB, manager *Manager, id, mode string) *Session {
|
|
t.Helper()
|
|
session, err := manager.Start(context.Background(), child_spec(t, id, mode))
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
wait_signal(t, session.Ready())
|
|
if state, err := session.Status(); state != Ready || err != nil {
|
|
t.Fatalf("state %s: %v", state, err)
|
|
}
|
|
return session
|
|
}
|
|
|
|
func start_active(t testing.TB, manager *Manager, id, mode string) *Session {
|
|
t.Helper()
|
|
session := start_ready(t, manager, id, mode)
|
|
if err := session.Activate(); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
return session
|
|
}
|
|
|
|
func wait_result(t testing.TB, result <-chan Result) Result {
|
|
t.Helper()
|
|
select {
|
|
case reply, ok := <-result:
|
|
if !ok {
|
|
t.Fatal("missing result")
|
|
}
|
|
return reply
|
|
case <-time.After(5 * time.Second):
|
|
t.Fatal("missing bounded result")
|
|
}
|
|
return Result{}
|
|
}
|
|
|
|
func TestActivationAndUnicodeRoundTrip(t *testing.T) {
|
|
manager := test_manager(t, DefaultLimits(), nil)
|
|
session := start_ready(t, manager, "sample", "echo")
|
|
if _, err := session.Invoke(context.Background(), "echo", nil); !errors.Is(err, ErrUnavailable) {
|
|
t.Fatal(err)
|
|
}
|
|
commands := session.Commands()
|
|
commands[0].ID = "changed"
|
|
commands[0].Bindings[0].Key = "changed"
|
|
if session.Commands()[0].ID != "echo" || session.Commands()[0].Bindings[0].Key == "changed" {
|
|
t.Fatal("registration alias")
|
|
}
|
|
if err := session.Activate(); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if err := session.Activate(); !errors.Is(err, ErrUnavailable) {
|
|
t.Fatal("double activation", err)
|
|
}
|
|
if _, err := session.Invoke(context.Background(), "unknown", nil); !errors.Is(err, ErrUnavailable) {
|
|
t.Fatal(err)
|
|
}
|
|
arguments := json.RawMessage(`{"text":"é猫😀é"}`)
|
|
pending, err := session.Invoke(context.Background(), "echo", arguments)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
result := wait_result(t, pending)
|
|
if result.Err != nil || string(result.Value) != string(arguments) || result.Session != session.Identity() {
|
|
t.Fatalf("%+v", result)
|
|
}
|
|
if _, open := <-pending; open {
|
|
t.Fatal("result channel not closed")
|
|
}
|
|
session.Disable()
|
|
if len(session.Commands()) != 0 || len(session.Bindings()) != 0 {
|
|
t.Fatal("disable did not revoke")
|
|
}
|
|
if _, err := session.Invoke(context.Background(), "echo", nil); !errors.Is(err, ErrUnavailable) {
|
|
t.Fatal(err)
|
|
}
|
|
wait_signal(t, session.Done())
|
|
if state, err := session.Status(); state != Stopped || err != nil {
|
|
t.Fatalf("state %s: %v", state, err)
|
|
}
|
|
replacement := start_active(t, manager, "sample", "echo")
|
|
if replacement.Identity() == session.Identity() {
|
|
t.Fatal("session identity reused")
|
|
}
|
|
}
|
|
|
|
func TestRepliesCorrelateOutOfOrder(t *testing.T) {
|
|
manager := test_manager(t, DefaultLimits(), nil)
|
|
session := start_active(t, manager, "sample", "reverse")
|
|
first, err := session.Invoke(context.Background(), "echo", json.RawMessage(`1`))
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
second, err := session.Invoke(context.Background(), "echo", json.RawMessage(`2`))
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
for index, pending := range []<-chan Result{first, second} {
|
|
result := wait_result(t, pending)
|
|
if result.Err != nil || string(result.Value) != strconv.Itoa(index+1) {
|
|
t.Fatalf("misrouted response: %+v", result)
|
|
}
|
|
}
|
|
}
|
|
|
|
func TestDisabledStagingCannotPublish(t *testing.T) {
|
|
manager := test_manager(t, DefaultLimits(), nil)
|
|
session := start_ready(t, manager, "sample", "echo")
|
|
session.Disable()
|
|
if err := session.Activate(); !errors.Is(err, ErrUnavailable) {
|
|
t.Fatal("disabled staging published", err)
|
|
}
|
|
wait_signal(t, session.Done())
|
|
if len(session.outbound) != 0 {
|
|
t.Fatal("cleanup retained unsent frames")
|
|
}
|
|
}
|
|
|
|
func TestBadChildrenAreContained(t *testing.T) {
|
|
for _, mode := range []string{"malformed", "oversized", "stderr", "wrong-major", "wrong-session", "duplicate-command", "invalid-binding", "unknown-response", "hang-start"} {
|
|
t.Run(mode, func(t *testing.T) {
|
|
limits := DefaultLimits()
|
|
limits.StartupTimeout = 500 * time.Millisecond
|
|
manager := test_manager(t, limits, nil)
|
|
session, err := manager.Start(context.Background(), child_spec(t, "broken", mode))
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
wait_signal(t, session.Done())
|
|
if state, err := session.Status(); state != Failed || err == nil || len(session.Commands()) != 0 {
|
|
t.Fatalf("state %s: %v", state, err)
|
|
}
|
|
// Failure releases capacity for a healthy tentacle.
|
|
start_active(t, manager, "healthy", "echo")
|
|
})
|
|
}
|
|
}
|
|
|
|
func TestRequestFailureCancellationAndDisable(t *testing.T) {
|
|
for _, mode := range []string{"crash", "duplicate-reply", "timeout", "cancel", "disable", "blocked-input"} {
|
|
t.Run(mode, func(t *testing.T) {
|
|
limits := DefaultLimits()
|
|
limits.RequestTimeout = 150 * time.Millisecond
|
|
manager := test_manager(t, limits, nil)
|
|
session := start_active(t, manager, "sample", mode)
|
|
ctx, cancel := context.WithCancel(context.Background())
|
|
defer cancel()
|
|
pending, err := session.Invoke(ctx, "echo", json.RawMessage(`{"text":"sample"}`))
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if mode == "cancel" {
|
|
cancel()
|
|
}
|
|
if mode == "disable" {
|
|
session.Disable()
|
|
}
|
|
result := wait_result(t, pending)
|
|
if mode != "duplicate-reply" && result.Err == nil {
|
|
t.Fatal("failed call succeeded")
|
|
}
|
|
if mode == "cancel" && !errors.Is(result.Err, context.Canceled) {
|
|
t.Fatal(result.Err)
|
|
}
|
|
if mode == "timeout" && !errors.Is(result.Err, context.DeadlineExceeded) {
|
|
t.Fatal(result.Err)
|
|
}
|
|
if mode == "disable" && !errors.Is(result.Err, ErrDisabled) {
|
|
t.Fatal(result.Err)
|
|
}
|
|
wait_signal(t, session.Done())
|
|
if len(session.Commands()) != 0 {
|
|
t.Fatal("failure retained commands")
|
|
}
|
|
})
|
|
}
|
|
}
|
|
|
|
func TestAdmissionAndAggregatePendingLimits(t *testing.T) {
|
|
limits := DefaultLimits()
|
|
limits.Processes, limits.Starting, limits.PendingPerProcess, limits.PendingTotal = 2, 1, 2, 3
|
|
manager := test_manager(t, limits, nil)
|
|
blocked, err := manager.Start(context.Background(), child_spec(t, "blocked", "hang-start"))
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if _, err := manager.Start(context.Background(), child_spec(t, "second", "echo")); !errors.Is(err, ErrBusy) {
|
|
t.Fatal("startup limit", err)
|
|
}
|
|
blocked.Disable()
|
|
wait_signal(t, blocked.Done())
|
|
first := start_active(t, manager, "first", "timeout")
|
|
second := start_active(t, manager, "second", "timeout")
|
|
if _, err := manager.Start(context.Background(), child_spec(t, "third", "echo")); !errors.Is(err, ErrBusy) {
|
|
t.Fatal("process limit", err)
|
|
}
|
|
for range 2 {
|
|
if _, err := first.Invoke(context.Background(), "echo", nil); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
}
|
|
if _, err := first.Invoke(context.Background(), "echo", nil); !errors.Is(err, ErrBusy) {
|
|
t.Fatal("per-child limit", err)
|
|
}
|
|
if _, err := second.Invoke(context.Background(), "echo", nil); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if _, err := second.Invoke(context.Background(), "echo", nil); !errors.Is(err, ErrBusy) {
|
|
t.Fatal("aggregate limit", err)
|
|
}
|
|
if stats := manager.Stats(); stats.Pending != 3 || stats.Processes != 2 {
|
|
t.Fatalf("%+v", stats)
|
|
}
|
|
first.Disable()
|
|
if stats := manager.Stats(); stats.Pending != 1 {
|
|
t.Fatalf("pending work not revoked: %+v", stats)
|
|
}
|
|
}
|
|
|
|
func TestParentCancellationAndConcurrentShutdown(t *testing.T) {
|
|
manager := test_manager(t, DefaultLimits(), nil)
|
|
ctx, cancel := context.WithCancel(context.Background())
|
|
session, err := manager.Start(ctx, child_spec(t, "sample", "echo"))
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
wait_signal(t, session.Ready())
|
|
if err := session.Activate(); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
var workers sync.WaitGroup
|
|
for range 16 {
|
|
workers.Add(1)
|
|
go func() {
|
|
defer workers.Done()
|
|
_, _ = session.Invoke(context.Background(), "echo", nil)
|
|
session.Disable()
|
|
}()
|
|
}
|
|
cancel()
|
|
workers.Wait()
|
|
wait_signal(t, session.Done())
|
|
if err := manager.Shutdown(context.Background()); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if _, err := manager.Start(context.Background(), child_spec(t, "after-shutdown", "echo")); !errors.Is(err, ErrUnavailable) {
|
|
t.Fatal(err)
|
|
}
|
|
}
|
|
|
|
func TestBlockedPipeCannotBlockSubmissionOrDisable(t *testing.T) {
|
|
limits := DefaultLimits()
|
|
limits.RequestTimeout = 10 * time.Second
|
|
manager := test_manager(t, limits, nil)
|
|
session := start_active(t, manager, "sample", "blocked-input")
|
|
arguments, err := json.Marshal(strings.Repeat("x", MaxArgumentBytes-2))
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
submitted := make(chan error, 1)
|
|
go func() {
|
|
for range limits.PendingPerProcess {
|
|
if _, err := session.Invoke(context.Background(), "echo", arguments); err != nil {
|
|
submitted <- err
|
|
return
|
|
}
|
|
}
|
|
session.Disable()
|
|
submitted <- nil
|
|
}()
|
|
select {
|
|
case err := <-submitted:
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
case <-time.After(2 * time.Second):
|
|
t.Fatal("pipe backpressure blocked host submission or disable")
|
|
}
|
|
wait_signal(t, session.Done())
|
|
}
|
|
|
|
func TestLaunchValidationAndMissingExecutable(t *testing.T) {
|
|
manager := test_manager(t, DefaultLimits(), nil)
|
|
if _, err := manager.Start(nil, child_spec(t, "sample", "echo")); err == nil {
|
|
t.Fatal("nil context accepted")
|
|
}
|
|
ctx, cancel := context.WithCancel(context.Background())
|
|
cancel()
|
|
if _, err := manager.Start(ctx, child_spec(t, "sample", "echo")); !errors.Is(err, context.Canceled) {
|
|
t.Fatal(err)
|
|
}
|
|
for _, spec := range []Spec{
|
|
{ID: "../unsafe", Program: []string{"/bin/true"}},
|
|
{ID: "sample", Program: []string{"relative"}},
|
|
{ID: "sample", Program: []string{"/bin/true", "nul\x00argument"}},
|
|
{ID: "sample", Program: []string{"/bin/true"}, Directory: "relative"},
|
|
} {
|
|
if _, err := manager.Start(context.Background(), spec); err == nil {
|
|
t.Fatal("invalid launch accepted")
|
|
}
|
|
}
|
|
session, err := manager.Start(context.Background(), Spec{ID: "missing", Program: []string{t.TempDir() + "/missing"}})
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
wait_signal(t, session.Done())
|
|
if state, err := session.Status(); state != Failed || !errors.Is(err, ErrExited) {
|
|
t.Fatalf("%s: %v", state, err)
|
|
}
|
|
if stats := manager.Stats(); stats != (Stats{}) {
|
|
t.Fatalf("failed spawn retained resources: %+v", stats)
|
|
}
|
|
}
|
|
|
|
func TestBindingConflictsLeaveCommandsAvailable(t *testing.T) {
|
|
manager := test_manager(t, DefaultLimits(), nil)
|
|
first := start_active(t, manager, "first", "echo")
|
|
bindings := first.Bindings()
|
|
if len(bindings) != 2 || bindings[0].Bound || bindings[0].Conflict != string(input.CycleSpeechMode) || !bindings[1].Bound {
|
|
t.Fatalf("core conflict: %+v", bindings)
|
|
}
|
|
second := start_active(t, manager, "second", "echo")
|
|
if bindings := second.Bindings(); bindings[1].Bound || bindings[1].Conflict != "tentacle/first/echo" {
|
|
t.Fatalf("tentacle conflict: %+v", bindings)
|
|
}
|
|
if len(second.Commands()) == 0 {
|
|
t.Fatal("unbound command disappeared")
|
|
}
|
|
first.Disable()
|
|
// No implicit rebinding of the existing second tentacle.
|
|
if second.Bindings()[1].Bound {
|
|
t.Fatal("silently rebound")
|
|
}
|
|
third := start_active(t, manager, "third", "echo")
|
|
if !third.Bindings()[1].Bound {
|
|
t.Fatal("revoked shortcut remains reserved")
|
|
}
|
|
}
|
|
|
|
func TestBindingsRespectUserOverridesAndScopes(t *testing.T) {
|
|
manager := test_manager(t, DefaultLimits(), []input.BindingOverride{{
|
|
Layout: input.DesktopKeyboardLayout, Scope: "global", Command: input.PresentTime, Keys: []string{"Cthulhu+Ctrl+F12"},
|
|
}})
|
|
session := start_active(t, manager, "sample", "echo")
|
|
if bindings := session.Bindings(); bindings[1].Bound || bindings[1].Conflict != string(input.PresentTime) {
|
|
t.Fatalf("user override conflict: %+v", bindings)
|
|
}
|
|
occupied := manager.occupied_bindings()
|
|
commands := []Command{{ID: "scoped", Bindings: []Binding{
|
|
{Layout: "desktop", Scope: "global", Key: "h"},
|
|
{Layout: "laptop", Scope: "global", Key: "Cthulhu+Ctrl+F12"},
|
|
{Layout: "laptop", Scope: "browse", Key: "Ctrl+Cthulhu+F12"},
|
|
}}}
|
|
bindings := resolve_bindings("scope", commands, occupied)
|
|
if bindings[0].Bound || !bindings[1].Bound || bindings[2].Bound {
|
|
t.Fatalf("scope/alias conflicts: %+v", bindings)
|
|
}
|
|
}
|
|
|
|
// This is a private protocol fixture, not a public SDK or installable tentacle.
|
|
func TestTentacleChild(t *testing.T) {
|
|
mode, found := strings.CutPrefix(os.Args[len(os.Args)-1], "tentacle-child:")
|
|
if !found {
|
|
return
|
|
}
|
|
if mode == "descendant" {
|
|
<-time.After(time.Hour)
|
|
os.Exit(0)
|
|
}
|
|
reader := bufio.NewReader(os.Stdin)
|
|
encoder := json.NewEncoder(os.Stdout)
|
|
var heldID string
|
|
var heldValue json.RawMessage
|
|
for {
|
|
frame, err := read_frame(reader)
|
|
if err != nil {
|
|
os.Exit(0)
|
|
}
|
|
var message struct {
|
|
ID string `json:"id"`
|
|
Method string `json:"method"`
|
|
Params json.RawMessage `json:"params"`
|
|
}
|
|
if json.Unmarshal(frame, &message) != nil {
|
|
os.Exit(2)
|
|
}
|
|
reply := func(value any) {
|
|
_ = encoder.Encode(map[string]any{"jsonrpc": "2.0", "id": message.ID, "result": value})
|
|
}
|
|
switch message.Method {
|
|
case "initialize":
|
|
switch mode {
|
|
case "hang-start":
|
|
continue
|
|
case "malformed":
|
|
fmt.Fprintln(os.Stdout, "not JSON")
|
|
continue
|
|
case "oversized":
|
|
fmt.Fprintln(os.Stdout, strings.Repeat("x", MaxFrameBytes+1))
|
|
continue
|
|
case "stderr":
|
|
fmt.Fprint(os.Stderr, strings.Repeat("x", MaxDiagnosticBytes+1))
|
|
continue
|
|
case "unknown-response":
|
|
_ = encoder.Encode(map[string]any{"jsonrpc": "2.0", "id": "999", "result": nil})
|
|
continue
|
|
}
|
|
var initialization Initialization
|
|
_ = json.Unmarshal(message.Params, &initialization)
|
|
registration := Registration{Protocol: initialization.Protocol, Session: initialization.Session, Commands: []Command{{
|
|
ID: "echo", Title: "Echo text", Description: "Private protocol fixture", Bindings: []Binding{
|
|
{Layout: "desktop", Scope: "global", Key: "Cthulhu+S"},
|
|
{Layout: "desktop", Scope: "global", Key: "Cthulhu+Ctrl+F12"},
|
|
},
|
|
}}}
|
|
switch mode {
|
|
case "wrong-major":
|
|
registration.Protocol.Major++
|
|
case "wrong-session":
|
|
registration.Session = "wrong"
|
|
case "duplicate-command":
|
|
registration.Commands = append(registration.Commands, registration.Commands[0])
|
|
case "invalid-binding":
|
|
registration.Commands[0].Bindings[0].Key = "not-a-key"
|
|
}
|
|
reply(registration)
|
|
case "initialized":
|
|
if mode == "blocked-input" {
|
|
<-time.After(time.Hour)
|
|
}
|
|
case "command.invoke":
|
|
switch mode {
|
|
case "crash":
|
|
os.Exit(7)
|
|
case "timeout", "cancel", "disable":
|
|
continue
|
|
case "spawn":
|
|
executable, _ := os.Executable()
|
|
child := exec.Command(executable, "-test.run=^TestTentacleChild$", "--", "tentacle-child:descendant")
|
|
child.Stdout, child.Stderr = os.Stdout, os.Stderr
|
|
if child.Start() != nil {
|
|
os.Exit(8)
|
|
}
|
|
reply(child.Process.Pid)
|
|
continue
|
|
}
|
|
var params struct {
|
|
Arguments json.RawMessage `json:"arguments"`
|
|
}
|
|
_ = json.Unmarshal(message.Params, ¶ms)
|
|
if mode == "reverse" && heldID == "" {
|
|
heldID, heldValue = message.ID, params.Arguments
|
|
continue
|
|
}
|
|
reply(params.Arguments)
|
|
if mode == "reverse" {
|
|
_ = encoder.Encode(map[string]any{"jsonrpc": "2.0", "id": heldID, "result": heldValue})
|
|
heldID, heldValue = "", nil
|
|
}
|
|
if mode == "duplicate-reply" {
|
|
reply(params.Arguments)
|
|
}
|
|
case "shutdown":
|
|
os.Exit(0)
|
|
default:
|
|
os.Exit(3)
|
|
}
|
|
}
|
|
}
|
|
|
|
func TestDisableKillsGroupDescendantsAndJoinsPipes(t *testing.T) {
|
|
manager := test_manager(t, DefaultLimits(), nil)
|
|
session := start_active(t, manager, "sample", "spawn")
|
|
pending, err := session.Invoke(context.Background(), "echo", nil)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
result := wait_result(t, pending)
|
|
var pid int
|
|
if result.Err != nil || json.Unmarshal(result.Value, &pid) != nil || pid <= 1 {
|
|
t.Fatalf("bad descendant result: %+v", result)
|
|
}
|
|
session.Disable()
|
|
wait_signal(t, session.Done())
|
|
// SIGKILL delivery/reaping is scheduled by the kernel; wait for /proc to
|
|
// report disappearance or zombie state, not an arbitrary sleep duration.
|
|
deadline := time.NewTimer(2 * time.Second)
|
|
defer deadline.Stop()
|
|
ticker := time.NewTicker(time.Millisecond)
|
|
defer ticker.Stop()
|
|
for {
|
|
data, err := os.ReadFile("/proc/" + strconv.Itoa(pid) + "/stat")
|
|
if os.IsNotExist(err) {
|
|
break
|
|
}
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
_, tail, found := strings.Cut(string(data), ") ")
|
|
if found && strings.HasPrefix(tail, "Z ") {
|
|
break
|
|
}
|
|
select {
|
|
case <-ticker.C:
|
|
case <-deadline.C:
|
|
t.Fatal("owned descendant remained running")
|
|
}
|
|
}
|
|
}
|