405 lines
17 KiB
Go
405 lines
17 KiB
Go
package event_ingress_test
|
|
|
|
import (
|
|
"bytes"
|
|
"context"
|
|
"crypto/ed25519"
|
|
"crypto/rand"
|
|
"encoding/base64"
|
|
"encoding/json"
|
|
"io"
|
|
"net/http"
|
|
"net/http/httptest"
|
|
"os"
|
|
"path/filepath"
|
|
"strings"
|
|
"sync"
|
|
"sync/atomic"
|
|
"testing"
|
|
"time"
|
|
|
|
"github.com/gin-gonic/gin"
|
|
"github.com/go-admin-team/go-admin-core/sdk"
|
|
"gorm.io/driver/postgres"
|
|
"gorm.io/driver/sqlite"
|
|
"gorm.io/gorm"
|
|
|
|
"go-admin/app/bell/event"
|
|
"go-admin/app/bell/integration/event_ingress"
|
|
"go-admin/app/bell/integration/machine_identity"
|
|
"go-admin/app/bell/receipt"
|
|
)
|
|
|
|
func TestPostgresConcurrentBusinessAndSecurityIdempotency(t *testing.T) {
|
|
dsn := os.Getenv("BELL_EVENT_INGRESS_TEST_DATABASE_URL")
|
|
if dsn == "" {
|
|
t.Skip("set BELL_EVENT_INGRESS_TEST_DATABASE_URL to run PostgreSQL concurrency verification")
|
|
}
|
|
db, err := gorm.Open(postgres.Open(dsn), &gorm.Config{})
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if err = db.AutoMigrate(&event.Event{}, &receipt.Receipt{}, &receipt.IngestAudit{}, &event_ingress.ReplayToken{}, &event_ingress.EvidenceStatus{}); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
parsed, err := event_ingress.ParseEvent(fixture(t, "dangerous-area.json"))
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
const workers = 12
|
|
var created, duplicate, failures atomic.Int32
|
|
var wait sync.WaitGroup
|
|
for range workers {
|
|
wait.Add(1)
|
|
go func() {
|
|
defer wait.Done()
|
|
result, ingestErr := (event_ingress.Service{DB: db}).Ingest(context.Background(), parsed)
|
|
if ingestErr != nil {
|
|
failures.Add(1)
|
|
return
|
|
}
|
|
if result.Disposition == "created" {
|
|
created.Add(1)
|
|
} else if result.Disposition == "duplicate" {
|
|
duplicate.Add(1)
|
|
}
|
|
}()
|
|
}
|
|
wait.Wait()
|
|
if created.Load() != 1 || duplicate.Load() != workers-1 || failures.Load() != 0 {
|
|
t.Fatalf("concurrent ingest created=%d duplicate=%d failures=%d", created.Load(), duplicate.Load(), failures.Load())
|
|
}
|
|
|
|
now := time.Now().UTC()
|
|
var consumed atomic.Int32
|
|
for range workers {
|
|
wait.Add(1)
|
|
go func() {
|
|
defer wait.Done()
|
|
if (event_ingress.PersistentReplayStore{DB: db}).Consume("yv:sense:school-a", "concurrent-token-id-0001", now.Add(time.Minute), now) {
|
|
consumed.Add(1)
|
|
}
|
|
}()
|
|
}
|
|
wait.Wait()
|
|
if consumed.Load() != 1 {
|
|
t.Fatalf("concurrent replay consume accepted %d requests", consumed.Load())
|
|
}
|
|
}
|
|
|
|
func TestRuntimeRegistrationIsOptionalAndMigrationGated(t *testing.T) {
|
|
gin.SetMode(gin.TestMode)
|
|
t.Setenv("BELL_EVENT_INGRESS_ENABLED", "")
|
|
disabled := gin.New()
|
|
if err := event_ingress.RegisterRuntime(disabled); err != nil || len(disabled.Routes()) != 0 {
|
|
t.Fatalf("disabled runtime err=%v routes=%#v", err, disabled.Routes())
|
|
}
|
|
|
|
db, err := gorm.Open(sqlite.Open("file:bell-runtime?mode=memory&cache=shared"), &gorm.Config{})
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
sdk.Runtime.SetDb("", db)
|
|
t.Cleanup(func() { sdk.Runtime.SetDb("", nil) })
|
|
t.Setenv("BELL_EVENT_INGRESS_ENABLED", "true")
|
|
if err = event_ingress.RegisterRuntime(gin.New()); err == nil {
|
|
t.Fatal("enabled runtime started without formal migration")
|
|
}
|
|
if err = db.AutoMigrate(&event_ingress.ReplayToken{}, &event_ingress.EvidenceStatus{}); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
registryPath := writeRegistry(t, "yovision-bell", "yv:sense:school-a", "sense-key-0001")
|
|
t.Setenv("BELL_MACHINE_PRINCIPAL_REGISTRY", registryPath)
|
|
registered := gin.New()
|
|
if err = event_ingress.RegisterRuntime(registered); err != nil {
|
|
t.Fatalf("enabled runtime did not register after migration: %v", err)
|
|
}
|
|
routes := registered.Routes()
|
|
if len(routes) != 1 || routes[0].Method != http.MethodPost || routes[0].Path != "/v1/events" {
|
|
t.Fatalf("unexpected ingress routes: %#v", routes)
|
|
}
|
|
}
|
|
|
|
func TestContractFixtureIdempotencyConflictAndReplayPersistence(t *testing.T) {
|
|
gin.SetMode(gin.TestMode)
|
|
databasePath := filepath.Join(t.TempDir(), "bell-ingress.sqlite")
|
|
db := openDatabasePath(t, databasePath)
|
|
body := fixture(t, "dangerous-area.json")
|
|
publicKey, privateKey, err := ed25519.GenerateKey(rand.Reader)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
registry, err := machine_identity.NewRegistry(machine_identity.KeyRecord{Principal: "yv:sense:school-a", KeyID: "sense-key-0001", PublicKey: publicKey, Audience: "yovision-bell", Scopes: []string{"events:ingest"}, Enabled: true})
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
now := time.Date(2026, 8, 31, 1, 0, 0, 0, time.UTC)
|
|
signer := machine_identity.Signer{Principal: "yv:sense:school-a", KeyID: "sense-key-0001", PrivateKey: privateKey, Now: func() time.Time { return now }}
|
|
newHandler := func() event_ingress.Handler {
|
|
return event_ingress.Handler{DB: db, Enabled: true, Verifier: machine_identity.Verifier{Registry: registry, Replay: event_ingress.PersistentReplayStore{DB: db}, Now: func() time.Time { return now }}}
|
|
}
|
|
|
|
firstToken := mint(t, signer, body)
|
|
first := request(t, newHandler(), body, firstToken)
|
|
if first.Code != http.StatusCreated {
|
|
t.Fatalf("first ingest status=%d body=%s", first.Code, first.Body.String())
|
|
}
|
|
var created event_ingress.IngestResult
|
|
decode(t, first, &created)
|
|
if created.Disposition != "created" || created.PayloadSHA256 != "4cc1e93820195caf713ea675ff33f178c9d4997dd8a81cb61287e9fea0e3d5e1" {
|
|
t.Fatalf("unexpected created result: %+v", created)
|
|
}
|
|
sqlDatabase, err := db.DB()
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if err = sqlDatabase.Close(); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
db = openDatabasePath(t, databasePath)
|
|
|
|
// A new process-local Handler and replay store still reject the old token,
|
|
// proving that security replay state is durable rather than in-memory.
|
|
replayedToken := request(t, newHandler(), body, firstToken)
|
|
if replayedToken.Code != http.StatusUnauthorized || !strings.Contains(replayedToken.Body.String(), "machine_token_replayed") {
|
|
t.Fatalf("token replay status=%d body=%s", replayedToken.Code, replayedToken.Body.String())
|
|
}
|
|
|
|
duplicate := request(t, newHandler(), body, mint(t, signer, body))
|
|
if duplicate.Code != http.StatusOK {
|
|
t.Fatalf("business duplicate status=%d body=%s", duplicate.Code, duplicate.Body.String())
|
|
}
|
|
var duplicateResult event_ingress.IngestResult
|
|
decode(t, duplicate, &duplicateResult)
|
|
if duplicateResult.Disposition != "duplicate" || duplicateResult.EventID != created.EventID {
|
|
t.Fatalf("duplicate did not retain event identity: %+v", duplicateResult)
|
|
}
|
|
numericVariant := bytes.Replace(body, []byte(`0.93`), []byte(`0.930`), 1)
|
|
numericDuplicate := request(t, newHandler(), numericVariant, mint(t, signer, numericVariant))
|
|
if numericDuplicate.Code != http.StatusOK || !strings.Contains(numericDuplicate.Body.String(), created.PayloadSHA256) {
|
|
t.Fatalf("JCS-equivalent numeric payload was not a duplicate: %d %s", numericDuplicate.Code, numericDuplicate.Body.String())
|
|
}
|
|
|
|
var changed map[string]any
|
|
if err = json.Unmarshal(body, &changed); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
changed["severity"] = "critical"
|
|
conflicting, _ := json.Marshal(changed)
|
|
conflict := request(t, newHandler(), conflicting, mint(t, signer, conflicting))
|
|
if conflict.Code != http.StatusConflict || !strings.Contains(conflict.Body.String(), "idempotency_conflict") || !strings.Contains(conflict.Body.String(), created.EventID) {
|
|
t.Fatalf("conflict status=%d body=%s", conflict.Code, conflict.Body.String())
|
|
}
|
|
assertCount(t, db, &event.Event{}, 1)
|
|
assertCount(t, db, &receipt.Receipt{}, 1)
|
|
assertCount(t, db, &receipt.IngestAudit{}, 4)
|
|
}
|
|
|
|
func TestEvidenceDegradationIdentityErrorsAndDisabledConnector(t *testing.T) {
|
|
db := openDatabase(t)
|
|
publicKey, privateKey, _ := ed25519.GenerateKey(rand.Reader)
|
|
registry, _ := machine_identity.NewRegistry(machine_identity.KeyRecord{Principal: "yv:brain:school-a", KeyID: "brain-key-0001", PublicKey: publicKey, Audience: "yovision-bell", Scopes: []string{"events:ingest"}, Enabled: true})
|
|
now := time.Date(2026, 8, 31, 1, 0, 0, 0, time.UTC)
|
|
signer := machine_identity.Signer{Principal: "yv:brain:school-a", KeyID: "brain-key-0001", PrivateKey: privateKey, Now: func() time.Time { return now }}
|
|
handler := event_ingress.Handler{DB: db, Enabled: true, Verifier: machine_identity.Verifier{Registry: registry, Replay: event_ingress.PersistentReplayStore{DB: db}, Now: func() time.Time { return now }}}
|
|
|
|
pending := fixture(t, "dangerous-area.json")
|
|
if response := requestWithID(t, handler, pending, mint(t, signer, pending), "short"); response.Code != http.StatusBadRequest || !strings.Contains(response.Body.String(), "invalid_request_id") {
|
|
t.Fatalf("invalid request id status=%d body=%s", response.Code, response.Body.String())
|
|
}
|
|
missingRequestID := requestWithID(t, handler, pending, mint(t, signer, pending), "")
|
|
if missingRequestID.Code != http.StatusCreated || !requestIDPatternForTest(missingRequestID.Header().Get("X-Request-ID")) {
|
|
t.Fatalf("trusted hop did not create a request id: %d %s", missingRequestID.Code, missingRequestID.Body.String())
|
|
}
|
|
queryResponse := requestTarget(t, handler, pending, mint(t, signer, pending), "/v1/events?debug=true")
|
|
if queryResponse.Code != http.StatusBadRequest || !strings.Contains(queryResponse.Body.String(), "invalid_request_target") {
|
|
t.Fatalf("query target was accepted: %d %s", queryResponse.Code, queryResponse.Body.String())
|
|
}
|
|
if response := request(t, handler, pending, mint(t, signer, pending)); response.Code != http.StatusOK {
|
|
t.Fatalf("pending evidence rejected: %d %s", response.Code, response.Body.String())
|
|
}
|
|
failed := fixture(t, "directional-line-crossed.json")
|
|
if response := request(t, handler, failed, mint(t, signer, failed)); response.Code != http.StatusCreated {
|
|
t.Fatalf("failed evidence rejected: %d %s", response.Code, response.Body.String())
|
|
}
|
|
assertCount(t, db, &event.Event{}, 2)
|
|
|
|
wrongAudienceToken, err := signer.Mint("yovision-sense", []string{"events:ingest"}, http.MethodPost, "/v1/events", pending)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if response := request(t, handler, pending, wrongAudienceToken); response.Code != http.StatusForbidden {
|
|
t.Fatalf("wrong audience was not forbidden: %d %s", response.Code, response.Body.String())
|
|
}
|
|
if response := request(t, event_ingress.Handler{Enabled: false}, pending, "none"); response.Code != http.StatusServiceUnavailable {
|
|
t.Fatalf("disabled connector status=%d", response.Code)
|
|
}
|
|
}
|
|
|
|
func TestEvidenceResolverCurrentMissingExpiredAndTimeout(t *testing.T) {
|
|
db := openDatabase(t)
|
|
now := time.Date(2026, 8, 31, 1, 0, 0, 0, time.UTC)
|
|
status := event_ingress.EvidenceStatus{EventID: "event-1", EvidenceID: "ev-school-east-0001", OwnerID: "sense-school-a", Status: "pending", Resolution: "snapshot", CurrentPayload: json.RawMessage(`{"schema_version":"yovision.evidence-reference/v1","evidence_id":"ev-school-east-0001","owner_id":"sense-school-a","type":"snapshot","status":"pending","captured_at":"2026-08-31T00:00:01.125Z","status_updated_at":"2026-08-31T00:00:01.125Z"}`), CreatedAt: now, UpdatedAt: now}
|
|
if err := db.Create(&status).Error; err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
_, privateKey, _ := ed25519.GenerateKey(rand.Reader)
|
|
signer := machine_identity.Signer{Principal: "yv:bell:school-a", KeyID: "bell-key-0001", PrivateKey: privateKey, Now: func() time.Time { return now }}
|
|
response := func(code int, body string) *http.Response {
|
|
return &http.Response{StatusCode: code, Body: io.NopCloser(strings.NewReader(body)), Header: make(http.Header)}
|
|
}
|
|
client := event_ingress.EvidenceClient{Endpoint: "https://sense.example", Signer: signer, HTTP: doFunc(func(*http.Request) (*http.Response, error) {
|
|
return response(http.StatusNotFound, `{}`), nil
|
|
})}
|
|
if err := client.Refresh(context.Background(), db, status); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if err := db.First(&status, "event_id = ? AND evidence_id = ?", "event-1", "ev-school-east-0001").Error; err != nil || status.Resolution != "unavailable" || status.LastError != "evidence_not_found" {
|
|
t.Fatalf("missing resolution=%s error=%s db=%v", status.Resolution, status.LastError, err)
|
|
}
|
|
client.HTTP = doFunc(func(*http.Request) (*http.Response, error) { return response(http.StatusGone, `{}`), nil })
|
|
if err := client.Refresh(context.Background(), db, status); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if err := db.First(&status, "event_id = ? AND evidence_id = ?", "event-1", "ev-school-east-0001").Error; err != nil || status.Resolution != "expired" {
|
|
t.Fatalf("expired resolution=%s db=%v", status.Resolution, err)
|
|
}
|
|
current := `{"schema_version":"yovision.evidence-reference/v1","evidence_id":"ev-school-east-0001","owner_id":"sense-school-a","type":"snapshot","status":"success","captured_at":"2026-08-31T00:00:01.125Z","status_updated_at":"2026-08-31T00:00:02.125Z","content_type":"image/jpeg","integrity":{"algorithm":"sha256","digest":"aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa","size_bytes":1}}`
|
|
client.HTTP = doFunc(func(*http.Request) (*http.Response, error) { return response(http.StatusOK, current), nil })
|
|
if err := client.Refresh(context.Background(), db, status); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if err := db.First(&status, "event_id = ? AND evidence_id = ?", "event-1", "ev-school-east-0001").Error; err != nil || status.Resolution != "current" || status.Status != "success" || status.LastError != "" {
|
|
t.Fatalf("current status=%s resolution=%s error=%s db=%v", status.Status, status.Resolution, status.LastError, err)
|
|
}
|
|
client.HTTP = doFunc(func(*http.Request) (*http.Response, error) { return nil, context.DeadlineExceeded })
|
|
if err := client.Refresh(context.Background(), db, status); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if err := db.First(&status, "event_id = ? AND evidence_id = ?", "event-1", "ev-school-east-0001").Error; err != nil || status.Resolution != "unavailable" || status.LastError != "evidence_timeout" {
|
|
t.Fatalf("timeout resolution=%s error=%s db=%v", status.Resolution, status.LastError, err)
|
|
}
|
|
}
|
|
|
|
func openDatabase(t *testing.T) *gorm.DB {
|
|
return openDatabasePath(t, filepath.Join(t.TempDir(), "bell-ingress.sqlite"))
|
|
}
|
|
|
|
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(&event.Event{}, &receipt.Receipt{}, &receipt.IngestAudit{}, &event_ingress.ReplayToken{}, &event_ingress.EvidenceStatus{}); 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, name string) []byte {
|
|
t.Helper()
|
|
path := filepath.Join("..", "..", "..", "..", "..", "contracts", "events", "v1", "examples", name)
|
|
body, err := os.ReadFile(path)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
return body
|
|
}
|
|
|
|
func writeRegistry(t *testing.T, audience, principal, keyID string) string {
|
|
t.Helper()
|
|
publicKey, _, err := ed25519.GenerateKey(rand.Reader)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
document := map[string]any{
|
|
"version": "yovision.machine-principal-registry/v1", "audience": audience,
|
|
"principals": []any{map[string]any{
|
|
"principal_id": principal, "enabled": true,
|
|
"keys": []any{map[string]any{
|
|
"kid": keyID, "public_key_base64url": base64.RawURLEncoding.EncodeToString(publicKey),
|
|
"status": "active", "scopes": []string{"events:ingest"},
|
|
}},
|
|
}},
|
|
}
|
|
encoded, err := json.Marshal(document)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
path := filepath.Join(t.TempDir(), "registry.json")
|
|
if err = os.WriteFile(path, encoded, 0o600); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
return path
|
|
}
|
|
|
|
func mint(t *testing.T, signer machine_identity.Signer, body []byte) string {
|
|
t.Helper()
|
|
token, err := signer.Mint("yovision-bell", []string{"events:ingest"}, http.MethodPost, "/v1/events", body)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
return token
|
|
}
|
|
|
|
func request(t *testing.T, handler event_ingress.Handler, body []byte, token string) *httptest.ResponseRecorder {
|
|
return requestWithID(t, handler, body, token, "request-id-0000001")
|
|
}
|
|
|
|
func requestWithID(t *testing.T, handler event_ingress.Handler, body []byte, token, requestID string) *httptest.ResponseRecorder {
|
|
return requestTargetWithID(t, handler, body, token, "/v1/events", requestID)
|
|
}
|
|
|
|
func requestTarget(t *testing.T, handler event_ingress.Handler, body []byte, token, target string) *httptest.ResponseRecorder {
|
|
return requestTargetWithID(t, handler, body, token, target, "request-id-0000001")
|
|
}
|
|
|
|
func requestTargetWithID(t *testing.T, handler event_ingress.Handler, body []byte, token, target, requestID string) *httptest.ResponseRecorder {
|
|
t.Helper()
|
|
request := httptest.NewRequest(http.MethodPost, target, bytes.NewReader(body))
|
|
request.Header.Set("Authorization", "Bearer "+token)
|
|
request.Header.Set("X-Request-ID", requestID)
|
|
response := httptest.NewRecorder()
|
|
context, _ := gin.CreateTestContext(response)
|
|
context.Request = request
|
|
handler.Post(context)
|
|
return response
|
|
}
|
|
|
|
func requestIDPatternForTest(value string) bool {
|
|
if len(value) < 16 || len(value) > 128 {
|
|
return false
|
|
}
|
|
for index, r := range value {
|
|
if !(r >= 'A' && r <= 'Z' || r >= 'a' && r <= 'z' || r >= '0' && r <= '9' || index > 0 && strings.ContainsRune("._:-", r)) {
|
|
return false
|
|
}
|
|
}
|
|
return true
|
|
}
|
|
|
|
func decode(t *testing.T, response *httptest.ResponseRecorder, target any) {
|
|
t.Helper()
|
|
if err := json.Unmarshal(response.Body.Bytes(), target); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
}
|
|
|
|
func assertCount(t *testing.T, db *gorm.DB, model any, expected int64) {
|
|
t.Helper()
|
|
var count int64
|
|
if err := db.Model(model).Count(&count).Error; err != nil || count != expected {
|
|
t.Fatalf("count %T=%d expected=%d err=%v", model, count, expected, err)
|
|
}
|
|
}
|
|
|
|
type doFunc func(*http.Request) (*http.Response, error)
|
|
|
|
func (function doFunc) Do(request *http.Request) (*http.Response, error) { return function(request) }
|