54 lines
1.8 KiB
Go
54 lines
1.8 KiB
Go
package store
|
|
|
|
import (
|
|
"errors"
|
|
"path/filepath"
|
|
"testing"
|
|
)
|
|
|
|
func TestStoreRejectsForeignDispatcherWithoutTouchingOutbox(t *testing.T) {
|
|
const foreignID = "862c8e9b-26b8-4642-9fac-c7d85f487a30"
|
|
const eventID = "retained-outbox-event"
|
|
const originalBody = `{"proof":"original"}`
|
|
path := filepath.Join(t.TempDir(), "dispatcher.sqlite")
|
|
owner, err := Open(path)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if err := owner.CloseAdmission(currentDispatcherID); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if _, err := owner.db.Exec(`INSERT INTO dispatcher_outbox(dispatcher_id,event_id,event_type,routing_key,body,confirmed)
|
|
VALUES(?,?,?,?,?,0)`, currentDispatcherID, eventID, "call.execute.result", "agent-call.saas.events.v1", []byte(originalBody)); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if err := owner.Close(); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
|
|
foreign, err := Open(path)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
defer foreign.Close()
|
|
if err := foreign.CloseAdmission(foreignID); !errors.Is(err, ErrDispatcherIdentityMismatch) {
|
|
t.Fatalf("foreign Dispatcher admitted to an existing database: %v", err)
|
|
}
|
|
var count int
|
|
var survivingID string
|
|
if err := foreign.db.QueryRow(`SELECT COUNT(*), MIN(dispatcher_id) FROM dispatcher_state`).Scan(&count, &survivingID); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if count != 1 || survivingID != currentDispatcherID {
|
|
t.Fatalf("foreign startup changed the original Dispatcher identity: count=%d identity=%q", count, survivingID)
|
|
}
|
|
var body []byte
|
|
var confirmed int
|
|
if err := foreign.db.QueryRow(`SELECT body,confirmed FROM dispatcher_outbox WHERE dispatcher_id=? AND event_id=?`, currentDispatcherID, eventID).Scan(&body, &confirmed); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if string(body) != originalBody || confirmed != 0 {
|
|
t.Fatal("foreign startup changed an unconfirmed original outbox record")
|
|
}
|
|
}
|