114 lines
3.4 KiB
Go
114 lines
3.4 KiB
Go
package event
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"fmt"
|
|
"os"
|
|
"sync"
|
|
"testing"
|
|
"time"
|
|
|
|
"git.ilapage.cn/ila/yovision/Bell/server/internal/platform"
|
|
"git.ilapage.cn/ila/yovision/Bell/server/migrations"
|
|
"github.com/jackc/pgx/v5"
|
|
)
|
|
|
|
func TestNormalizeRejectsUnsafeEvidence(t *testing.T) {
|
|
value := `file:C:\camera\secret.mp4`
|
|
_, _, err := normalize(Command{ProducerID: "test", SourceEventID: "1", EventType: "intrusion", OccurredAt: time.Now(), Location: "gate", Severity: "high", EvidenceRef: &value})
|
|
if !errors.Is(err, ErrInvalid) {
|
|
t.Fatalf("expected invalid evidence reference, got %v", err)
|
|
}
|
|
}
|
|
|
|
func TestPostgresIdempotencyAndImmutability(t *testing.T) {
|
|
url := os.Getenv("BELL_TEST_DATABASE_URL")
|
|
if url == "" {
|
|
t.Skip("BELL_TEST_DATABASE_URL is not set")
|
|
}
|
|
ctx := context.Background()
|
|
db, err := platform.OpenDatabase(ctx, url)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
defer db.Close()
|
|
if err := migrations.Apply(ctx, db); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
service := Service{DB: db}
|
|
key := fmt.Sprintf("event-test-%d", time.Now().UnixNano())
|
|
command := Command{ProducerID: "integration-test", SourceEventID: key, EventType: "danger-zone", OccurredAt: time.Now().UTC().Truncate(time.Millisecond), Location: "test-zone", Severity: "high", Attributes: map[string]any{"track_id": "anonymous-1"}}
|
|
const count = 16
|
|
results := make(chan Result, count)
|
|
errorsFound := make(chan error, count)
|
|
var wg sync.WaitGroup
|
|
for range count {
|
|
wg.Add(1)
|
|
go func() {
|
|
defer wg.Done()
|
|
result, err := service.Ingest(ctx, command)
|
|
if err != nil {
|
|
errorsFound <- err
|
|
return
|
|
}
|
|
results <- result
|
|
}()
|
|
}
|
|
wg.Wait()
|
|
close(results)
|
|
close(errorsFound)
|
|
for err := range errorsFound {
|
|
t.Errorf("concurrent ingest: %v", err)
|
|
}
|
|
var eventID, receiptID string
|
|
seen := 0
|
|
for result := range results {
|
|
seen++
|
|
if eventID == "" {
|
|
eventID = result.Event.ID
|
|
receiptID = result.Receipt.ID
|
|
}
|
|
if result.Event.ID != eventID || result.Receipt.ID != receiptID {
|
|
t.Errorf("unstable identities: %#v", result)
|
|
}
|
|
}
|
|
if seen != count {
|
|
t.Fatalf("received %d results, want %d", seen, count)
|
|
}
|
|
conflict := command
|
|
conflict.Location = "different"
|
|
if _, err := service.Ingest(ctx, conflict); !errors.Is(err, ErrConflict) {
|
|
t.Fatalf("expected conflict, got %v", err)
|
|
}
|
|
if _, err := db.Exec(ctx, `UPDATE bell_events SET location='tampered' WHERE id=$1`, eventID); err == nil {
|
|
t.Fatal("immutable Event accepted update")
|
|
}
|
|
db.Close()
|
|
reopened, err := platform.OpenDatabase(ctx, url)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
defer reopened.Close()
|
|
afterRestart, err := (Service{DB: reopened}).Ingest(ctx, command)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if afterRestart.Event.ID != eventID || afterRestart.Receipt.ID != receiptID || !afterRestart.Duplicate {
|
|
t.Fatalf("receipt changed after restart: %#v", afterRestart)
|
|
}
|
|
failing := command
|
|
failing.SourceEventID = key + "-rollback"
|
|
failedService := Service{DB: reopened, AfterPersist: func(context.Context, pgx.Tx, Event) error { return errors.New("forced matching failure") }}
|
|
if _, err := failedService.Ingest(ctx, failing); err == nil {
|
|
t.Fatal("expected hook failure")
|
|
}
|
|
var persisted int
|
|
if err := reopened.QueryRow(ctx, `SELECT count(*) FROM bell_events WHERE producer_id=$1 AND source_event_id=$2`, failing.ProducerID, failing.SourceEventID).Scan(&persisted); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if persisted != 0 {
|
|
t.Fatal("Event committed without atomic match")
|
|
}
|
|
}
|