236 lines
10 KiB
Go
236 lines
10 KiB
Go
package bell_connector_test
|
|
|
|
import (
|
|
"bytes"
|
|
"context"
|
|
"crypto/ed25519"
|
|
"crypto/rand"
|
|
"crypto/tls"
|
|
"encoding/json"
|
|
"net/http"
|
|
"net/http/httptest"
|
|
"os"
|
|
"path/filepath"
|
|
"strings"
|
|
"sync/atomic"
|
|
"testing"
|
|
"time"
|
|
|
|
"github.com/gin-gonic/gin"
|
|
"gorm.io/driver/sqlite"
|
|
"gorm.io/gorm"
|
|
|
|
"git.ilapage.cn/ila/yovision/Sense/server/app/sense/integration/bell_connector"
|
|
"git.ilapage.cn/ila/yovision/Sense/server/app/sense/integration/machine_identity"
|
|
"git.ilapage.cn/ila/yovision/Sense/server/app/sense/outbox"
|
|
)
|
|
|
|
func TestPersistentOutboxRetriesWithoutChangingOriginalIdentity(t *testing.T) {
|
|
db, databasePath := openDatabase(t)
|
|
body := fixture(t)
|
|
_, privateKey, _ := ed25519.GenerateKey(rand.Reader)
|
|
signer := machine_identity.Signer{Principal: "yv:sense:school-a", KeyID: "sense-key-0001", PrivateKey: privateKey, Now: func() time.Time { return time.Date(2026, 8, 31, 1, 0, 0, 0, time.UTC) }}
|
|
var attempts atomic.Int32
|
|
var received [][]byte
|
|
server := httptest.NewServer(http.HandlerFunc(func(writer http.ResponseWriter, request *http.Request) {
|
|
payload := make([]byte, request.ContentLength)
|
|
_, _ = request.Body.Read(payload)
|
|
received = append(received, payload)
|
|
requestID := request.Header.Get("X-Request-ID")
|
|
writer.Header().Set("X-Request-ID", requestID)
|
|
if request.Header.Get("X-YoVision-Relay-ID") != "sense-school-a" || !strings.HasPrefix(request.Header.Get("Authorization"), "Bearer ") || len(requestID) < 16 || len(requestID) > 128 {
|
|
http.Error(writer, "missing relay identity", http.StatusUnauthorized)
|
|
return
|
|
}
|
|
if attempts.Add(1) == 1 {
|
|
writer.WriteHeader(http.StatusServiceUnavailable)
|
|
_, _ = writer.Write([]byte(`{"code":"ingest_unavailable","message":"temporarily unavailable"}`))
|
|
return
|
|
}
|
|
writer.Header().Set("Content-Type", "application/json")
|
|
writer.WriteHeader(http.StatusCreated)
|
|
_, _ = writer.Write([]byte(`{"event_id":"bell-event-1","producer_id":"brain-school-a","source_event_id":"evt-area-20260831-0001","disposition":"created","payload_sha256":"4cc1e93820195caf713ea675ff33f178c9d4997dd8a81cb61287e9fea0e3d5e1"}`))
|
|
}))
|
|
defer server.Close()
|
|
|
|
now := time.Date(2026, 8, 31, 1, 0, 0, 0, time.UTC)
|
|
if _, err := bell_connector.EnqueueEvent(db, body, now); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
client := &bell_connector.Client{Endpoint: server.URL, RelayID: "sense-school-a", Signer: signer, HTTP: server.Client(), Enabled: true}
|
|
relay := bell_connector.Relay{DB: db, Client: client, Now: func() time.Time { return now }, Backoff: func(int) time.Duration { return 0 }}
|
|
if delivered, err := relay.DeliverBatch(context.Background(), "worker-1", 10); err != nil || delivered != 0 {
|
|
t.Fatalf("unavailable delivery=%d err=%v", delivered, err)
|
|
}
|
|
var queued outbox.Message
|
|
if err := db.First(&queued).Error; err != nil || queued.State != outbox.StateRetry {
|
|
t.Fatalf("outbox was not retained for retry: state=%s err=%v", queued.State, err)
|
|
}
|
|
sqlDatabase, err := db.DB()
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if err = sqlDatabase.Close(); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
db = openDatabasePath(t, databasePath)
|
|
if err = db.First(&queued).Error; err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
|
|
// Constructing a new relay simulates restart; the durable row is claimed
|
|
// and delivered with its original bytes and business identity.
|
|
restarted := bell_connector.Relay{DB: db, Client: client, Now: func() time.Time { return now }, Backoff: func(int) time.Duration { return 0 }}
|
|
if delivered, err := restarted.DeliverBatch(context.Background(), "worker-2", 10); err != nil || delivered != 1 {
|
|
t.Fatalf("recovery delivery=%d err=%v", delivered, err)
|
|
}
|
|
if len(received) != 2 || string(received[0]) != string(body) || string(received[1]) != string(body) || bell_connector.PreserveIdentity(received[0], received[1]) != nil {
|
|
t.Fatal("relay changed the frozen payload or original identity")
|
|
}
|
|
if err := db.First(&queued).Error; err != nil || queued.State != outbox.StateDelivered {
|
|
t.Fatalf("outbox not delivered: state=%s err=%v", queued.State, err)
|
|
}
|
|
}
|
|
|
|
func TestTerminalConflictDisabledTimeoutAndReplayRestart(t *testing.T) {
|
|
db, databasePath := openDatabase(t)
|
|
body := fixture(t)
|
|
_, privateKey, _ := ed25519.GenerateKey(rand.Reader)
|
|
signer := machine_identity.Signer{Principal: "yv:sense:school-a", KeyID: "sense-key-0001", PrivateKey: privateKey}
|
|
policy := machine_identity.TransportPolicy{TLSMinVersion: tls.VersionTLS12, VerifyCertificate: true, VerifyHostname: true, ConnectTimeout: time.Second, ResponseHeaderTimeout: time.Second, RequestTimeout: time.Second, MaxRequestBytes: 64 * 1024}
|
|
if _, err := bell_connector.NewClient("https://user:pass@bell.example", "sense-school-a", signer, policy); err == nil || !strings.Contains(err.Error(), "HTTPS origin") {
|
|
t.Fatalf("endpoint userinfo was not rejected: %v", err)
|
|
}
|
|
conflictServer := httptest.NewServer(http.HandlerFunc(func(writer http.ResponseWriter, request *http.Request) {
|
|
writer.WriteHeader(http.StatusConflict)
|
|
_, _ = writer.Write([]byte(`{"code":"idempotency_conflict","message":"conflict","existing_event_id":"bell-1"}`))
|
|
}))
|
|
defer conflictServer.Close()
|
|
if _, err := bell_connector.EnqueueEvent(db, body, time.Now()); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
client := &bell_connector.Client{Endpoint: conflictServer.URL, Signer: signer, HTTP: conflictServer.Client(), Enabled: true}
|
|
if _, err := (bell_connector.Relay{DB: db, Client: client}).DeliverBatch(context.Background(), "worker", 1); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
var message outbox.Message
|
|
if err := db.First(&message).Error; err != nil || message.State != outbox.StateDead {
|
|
t.Fatalf("terminal conflict was retried: state=%s err=%v", message.State, err)
|
|
}
|
|
if _, err := (bell_connector.Client{Enabled: false}).Send(context.Background(), body); err == nil || !strings.Contains(err.Error(), "connector_disabled") {
|
|
t.Fatalf("disabled connector error=%v", err)
|
|
}
|
|
cancelled, cancel := context.WithCancel(context.Background())
|
|
cancel()
|
|
if _, err := client.Send(cancelled, body); err == nil {
|
|
t.Fatal("cancelled/timeout request unexpectedly succeeded")
|
|
}
|
|
|
|
now := time.Now().UTC()
|
|
firstStore := bell_connector.PersistentReplayStore{DB: db}
|
|
if !firstStore.Consume("yv:bell:school-a", "abcdefghijklmnopqrstuv", now.Add(time.Minute), now) {
|
|
t.Fatal("first replay consume failed")
|
|
}
|
|
sqlDatabase, err := db.DB()
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if err = sqlDatabase.Close(); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
db = openDatabasePath(t, databasePath)
|
|
restartedStore := bell_connector.PersistentReplayStore{DB: db}
|
|
if restartedStore.Consume("yv:bell:school-a", "abcdefghijklmnopqrstuv", now.Add(time.Minute), now) {
|
|
t.Fatal("replay was accepted after store restart")
|
|
}
|
|
}
|
|
|
|
func TestBrainIngressAndBellEvidenceEndpoint(t *testing.T) {
|
|
gin.SetMode(gin.TestMode)
|
|
db, _ := openDatabase(t)
|
|
brainPublic, brainPrivate, _ := ed25519.GenerateKey(rand.Reader)
|
|
bellPublic, bellPrivate, _ := ed25519.GenerateKey(rand.Reader)
|
|
registry, err := machine_identity.NewRegistry(
|
|
machine_identity.KeyRecord{Principal: "yv:brain:school-a", KeyID: "brain-key-0001", PublicKey: brainPublic, Audience: "yovision-sense", Scopes: []string{"events:ingest"}, Enabled: true},
|
|
machine_identity.KeyRecord{Principal: "yv:bell:school-a", KeyID: "bell-key-0001", PublicKey: bellPublic, Audience: "yovision-sense", Scopes: []string{"evidence:read"}, Enabled: true},
|
|
)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
now := time.Date(2026, 8, 31, 1, 0, 0, 0, time.UTC)
|
|
verifier := machine_identity.Verifier{Registry: registry, Replay: bell_connector.PersistentReplayStore{DB: db}, Now: func() time.Time { return now }}
|
|
body := fixture(t)
|
|
brainSigner := machine_identity.Signer{Principal: "yv:brain:school-a", KeyID: "brain-key-0001", PrivateKey: brainPrivate, Now: func() time.Time { return now }}
|
|
token, err := brainSigner.Mint("yovision-sense", []string{"events:ingest"}, http.MethodPost, "/v1/events", body)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
request := httptest.NewRequest(http.MethodPost, "/v1/events", bytes.NewReader(body))
|
|
request.Header.Set("Authorization", "Bearer "+token)
|
|
request.Header.Set("X-Request-ID", "request-id-0000001")
|
|
response := httptest.NewRecorder()
|
|
ginContext, _ := gin.CreateTestContext(response)
|
|
ginContext.Request = request
|
|
(bell_connector.IngressHandler{DB: db, Verifier: verifier, EvidenceOwnerID: "sense-school-a", Now: func() time.Time { return now }}).Post(ginContext)
|
|
if response.Code != http.StatusAccepted {
|
|
t.Fatalf("Brain ingress status=%d body=%s", response.Code, response.Body.String())
|
|
}
|
|
var evidence bell_connector.EvidenceRecord
|
|
if err = db.First(&evidence, "evidence_id = ?", "ev-school-east-0001").Error; err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
bellSigner := machine_identity.Signer{Principal: "yv:bell:school-a", KeyID: "bell-key-0001", PrivateKey: bellPrivate, Now: func() time.Time { return now }}
|
|
path := "/v1/evidence/ev-school-east-0001"
|
|
token, err = bellSigner.Mint("yovision-sense", []string{"evidence:read"}, http.MethodGet, path, nil)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
getRequest := httptest.NewRequest(http.MethodGet, path, nil)
|
|
getRequest.Header.Set("Authorization", "Bearer "+token)
|
|
getRequest.Header.Set("X-Request-ID", "request-id-0000002")
|
|
getResponse := httptest.NewRecorder()
|
|
getContext, _ := gin.CreateTestContext(getResponse)
|
|
getContext.Request = getRequest
|
|
getContext.Params = gin.Params{{Key: "evidence_id", Value: "ev-school-east-0001"}}
|
|
(bell_connector.EvidenceHandler{DB: db, Verifier: verifier, Now: func() time.Time { return now }}).Get(getContext)
|
|
if getResponse.Code != http.StatusOK || !bytes.Equal(getResponse.Body.Bytes(), evidence.Payload) {
|
|
t.Fatalf("evidence lookup status=%d body=%s", getResponse.Code, getResponse.Body.String())
|
|
}
|
|
}
|
|
|
|
func openDatabase(t *testing.T) (*gorm.DB, string) {
|
|
databasePath := filepath.Join(t.TempDir(), "sense-bell-connector.sqlite")
|
|
return openDatabasePath(t, databasePath), databasePath
|
|
}
|
|
|
|
func openDatabasePath(t *testing.T, databasePath string) *gorm.DB {
|
|
t.Helper()
|
|
db, err := gorm.Open(sqlite.Open(databasePath), &gorm.Config{})
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if err = db.AutoMigrate(&outbox.Message{}, &outbox.DeliveryRecord{}, &outbox.Attempt{}, &bell_connector.ReplayToken{}, &bell_connector.InboundEvent{}, &bell_connector.EvidenceRecord{}); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
sqlDatabase, err := db.DB()
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
t.Cleanup(func() { _ = sqlDatabase.Close() })
|
|
return db
|
|
}
|
|
|
|
func fixture(t *testing.T) []byte {
|
|
t.Helper()
|
|
path := filepath.Join("..", "..", "..", "..", "contracts", "events", "v1", "examples", "dangerous-area.json")
|
|
body, err := os.ReadFile(path)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
var event map[string]any
|
|
if json.Unmarshal(body, &event) != nil {
|
|
t.Fatal("invalid fixture")
|
|
}
|
|
return body
|
|
}
|