114 lines
3.6 KiB
Go
114 lines
3.6 KiB
Go
package launch
|
|
|
|
import (
|
|
"encoding/json"
|
|
"os"
|
|
"path/filepath"
|
|
"sync"
|
|
"testing"
|
|
"time"
|
|
)
|
|
|
|
// A bundle that asks to take a worker's work and answers each piece it is handed.
|
|
const takingBundle = `#!/bin/sh
|
|
while IFS= read -r line; do
|
|
id=$(printf '%s' "$line" | sed -n 's/.*"id":\([0-9]*\)[,}].*/\1/p')
|
|
case "$line" in
|
|
*'"method":"initialize"'*) printf '{"jsonrpc":"2.0","id":%s,"result":{}}\n' "$id" ;;
|
|
*'"method":"tools/list"'*)
|
|
printf '{"jsonrpc":"2.0","id":%s,"result":{"tools":[]}}\n' "$id"
|
|
printf '{"jsonrpc":"2.0","id":"s1","method":"mesh/seat.take","params":{"worker":"SEAT_CHANNEL_TELEGRAM_worker"}}\n' ;;
|
|
*'"method":"mesh/work"'*) printf '{"jsonrpc":"2.0","id":%s,"result":{"taken":true}}\n' "$id" ;;
|
|
esac
|
|
done
|
|
`
|
|
|
|
type seating struct {
|
|
subscribing
|
|
mu sync.Mutex
|
|
verb string
|
|
params string
|
|
handOn func(string, any) (json.RawMessage, error)
|
|
}
|
|
|
|
func (b *seating) Seat(verb string, params json.RawMessage, handOn func(string, any) (json.RawMessage, error)) (json.RawMessage, error) {
|
|
b.mu.Lock()
|
|
defer b.mu.Unlock()
|
|
b.verb, b.params, b.handOn = verb, string(params), handOn
|
|
return json.RawMessage(`{}`), nil
|
|
}
|
|
|
|
// novox/hq ADR 0259 §3: a bundle's seat traffic reaches the runtime's bus from its own channel, and the work
|
|
// it takes is handed back to it.
|
|
func TestABundleTakesASeatsWorkThroughItsOwnChannel(t *testing.T) {
|
|
entry := filepath.Join(t.TempDir(), "bundle")
|
|
if err := os.WriteFile(entry, []byte(takingBundle), 0o755); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
b := &seating{}
|
|
l, err := Start("telegram", entry, os.Environ(), b, t.Logf)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
t.Cleanup(l.Stop)
|
|
var handOn func(string, any) (json.RawMessage, error)
|
|
for i := 0; handOn == nil; i++ {
|
|
if i > 100 {
|
|
t.Fatal("the bundle never asked to take its work")
|
|
}
|
|
time.Sleep(20 * time.Millisecond)
|
|
b.mu.Lock()
|
|
handOn = b.handOn
|
|
b.mu.Unlock()
|
|
}
|
|
if b.verb != "take" || b.params != `{"worker":"SEAT_CHANNEL_TELEGRAM_worker"}` {
|
|
t.Fatalf("asked %s %s", b.verb, b.params)
|
|
}
|
|
got, err := handOn("mesh/work", map[string]any{"work": map[string]any{"subject": "mesh.seat.channel.accept.show.telegram"}})
|
|
if err != nil || string(got) != `{"taken":true}` {
|
|
t.Fatalf("the work was not handed to the bundle: %s %v", got, err)
|
|
}
|
|
}
|
|
|
|
// A bundle that pulls one ask from its worker when it is listed.
|
|
const pullingBundle = `#!/bin/sh
|
|
while IFS= read -r line; do
|
|
id=$(printf '%s' "$line" | sed -n 's/.*"id":\([0-9]*\)[,}].*/\1/p')
|
|
case "$line" in
|
|
*'"method":"initialize"'*) printf '{"jsonrpc":"2.0","id":%s,"result":{}}\n' "$id" ;;
|
|
*'"method":"tools/list"'*)
|
|
printf '{"jsonrpc":"2.0","id":%s,"result":{"tools":[]}}\n' "$id"
|
|
printf '{"jsonrpc":"2.0","id":"p1","method":"mesh/seat.pull","params":{"worker":"SEAT_NODE_BUILD_GO_DELIVERY_worker","wait_ms":0}}\n' ;;
|
|
esac
|
|
done
|
|
`
|
|
|
|
// 1887/A11: a bundle's pull reaches the runtime's bus from its own channel, as its other seat traffic does.
|
|
func TestABundlePullsASeatsWorkThroughItsOwnChannel(t *testing.T) {
|
|
entry := filepath.Join(t.TempDir(), "bundle")
|
|
if err := os.WriteFile(entry, []byte(pullingBundle), 0o755); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
b := &seating{}
|
|
l, err := Start("mesh-build-go", entry, os.Environ(), b, t.Logf)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
t.Cleanup(l.Stop)
|
|
for i := 0; ; i++ {
|
|
b.mu.Lock()
|
|
verb, params := b.verb, b.params
|
|
b.mu.Unlock()
|
|
if verb == "pull" {
|
|
if params != `{"worker":"SEAT_NODE_BUILD_GO_DELIVERY_worker","wait_ms":0}` {
|
|
t.Fatalf("pulled with %s", params)
|
|
}
|
|
return
|
|
}
|
|
if i > 100 {
|
|
t.Fatal("the bundle's pull never reached the runtime")
|
|
}
|
|
time.Sleep(20 * time.Millisecond)
|
|
}
|
|
}
|