diff --git a/machinery/go.mod b/machinery/go.mod index 353bf3d..2630af7 100644 --- a/machinery/go.mod +++ b/machinery/go.mod @@ -32,6 +32,7 @@ require ( github.com/pion/rtp v1.8.19 github.com/pion/webrtc/v4 v4.1.2 github.com/sirupsen/logrus v1.9.3 + github.com/stretchr/testify v1.10.0 github.com/swaggo/files v1.0.1 github.com/swaggo/gin-swagger v1.6.0 github.com/swaggo/swag v1.16.4 @@ -58,6 +59,7 @@ require ( github.com/clbanning/mxj v1.8.4 // indirect github.com/clbanning/mxj/v2 v2.7.0 // indirect github.com/cloudwego/base64x v0.1.5 // indirect + github.com/davecgh/go-spew v1.1.1 // indirect github.com/elastic/go-windows v1.0.2 // indirect github.com/elgs/gostrgen v0.0.0-20161222160715-9d61ae07eeae // indirect github.com/erikstmartin/go-testdb v0.0.0-20160219214506-8d10e4a1bae5 // indirect @@ -108,6 +110,7 @@ require ( github.com/pion/stun/v3 v3.0.0 // indirect github.com/pion/transport/v3 v3.0.7 // indirect github.com/pion/turn/v4 v4.0.0 // indirect + github.com/pmezard/go-difflib v1.0.0 // indirect github.com/prometheus/procfs v0.15.1 // indirect github.com/twitchyliquid64/golang-asm v0.15.1 // indirect github.com/ugorji/go/codec v1.2.12 // indirect diff --git a/machinery/src/components/kerberos.go b/machinery/src/components/kerberos.go index 350966b..5637824 100644 --- a/machinery/src/components/kerberos.go +++ b/machinery/src/components/kerberos.go @@ -341,6 +341,11 @@ func RunAgent(configDirectory string, configuration *models.Configuration, commu communication.HandleONVIF = make(chan models.OnvifAction, 10) go onvif.HandleONVIFActions(configuration, communication) + // Handle ONVIF event stream — opt-in via Capture.ONVIFMotion="true". + // Stops when the agent's shared context is cancelled. The function + // is a no-op if ONVIFMotion is not enabled. + go onvif.HandleONVIFEventStream(*communication.Context, configuration, communication) + communication.HandleAudio = make(chan models.AudioDataPartial, 10) if rtspBackChannelClient.HasBackChannel { communication.HasBackChannel = true diff --git a/machinery/src/config/main.go b/machinery/src/config/main.go index 3b93828..b5fad8b 100644 --- a/machinery/src/config/main.go +++ b/machinery/src/config/main.go @@ -377,6 +377,9 @@ func applyAgentEnvVars(configuration *models.Configuration, prefix string, apply case "AGENT_CAPTURE_MOTION": configuration.Config.Capture.Motion = value break + case "AGENT_CAPTURE_ONVIF_MOTION": + configuration.Config.Capture.ONVIFMotion = value + break case "AGENT_CAPTURE_SNAPSHOTS": configuration.Config.Capture.Snapshots = value break diff --git a/machinery/src/models/config.go b/machinery/src/models/config.go index 152c535..7c70862 100644 --- a/machinery/src/models/config.go +++ b/machinery/src/models/config.go @@ -75,6 +75,14 @@ type Capture struct { Fragmented string `json:"fragmented,omitempty" bson:"fragmented,omitempty"` FragmentedDuration int64 `json:"fragmentedduration,omitempty" bson:"fragmentedduration,omitempty"` PixelChangeThreshold *int `json:"pixelChangeThreshold,omitempty"` + // ONVIFMotion routes the camera's ONVIF motion events into the + // agent's motion-triggered recording pipeline. When "true" the + // agent opens an event/stream against the configured ONVIF + // endpoint and forwards Motion+Active events to HandleMotion. + // Requires Capture.IPCamera.ONVIFXAddr / ONVIFUsername / + // ONVIFPassword to be set. Default empty (disabled) keeps the + // existing pixel-diff motion detection as the only source. + ONVIFMotion string `json:"onvif_motion,omitempty" bson:"onvif_motion,omitempty"` } // IPCamera configuration, such as the RTSP url of the IPCamera and the FPS. diff --git a/machinery/src/onvif/events.go b/machinery/src/onvif/events.go new file mode 100644 index 0000000..005694e --- /dev/null +++ b/machinery/src/onvif/events.go @@ -0,0 +1,225 @@ +package onvif + +import ( + "context" + "errors" + "strconv" + "strings" + "time" + + "github.com/kerberos-io/agent/machinery/src/log" + "github.com/kerberos-io/agent/machinery/src/models" + "github.com/kerberos-io/onvif/event/stream" +) + +// The library handles in-stream reconnect; these guards cover the +// initial-connect path the library cannot see. +const ( + initialBackoff = time.Second + maxBackoff = 5 * time.Minute +) + +// HandleONVIFEventStream opens an event/stream against the configured +// ONVIF camera and routes Motion events into communication.HandleMotion. +// +// Behind the Capture.ONVIFMotion flag; the goroutine returns +// immediately when not enabled. The flag is read once at start, so +// toggling at runtime requires an agent restart. On transient +// construction failure (camera not yet ready at boot, brief network +// blip, credential reload) the goroutine retries with exponential +// backoff. Exits when ctx is cancelled. +func HandleONVIFEventStream(ctx context.Context, configuration *models.Configuration, communication *models.Communication) { + log.Log.Debug("onvif.HandleONVIFEventStream(): started") + defer log.Log.Debug("onvif.HandleONVIFEventStream(): finished") + + if !isONVIFMotionEnabled(configuration.Config.Capture.ONVIFMotion) { + return + } + if configuration.Config.Capture.IPCamera.ONVIFXAddr == "" { + log.Log.Warning("onvif.HandleONVIFEventStream(): ONVIFMotion enabled but ONVIFXAddr is empty; nothing to do") + return + } + + backoff := initialBackoff + for { + if ctx.Err() != nil { + return + } + recoverable := runStreamOnce(ctx, configuration, communication) + if !recoverable { + return + } + if !sleepCtx(ctx, backoff) { + return + } + backoff *= 2 + if backoff > maxBackoff { + backoff = maxBackoff + } + } +} + +// runStreamOnce returns true when the caller should retry construction +// (transient failure), false on clean ctx-driven shutdown. +func runStreamOnce(ctx context.Context, configuration *models.Configuration, communication *models.Communication) (retry bool) { + camera := configuration.Config.Capture.IPCamera + + device, _, err := ConnectToOnvifDevice(&camera) + if err != nil { + log.Log.Error("onvif.HandleONVIFEventStream(): connect: " + err.Error()) + return true + } + + deviceID := resolveDeviceID(configuration.Name, camera.ONVIFXAddr) + s, err := stream.NewStream(ctx, device, stream.Options{DeviceID: deviceID}) + if err != nil { + log.Log.Error("onvif.HandleONVIFEventStream(): open stream: " + err.Error()) + return true + } + defer func() { + if err := s.Close(); err != nil { + log.Log.Debug("onvif.HandleONVIFEventStream(): close: " + err.Error()) + } + }() + + log.Log.Info("onvif.HandleONVIFEventStream(): consuming events for " + deviceID) + + // recovering = the first successful event after an error streak + // logs a recovery line so on-call operators see the clear-of- + // condition for the ERROR they were paged on. + var recovering bool + for { + select { + case <-ctx.Done(): + return false + case ev, ok := <-s.Events(): + if !ok { + return false + } + if recovering { + log.Log.Info("onvif.HandleONVIFEventStream(): event stream recovered for " + deviceID) + recovering = false + } + dispatchEvent(ctx, ev, configuration, communication) + case e, ok := <-s.Errors(): + if !ok { + return false + } + recovering = true + logStreamError(e) + } + } +} + +// dispatchEvent routes motion-active events to HandleMotion. +// +// The ctx pre-check + ctx-in-select guards a shutdown race: the agent +// closes HandleMotion shortly after cancelling ctx, and a stale event +// reaching the send would otherwise panic on a closed channel. +func dispatchEvent(ctx context.Context, ev stream.Event, configuration *models.Configuration, communication *models.Communication) { + topic := sanitiseTopic(ev.Topic) + if ev.Kind != stream.KindMotion { + log.Log.Debug("onvif.dispatchEvent(): non-motion event " + ev.Kind.String() + " topic=" + topic) + return + } + if ev.State != stream.StateActive { + return + } + if !isTransition(ev.Operation) { + log.Log.Debug("onvif.dispatchEvent(): " + ev.Operation.String() + " is not a transition, not a trigger: topic=" + topic) + return + } + if configuration.Config.Capture.Recording == "false" { + return + } + if ctx.Err() != nil { + return + } + + dataToPass := models.MotionDataPartial{ + Timestamp: time.Now().Unix(), + NumberOfChanges: 0, // ONVIF does not quantify motion area. + } + select { + case <-ctx.Done(): + case communication.HandleMotion <- dataToPass: + // Logged on the send, not before it: this line records that a + // recording started, so a dropped event must not leave one. + log.Log.Debug("onvif.dispatchEvent(): recording trigger " + ev.Kind.String() + " topic=" + topic) + default: + log.Log.Debug("onvif.dispatchEvent(): HandleMotion full, dropping ONVIF motion event") + } +} + +// isTransition reports whether an operation represents a state change. +// A camera replays every property's current state as Initialized on +// each new subscription and announces removals as Deleted; neither is +// motion starting. Absent (Unknown) counts — PropertyOperation is +// optional per WS-Notification and many non-property events omit it. +func isTransition(op stream.PropertyOperation) bool { + return op == stream.PropertyChanged || op == stream.PropertyUnknown +} + +// maxLoggedTopic bounds a topic in the log; the wire imposes no limit, +// and the reject path logs every event received. +const maxLoggedTopic = 256 + +// sanitiseTopic makes a camera-controlled topic safe to concatenate +// into a log line. logrus's coloured text formatter writes the message +// unquoted, so a raw newline would let a camera forge entries in the +// log being used to diagnose it. +func sanitiseTopic(topic string) string { + if len(topic) > maxLoggedTopic { + topic = topic[:maxLoggedTopic] + "…(truncated)" + } + quoted := strconv.Quote(topic) + return quoted[1 : len(quoted)-1] +} + +// logStreamError logs at a level matching severity: recreate is loud +// because it usually means the camera is offline; pull and renew are +// debug because the library recovers from them automatically. +func logStreamError(e error) { + var recreate stream.ErrRecreateFailed + var pull stream.ErrPullFailed + var renew stream.ErrRenewFailed + switch { + case errors.As(e, &recreate): + log.Log.Error("onvif.HandleONVIFEventStream(): subscription recreate failed (camera may be offline): " + recreate.Err.Error()) + case errors.As(e, &renew): + log.Log.Debug("onvif.HandleONVIFEventStream(): renew failed (will recover via pull/recreate): " + renew.Err.Error()) + case errors.As(e, &pull): + log.Log.Debug("onvif.HandleONVIFEventStream(): pull failed (will retry): " + pull.Err.Error()) + default: + log.Log.Info("onvif.HandleONVIFEventStream(): stream error: " + e.Error()) + } +} + +func isONVIFMotionEnabled(v string) bool { + return strings.EqualFold(strings.TrimSpace(v), "true") +} + +// resolveDeviceID falls back from operator-supplied name to ONVIF +// endpoint to a constant placeholder so log lines always have +// something to grep. +func resolveDeviceID(configName, xaddr string) string { + if n := strings.TrimSpace(configName); n != "" { + return n + } + if x := strings.TrimSpace(xaddr); x != "" { + return x + } + return "unknown" +} + +// sleepCtx returns false if ctx was cancelled, true if d elapsed. +func sleepCtx(ctx context.Context, d time.Duration) bool { + t := time.NewTimer(d) + defer t.Stop() + select { + case <-ctx.Done(): + return false + case <-t.C: + return true + } +} diff --git a/machinery/src/onvif/events_test.go b/machinery/src/onvif/events_test.go new file mode 100644 index 0000000..2dba9b6 --- /dev/null +++ b/machinery/src/onvif/events_test.go @@ -0,0 +1,375 @@ +package onvif + +import ( + "bytes" + "context" + "strings" + "testing" + "time" + + "github.com/kerberos-io/agent/machinery/src/models" + "github.com/kerberos-io/onvif/event/stream" + "github.com/sirupsen/logrus" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func makeConfig(recording, onvifMotion, name string) *models.Configuration { + return &models.Configuration{ + Name: name, + Config: models.Config{ + Capture: models.Capture{ + Recording: recording, + ONVIFMotion: onvifMotion, + }, + }, + } +} + +func makeCommunication(buffer int) *models.Communication { + return &models.Communication{ + HandleMotion: make(chan models.MotionDataPartial, buffer), + } +} + +// --- dispatchEvent --------------------------------------------------- + +func TestDispatchEvent_MotionActive_SendsToHandleMotion(t *testing.T) { + cfg := makeConfig("true", "true", "cam-1") + comm := makeCommunication(1) + ev := stream.Event{Kind: stream.KindMotion, State: stream.StateActive} + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + dispatchEvent(ctx, ev, cfg, comm) + + select { + case m := <-comm.HandleMotion: + assert.NotZero(t, m.Timestamp) + case <-time.After(time.Second): + t.Fatal("expected motion data on HandleMotion") + } +} + +func TestDispatchEvent_MotionInactive_DoesNotSend(t *testing.T) { + cfg := makeConfig("true", "true", "cam-1") + comm := makeCommunication(1) + ev := stream.Event{Kind: stream.KindMotion, State: stream.StateInactive} + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + dispatchEvent(ctx, ev, cfg, comm) + + select { + case <-comm.HandleMotion: + t.Fatal("inactive motion must not reach HandleMotion (motion-stop is a follow-up)") + case <-time.After(100 * time.Millisecond): + } +} + +func TestDispatchEvent_NonMotionKindIgnored(t *testing.T) { + cfg := makeConfig("true", "true", "cam-1") + comm := makeCommunication(1) + ev := stream.Event{Kind: stream.KindDigitalInput, State: stream.StateActive} + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + dispatchEvent(ctx, ev, cfg, comm) + + select { + case <-comm.HandleMotion: + t.Fatal("non-motion kinds must not reach HandleMotion") + case <-time.After(100 * time.Millisecond): + } +} + +// captureDebugLog redirects logrus to a buffer at debug level for the +// duration of a test and returns what was written. It mutates package +// globals, so callers must not run in parallel. +func captureDebugLog(t *testing.T) *bytes.Buffer { + t.Helper() + var buf bytes.Buffer + prevOut, prevLevel := logrus.StandardLogger().Out, logrus.GetLevel() + logrus.SetOutput(&buf) + logrus.SetLevel(logrus.DebugLevel) + t.Cleanup(func() { + logrus.SetOutput(prevOut) + logrus.SetLevel(prevLevel) + }) + return &buf +} + +// TestDispatchEvent_LogsTheTriggeringTopic — a dispatched event is what +// actually starts a recording, so its topic is the one an operator needs +// when a camera records for the wrong reason (or the right reason and +// nobody can prove which). Rejected events were already logged; without +// this the triggering topic is only knowable by elimination. +func TestDispatchEvent_LogsTheTriggeringTopic(t *testing.T) { + buf := captureDebugLog(t) + + cfg := makeConfig("true", "true", "cam-1") + comm := makeCommunication(1) + ev := stream.Event{ + Kind: stream.KindMotion, + State: stream.StateActive, + Topic: "tns1:RuleEngine/tnsaxis:VMD3/vmd3_video_1", + } + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + dispatchEvent(ctx, ev, cfg, comm) + + assert.Contains(t, buf.String(), "tns1:RuleEngine/tnsaxis:VMD3/vmd3_video_1", + "the dispatched event's topic must appear in the log") + assert.Contains(t, buf.String(), "Motion", + "the dispatched event's Kind must appear in the log") +} + +// TestDispatchEvent_PropertyOperation — a camera replays the current +// state of every property topic as Initialized whenever a pull-point +// subscription is created. If that counts as a trigger, every +// reconnect restarts a recording for any motion property that happens +// to be active, and a flapping subscription manufactures motion out of +// nothing. Only reject Initialized specifically: PropertyOperation is +// optional per WS-Notification and absent on many non-property events, +// which decode reports as PropertyUnknown. +func TestDispatchEvent_PropertyOperation(t *testing.T) { + tests := []struct { + name string + op stream.PropertyOperation + wantSend bool + }{ + {"changed is a real transition", stream.PropertyChanged, true}, + {"absent attribute still counts", stream.PropertyUnknown, true}, + {"initialized is a subscription state replay", stream.PropertyInitialized, false}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + cfg := makeConfig("true", "true", "cam-1") + comm := makeCommunication(1) + ev := stream.Event{ + Kind: stream.KindMotion, + State: stream.StateActive, + Operation: tt.op, + Topic: "tns1:RuleEngine/tnsaxis:VMD3/vmd3_video_1", + } + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + dispatchEvent(ctx, ev, cfg, comm) + + select { + case <-comm.HandleMotion: + if !tt.wantSend { + t.Fatalf("%v must not trigger a recording", tt.op) + } + case <-time.After(100 * time.Millisecond): + if tt.wantSend { + t.Fatalf("%v must trigger a recording", tt.op) + } + } + }) + } +} + +func TestDispatchEvent_RecordingDisabled_DoesNotSend(t *testing.T) { + cfg := makeConfig("false", "true", "cam-1") + comm := makeCommunication(1) + ev := stream.Event{Kind: stream.KindMotion, State: stream.StateActive} + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + dispatchEvent(ctx, ev, cfg, comm) + + select { + case <-comm.HandleMotion: + t.Fatal("Recording=false must gate the send (matches computervision behaviour)") + case <-time.After(100 * time.Millisecond): + } +} + +func TestDispatchEvent_HandleMotionFull_DropsRatherThanBlocks(t *testing.T) { + cfg := makeConfig("true", "true", "cam-1") + // Pre-fill the buffer so the next send would block. + comm := &models.Communication{HandleMotion: make(chan models.MotionDataPartial, 1)} + comm.HandleMotion <- models.MotionDataPartial{} + ev := stream.Event{Kind: stream.KindMotion, State: stream.StateActive} + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + done := make(chan struct{}) + go func() { + dispatchEvent(ctx, ev, cfg, comm) + close(done) + }() + + select { + case <-done: + case <-time.After(time.Second): + t.Fatal("dispatchEvent must drop when HandleMotion is full, not block") + } +} + +func TestDispatchEvent_CtxCancelledAndHandleMotionClosed_DoesNotPanic(t *testing.T) { + // Regression for the shutdown race: between cancel() and + // close(HandleMotion) the agent leaves a 3s window. If dispatchEvent + // runs in that window AFTER the channel is closed, a non-protected + // send would panic. The ctx pre-check must short-circuit before the + // send is attempted. + cfg := makeConfig("true", "true", "cam-1") + comm := &models.Communication{HandleMotion: make(chan models.MotionDataPartial, 1)} + close(comm.HandleMotion) + ev := stream.Event{Kind: stream.KindMotion, State: stream.StateActive} + + ctx, cancel := context.WithCancel(context.Background()) + cancel() // already cancelled, matching the shutdown sequence + + assert.NotPanics(t, func() { + dispatchEvent(ctx, ev, cfg, comm) + }) +} + +// --- isONVIFMotionEnabled -------------------------------------------- + +func TestIsONVIFMotionEnabled_CaseAndWhitespace(t *testing.T) { + tests := []struct { + in string + want bool + }{ + {"true", true}, + {"True", true}, + {"TRUE", true}, + {" true", true}, + {"true ", true}, + {" true ", true}, + {"false", false}, + {"False", false}, + {"", false}, + {"yes", false}, + {"1", false}, + } + for _, tc := range tests { + t.Run(tc.in, func(t *testing.T) { + assert.Equal(t, tc.want, isONVIFMotionEnabled(tc.in)) + }) + } +} + +// --- resolveDeviceID ------------------------------------------------- + +func TestResolveDeviceID_FallbackChain(t *testing.T) { + tests := []struct { + name string + cfgName string + xaddr string + want string + }{ + {"name_set", "front-door", "192.168.1.10", "front-door"}, + {"name_empty_xaddr_set", "", "192.168.1.10", "192.168.1.10"}, + {"name_whitespace_only_xaddr_set", " ", "192.168.1.10", "192.168.1.10"}, + {"both_empty", "", "", "unknown"}, + {"name_with_trailing_whitespace", "cam-2 ", "192.168.1.10", "cam-2"}, + } + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + assert.Equal(t, tc.want, resolveDeviceID(tc.cfgName, tc.xaddr)) + }) + } +} + +// TestDispatchEvent_OnlyRealTransitionsTrigger — a camera replays every +// property topic's state on each new subscription (Initialized) and +// announces removals (Deleted). Neither is a motion transition, and a +// flapping pull-point would otherwise manufacture recordings out of +// replayed state. PropertyOperation is optional per WS-Notification, so +// absent (Unknown) still counts — many non-property events omit it. +func TestDispatchEvent_OnlyRealTransitionsTrigger(t *testing.T) { + tests := []struct { + op stream.PropertyOperation + wantSend bool + }{ + {stream.PropertyChanged, true}, + {stream.PropertyUnknown, true}, + {stream.PropertyInitialized, false}, + {stream.PropertyDeleted, false}, + } + + for _, tt := range tests { + t.Run(tt.op.String(), func(t *testing.T) { + cfg := makeConfig("true", "true", "cam-1") + comm := makeCommunication(1) + ev := stream.Event{Kind: stream.KindMotion, State: stream.StateActive, Operation: tt.op} + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + dispatchEvent(ctx, ev, cfg, comm) + + if tt.wantSend { + require.Len(t, comm.HandleMotion, 1, "%v must trigger a recording", tt.op) + return + } + require.Empty(t, comm.HandleMotion, "%v must not trigger a recording", tt.op) + }) + } +} + +// TestSanitiseTopic — ev.Topic is camera-controlled and reaches the log +// unmodified. logrus's coloured text formatter (the default) writes the +// message without quoting, so an embedded newline forges whole log +// lines: a compromised camera can fabricate ERROR entries or spoof +// another device's id, in the logs an operator is reading to diagnose +// that very camera. Length is also unbounded on the wire, and the +// reject path logs every event, so an oversized topic is a cheap way to +// evict a container's whole retained history. +func TestSanitiseTopic(t *testing.T) { + tests := []struct { + name string + in string + want string + }{ + {"ordinary topic passes through", "tns1:RuleEngine/tnsaxis:VMD3/vmd3_video_1", "tns1:RuleEngine/tnsaxis:VMD3/vmd3_video_1"}, + {"newline cannot forge a line", "a\nERRO[fake] boom", `a\nERRO[fake] boom`}, + {"carriage return", "a\rb", `a\rb`}, + {"tab", "a\tb", `a\tb`}, + {"NUL", "a\x00b", `a\x00b`}, + {"empty", "", ""}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + got := sanitiseTopic(tt.in) + assert.Equal(t, tt.want, got) + assert.NotContains(t, got, "\n", "no raw newline may survive") + assert.NotContains(t, got, "\r", "no raw carriage return may survive") + }) + } +} + +func TestSanitiseTopic_Truncates(t *testing.T) { + got := sanitiseTopic(strings.Repeat("x", maxLoggedTopic*2)) + assert.LessOrEqual(t, len(got), maxLoggedTopic+len("…(truncated)")) + assert.Contains(t, got, "truncated") +} + +// TestDispatchEvent_LogsTriggerOnlyWhenSent — the trigger line is the +// record that a recording started. Logging it before the send means a +// dropped event (full channel, or shutdown) leaves a line claiming a +// recording that never began. +func TestDispatchEvent_LogsTriggerOnlyWhenSent(t *testing.T) { + buf := captureDebugLog(t) + + cfg := makeConfig("true", "true", "cam-1") + comm := &models.Communication{HandleMotion: make(chan models.MotionDataPartial, 1)} + comm.HandleMotion <- models.MotionDataPartial{} // full + ev := stream.Event{Kind: stream.KindMotion, State: stream.StateActive, Topic: "tns1:VideoSource/MotionAlarm"} + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + dispatchEvent(ctx, ev, cfg, comm) + + assert.NotContains(t, buf.String(), "recording trigger", + "a dropped event must not be logged as a trigger") + assert.Contains(t, buf.String(), "dropping", "the drop itself must still be logged") +}