diff --git a/internal/events/test/bus_contract_test.go b/internal/events/test/bus_contract_test.go index ebd1afd2..fdd7a195 100644 --- a/internal/events/test/bus_contract_test.go +++ b/internal/events/test/bus_contract_test.go @@ -4,6 +4,7 @@ package test // 只调导出方法,不碰 b.subs、不读源码。 import ( + "errors" "testing" "time" @@ -96,3 +97,88 @@ func TestSetGuardRoundTrip(t *testing.T) { t.Fatal("★ Guard() 没拿回 SetGuard 设置的守卫") } } + +// denyGuard 拒绝来自指定发布者的一切发布。 +type denyGuard struct { + denyPublisher string + scrubTo string +} + +func (g *denyGuard) AllowPublish(p events.Publisher, et events.EventType) (bool, string) { + if p.Name == g.denyPublisher { + return false, "policy forbids " + p.Name + } + return true, "" +} +func (g *denyGuard) AllowSubscribe(events.Publisher, events.EventType) (bool, string) { + return true, "" +} +func (g *denyGuard) ScrubPayload(et events.EventType, m map[string]interface{}) map[string]interface{} { + return m +} +func (g *denyGuard) ScrubInbound(et events.EventType, m map[string]interface{}) map[string]interface{} { + if g.scrubTo != "" { + for k := range m { + m[k] = g.scrubTo + } + } + return m +} + +// TestPublishAsReturnsDeniedErrorAndBlocksDelivery:被策略拒绝的发布必须 +// 返回可诊断错误,且订阅者收不到任何东西。 +func TestPublishAsReturnsDeniedErrorAndBlocksDelivery(t *testing.T) { + bus := events.NewBus() + bus.SetGuard(&denyGuard{denyPublisher: "bad-plugin"}) + + var delivered int + bus.Subscribe(events.EventType("contract.denied"), func(*events.Event) { delivered++ }) + + err := bus.PublishAs(events.Publisher{Name: "bad-plugin"}, + &events.Event{Type: events.EventType("contract.denied"), Payload: map[string]interface{}{"x": 1}}) + if err == nil { + t.Fatal("denied publish returned nil error") + } + var denied *events.DeniedError + if !errors.As(err, &denied) || denied.Publisher != "bad-plugin" { + t.Fatalf("error is not a diagnosable DeniedError: %v", err) + } + if delivered != 0 { + t.Fatalf("denied event still reached subscribers: %d", delivered) + } +} + +func TestSubscribersReceiveScrubbedPayload(t *testing.T) { + bus := events.NewBus() + bus.SetGuard(&denyGuard{scrubTo: "[redacted]"}) + got := make(chan map[string]interface{}, 1) + bus.Subscribe(events.EventType("contract.scrub"), func(e *events.Event) { got <- e.Payload }) + + if err := bus.PublishAs(events.Publisher{Name: "p"}, + &events.Event{Type: events.EventType("contract.scrub"), + Payload: map[string]interface{}{"token": "SECRET", "x": 1}}); err != nil { + t.Fatalf("publish: %v", err) + } + select { + case p := <-got: + for k, v := range p { + if v == "SECRET" { + t.Fatalf("plaintext secret reached subscriber: %v", p) + } + _ = k + } + case <-time.After(2 * time.Second): + t.Fatal("subscriber never received the event") + } +} + +func TestHandlerPanicDoesNotBreakOtherSubscribers(t *testing.T) { + bus := events.NewBus() + var ran int + bus.Subscribe(events.EventType("contract.panic"), func(*events.Event) { panic("boom") }) + bus.Subscribe(events.EventType("contract.panic"), func(*events.Event) { ran++ }) + bus.Publish(&events.Event{Type: events.EventType("contract.panic")}) + if ran != 1 { + t.Fatalf("healthy subscriber skipped after panic: ran=%d", ran) + } +}