package event_test import ( "context" "sync" "sync/atomic" "testing" "time" "github.com/yaoapp/yao/event" "github.com/yaoapp/yao/event/types" ) // --- Test handler --- type recordHandler struct { mu sync.Mutex calls []string // records ev.Type for each Handle call shutdown bool } func (h *recordHandler) Handle(ctx context.Context, ev *types.Event, resp chan<- types.Result) { h.mu.Lock() h.calls = append(h.calls, ev.Type) h.mu.Unlock() if ev.IsCall { var p string if err := ev.Should(&p); err == nil { resp <- types.Result{Data: "echo:" + p} } else { resp <- types.Result{Data: "echo:" + ev.Type} } } } func (h *recordHandler) Shutdown(ctx context.Context) error { h.mu.Lock() defer h.mu.Unlock() h.shutdown = true return nil } func (h *recordHandler) getCalls() []string { h.mu.Lock() defer h.mu.Unlock() cp := make([]string, len(h.calls)) copy(cp, h.calls) return cp } // --- Phase 3: Push / Call basic routing (no queue) --- func TestPush_NoQueue(t *testing.T) { event.Reset() defer event.Reset() h := &recordHandler{} event.Register("foo", h) if err := event.Start(); err != nil { t.Fatal(err) } defer func() { _ = event.Stop(context.Background()) }() id, err := event.Push(context.Background(), "foo.bar", "payload1") if err != nil { t.Fatalf("Push failed: %v", err) } if id == "" { t.Fatal("expected non-empty event ID") } // Wait for async handler time.Sleep(50 * time.Millisecond) calls := h.getCalls() if len(calls) != 1 || calls[0] != "foo.bar" { t.Fatalf("expected [foo.bar], got %v", calls) } } func TestCall_NoQueue(t *testing.T) { event.Reset() defer event.Reset() h := &recordHandler{} event.Register("foo", h) _ = event.Start() defer func() { _ = event.Stop(context.Background()) }() id, data, err := event.Call(context.Background(), "foo.get", "hello") if err != nil { t.Fatalf("Call failed: %v", err) } if id == "" { t.Fatal("expected non-empty event ID") } if data != "echo:hello" { t.Fatalf("expected echo:hello, got %v", data) } } func TestPush_UnregisteredPrefix(t *testing.T) { event.Reset() defer event.Reset() _ = event.Start() defer func() { _ = event.Stop(context.Background()) }() _, err := event.Push(context.Background(), "unknown.thing", nil) if err != event.ErrNoHandler { t.Fatalf("expected ErrNoHandler, got %v", err) } } func TestPush_NotStarted(t *testing.T) { event.Reset() defer event.Reset() event.Register("foo", &recordHandler{}) _, err := event.Push(context.Background(), "foo.bar", nil) if err != event.ErrNotStarted { t.Fatalf("expected ErrNotStarted, got %v", err) } } func TestPush_SIDAndAuth(t *testing.T) { event.Reset() defer event.Reset() var captured *types.Event var mu sync.Mutex h := &captureHandler{onHandle: func(ev *types.Event) { mu.Lock() captured = ev mu.Unlock() }} event.Register("foo", h) _ = event.Start() defer func() { _ = event.Stop(context.Background()) }() ctx := event.WithSID(context.Background(), "sess-abc") ctx = event.WithAuth(ctx, &types.AuthorizedInfo{UserID: "u-1"}) _, err := event.Push(ctx, "foo.bar", "data") if err != nil { t.Fatalf("Push failed: %v", err) } time.Sleep(50 * time.Millisecond) mu.Lock() defer mu.Unlock() if captured == nil { t.Fatal("handler was not called") } if captured.SID != "sess-abc" { t.Fatalf("expected SID sess-abc, got %s", captured.SID) } if captured.Auth == nil || captured.Auth.UserID != "u-1" { t.Fatalf("expected Auth.UserID u-1, got %+v", captured.Auth) } } // captureHandler captures the event for inspection. type captureHandler struct { onHandle func(*types.Event) } func (h *captureHandler) Handle(ctx context.Context, ev *types.Event, resp chan<- types.Result) { if h.onHandle != nil { h.onHandle(ev) } if ev.IsCall { resp <- types.Result{Data: "ok"} } } func (h *captureHandler) Shutdown(ctx context.Context) error { return nil } // --- Coverage: prefixOf without dot --- func TestPush_TypeWithoutDot(t *testing.T) { event.Reset() defer event.Reset() h := &recordHandler{} event.Register("nodot", h) _ = event.Start() defer func() { _ = event.Stop(context.Background()) }() id, err := event.Push(context.Background(), "nodot", "payload") if err != nil { t.Fatalf("Push failed: %v", err) } if id == "" { t.Fatal("expected non-empty event ID") } time.Sleep(50 * time.Millisecond) calls := h.getCalls() if len(calls) != 1 || calls[0] != "nodot" { t.Fatalf("expected [nodot], got %v", calls) } } // --- Coverage: Call unregistered prefix --- func TestCall_UnregisteredPrefix(t *testing.T) { event.Reset() defer event.Reset() _ = event.Start() defer func() { _ = event.Stop(context.Background()) }() _, _, err := event.Call(context.Background(), "unknown.thing", nil) if err != event.ErrNoHandler { t.Fatalf("expected ErrNoHandler, got %v", err) } } // --- Coverage: Call with queue (happy path) --- func TestCall_WithQueue(t *testing.T) { event.Reset() defer event.Reset() h := &recordHandler{} event.Register("foo", h) _ = event.Start() defer func() { _ = event.Stop(context.Background()) }() qID, err := event.QueueCreate("foo") if err != nil { t.Fatalf("QueueCreate failed: %v", err) } defer event.QueueRelease(qID) id, data, err := event.Call(context.Background(), "foo.get", "hello", event.Queue(qID)) if err != nil { t.Fatalf("Call with queue failed: %v", err) } if id == "" { t.Fatal("expected non-empty event ID") } if data != "echo:hello" { t.Fatalf("expected echo:hello, got %v", data) } } // --- Coverage: Call with non-existent queue --- func TestCall_QueueNotFound(t *testing.T) { event.Reset() defer event.Reset() event.Register("foo", &recordHandler{}) _ = event.Start() defer func() { _ = event.Stop(context.Background()) }() _, _, err := event.Call(context.Background(), "foo.get", nil, event.Queue("no-such-queue")) if err != event.ErrQueueNotFound { t.Fatalf("expected ErrQueueNotFound, got %v", err) } } // --- Coverage: Call ctx timeout --- func TestCall_CtxTimeout(t *testing.T) { event.Reset() defer event.Reset() h := &captureHandler{onHandle: func(ev *types.Event) { time.Sleep(500 * time.Millisecond) }} event.Register("slow", h) _ = event.Start() defer func() { _ = event.Stop(context.Background()) }() ctx, cancel := context.WithTimeout(context.Background(), 20*time.Millisecond) defer cancel() _, _, err := event.Call(ctx, "slow.op", nil) if err == nil { t.Fatal("expected timeout error") } } // --- Coverage: Call no-queue dispatch failure (ctx cancelled) --- func TestCall_NoQueue_DispatchFail(t *testing.T) { event.Reset() defer event.Reset() h := &concurrencyHandler{ peak: &atomic.Int32{}, current: &atomic.Int32{}, delay: 200 * time.Millisecond, } event.Register("tiny", h, event.MaxWorkers(1), event.ReservedWorkers(0)) _ = event.Start() // Saturate the single total slot with a Call in background bgDone := make(chan struct{}) go func() { defer close(bgDone) _, _, _ = event.Call(context.Background(), "tiny.work", nil) }() time.Sleep(10 * time.Millisecond) // Another Call with already-cancelled context should fail at dispatch ctx, cancel := context.WithCancel(context.Background()) cancel() _, _, err := event.Call(ctx, "tiny.op", nil) if err == nil { t.Fatal("expected error for cancelled ctx call") } // Wait for background goroutine to finish before Stop <-bgDone _ = event.Stop(context.Background()) }