### Question:
EventProcessor { public Event process(String stream, Event event) { LOG.debug("Entering EventProcessor#onNext: {}", event); boolean isEnabled = config.isMREClientEnabled(); if (!isEnabled) { LOG.debug("Mantis Realtime Events Publisher is disabled." + "Set the property defined in your MrePublishConfiguration object to true to enable."); return null; } if (config.isDeepCopyEventMapEnabled()) { event = new Event(event.getMap(), true); } maskSensitiveFields(event); if (config.isTeeEnabled()) { tee.tee(config.teeStreamName(), event); } List<Subscription> matchingSubscriptions = new ArrayList<>(); if (streamManager.hasSubscriptions(stream)) { final Set<Subscription> streamSubscriptions = streamManager.getStreamSubscriptions(stream); for (Subscription s : streamSubscriptions) { try { if (s.matches(event)) { matchingSubscriptions.add(s); } } catch (Exception e) { streamManager.getStreamMetrics(stream) .ifPresent(m -> m.getMantisQueryFailedCounter().increment()); int rndNo = randomGenerator.nextInt(1_000_000); if (rndNo < 10) { sendError(s, e.getMessage()); } } } } Event projectedEvent = null; if (!matchingSubscriptions.isEmpty()) { projectedEvent = projectSupersetEvent(stream, matchingSubscriptions, event); } else { if (LOG.isTraceEnabled()) { LOG.trace("no matching subscriptions"); } } if (LOG.isDebugEnabled()) { LOG.debug("Exit EventProcessor#onNext: {}", event); } return projectedEvent; } EventProcessor(MrePublishConfiguration config, StreamManager streamManager, Tee tee); Event process(String stream, Event event);  }### Answer:
@Test void shouldReturnEnrichedEventForStream() throws Exception { when(streamManager.hasSubscriptions(anyString())).thenReturn(true); SettableConfig config = new DefaultSettableConfig(); PropertyRepository repository = DefaultPropertyFactory.from(config); MrePublishConfiguration mrePublishConfiguration = new SampleArchaiusMrePublishConfiguration(repository); Subscription subscription = new MQLSubscription("id", "select * where true"); Set<Subscription> subscriptions = new ConcurrentSkipListSet<>(); subscriptions.add(subscription); when(streamManager.getStreamSubscriptions(anyString())).thenReturn(subscriptions); Event event = new Event(); event.set("k1", "v1"); Event actual = eventProcessor.process(StreamType.DEFAULT_EVENT_STREAM, event); assertEquals(actual.get("mantisStream"), StreamType.DEFAULT_EVENT_STREAM); assertEquals(actual.get("type"), "EVENT"); assertEquals(actual.get("k1"), "v1"); assertEquals(((ArrayList)actual.get("matched-clients")).size(), 1); assertEquals(((ArrayList)actual.get("matched-clients")).get(0), "id"); }
@Test void shouldReturnEmptyEventForStream() throws Exception { when(streamManager.hasSubscriptions(anyString())).thenReturn(false); Event event = new Event(); event.set("k1", "v1"); Event actual = eventProcessor.process(StreamType.DEFAULT_EVENT_STREAM, event); assertNull(actual); Subscription subscription = mock(MQLSubscription.class); when(subscription.matches(any(Event.class))).thenReturn(false); Set<Subscription> subscriptions = new ConcurrentSkipListSet<>(); subscriptions.add(subscription); when(streamManager.getStreamSubscriptions(anyString())).thenReturn(subscriptions); actual = eventProcessor.process(StreamType.DEFAULT_EVENT_STREAM, event); assertNull(actual); }