106 lines
2.9 KiB
Go
106 lines
2.9 KiB
Go
package outbox
|
|
|
|
import (
|
|
"context"
|
|
"io"
|
|
"os"
|
|
"os/exec"
|
|
"path/filepath"
|
|
"runtime"
|
|
"strings"
|
|
"testing"
|
|
"time"
|
|
)
|
|
|
|
func TestIntegrationSenderCLI(t *testing.T) {
|
|
s, api, base, _ := realStore(t)
|
|
ctx, stop := context.WithTimeout(context.Background(), 30*time.Second)
|
|
defer stop()
|
|
binary := filepath.Join(t.TempDir(), "sender")
|
|
build := exec.CommandContext(ctx, filepath.Join(runtime.GOROOT(), "bin", "go"), "build", "-o", binary, "../../cmd/sender")
|
|
if raw, err := build.CombinedOutput(); err != nil {
|
|
t.Fatalf("build sender: %v %s", err, raw)
|
|
}
|
|
values := map[string]string{"ORBIT_SENDER_DSN": os.Getenv("ORBIT_TEST_SENDER_DSN"), "ORBIT_URL": base, "ORBIT_CREDENTIALS_FILE": os.Getenv("ORBIT_TEST_SENDER_CREDENTIALS_FILE"), "ORBIT_NAMESPACE_ID": s.namespace, "ORBIT_QUEUE_ID": s.queue, "ORBIT_TOKEN": ""}
|
|
env := []string{}
|
|
for _, entry := range os.Environ() {
|
|
key, _, _ := strings.Cut(entry, "=")
|
|
if _, ok := values[key]; !ok {
|
|
env = append(env, entry)
|
|
}
|
|
}
|
|
for key, value := range values {
|
|
env = append(env, key+"="+value)
|
|
}
|
|
call := func(args ...string) string {
|
|
t.Helper()
|
|
cmd := exec.CommandContext(ctx, binary, args...)
|
|
cmd.Env = env
|
|
raw, err := cmd.CombinedOutput()
|
|
if err != nil {
|
|
t.Fatalf("sender command failed: %v %s", err, raw)
|
|
}
|
|
return string(raw)
|
|
}
|
|
id := testID(t)
|
|
if !strings.Contains(call("enqueue", id, "objects/cli.txt"), "duplicate=false") || !strings.Contains(call("enqueue", id, "objects/cli.txt"), "duplicate=true") || !strings.Contains(call("status", id), "state=PENDING") {
|
|
t.Fatal("CLI local transaction mismatch")
|
|
}
|
|
cmd := exec.CommandContext(ctx, binary, "send")
|
|
cmd.Env = env
|
|
cmd.Stdout = io.Discard
|
|
cmd.Stderr = io.Discard
|
|
if err := cmd.Start(); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
done := make(chan struct{})
|
|
var waitErr error
|
|
go func() { waitErr = cmd.Wait(); close(done) }()
|
|
t.Cleanup(func() {
|
|
select {
|
|
case <-done:
|
|
return
|
|
default:
|
|
}
|
|
cmd.Process.Kill()
|
|
select {
|
|
case <-done:
|
|
case <-time.After(5 * time.Second):
|
|
t.Error("owned sender CLI did not exit")
|
|
}
|
|
})
|
|
var r Record
|
|
for {
|
|
r = status(t, s, id)
|
|
if r.State == "ACCEPTED" {
|
|
break
|
|
}
|
|
if r.State == "REVIEW" {
|
|
t.Fatal("CLI needs review", r.LastCode)
|
|
}
|
|
select {
|
|
case <-done:
|
|
t.Fatal("sender exited before acceptance", waitErr)
|
|
case <-ctx.Done():
|
|
t.Fatal("CLI sender timeout")
|
|
case <-time.After(100 * time.Millisecond):
|
|
}
|
|
}
|
|
if err := cmd.Process.Signal(os.Interrupt); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
select {
|
|
case <-done:
|
|
if waitErr != nil {
|
|
t.Fatal("sender did not stop gracefully", waitErr)
|
|
}
|
|
case <-ctx.Done():
|
|
t.Fatal("sender shutdown timeout")
|
|
}
|
|
if !strings.Contains(call("status", id), "state=ACCEPTED") || countRows(t, s, "sender_jobs") != 1 || r.SendCount != 1 {
|
|
t.Fatal("CLI acceptance did not persist")
|
|
}
|
|
cancelAccepted(t, api, *r.TaskID)
|
|
t.Log("built CLI enqueue duplicate/status/send/graceful stop exercised against real business DB and Orbit")
|
|
}
|