src/test/java/org/onap/dcae/common/publishing/DMaaPEventPublisherTest.java

   1 /*-
   2  * ============LICENSE_START=======================================================
   3  * org.onap.dcaegen2.collectors.ves
   4  * ================================================================================
   5  * Copyright (C) 2018,2020 Nokia. All rights reserved.
   6  * Copyright (C) 2020 AT&T. All rights reserved.
   7  * ================================================================================
   8  * Licensed under the Apache License, Version 2.0 (the "License");
   9  * you may not use this file except in compliance with the License.
  10  * You may obtain a copy of the License at
  11  *
  12  *      http://www.apache.org/licenses/LICENSE-2.0
  13  *
  14  * Unless required by applicable law or agreed to in writing, software
  15  * distributed under the License is distributed on an "AS IS" BASIS,
  16  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  17  * See the License for the specific language governing permissions and
  18  * limitations under the License.
  19  * ============LICENSE_END=========================================================
  20  */
  21 package org.onap.dcae.common.publishing;
  22
  23 import static io.vavr.API.Option;
  24 import static org.mockito.ArgumentMatchers.anyString;
  25 import static org.mockito.BDDMockito.given;
  26 import static org.mockito.Mockito.mock;
  27 import static org.mockito.Mockito.verify;
  28 import static org.mockito.Mockito.when;
  29
  30 import com.att.nsa.cambria.client.CambriaBatchingPublisher;
  31 import java.io.IOException;
  32 import org.json.JSONObject;
  33 import org.junit.Before;
  34 import org.junit.Test;
  35 import org.slf4j.Logger;
  36
  37 public class DMaaPEventPublisherTest {
  38
  39     private static final String STREAM_ID = "sampleStreamId";
  40
  41     private DMaaPEventPublisher eventPublisher;
  42     private CambriaBatchingPublisher cambriaPublisher;
  43     private DMaaPPublishersCache DMaaPPublishersCache;
  44
  45     @Before
  46     public void setUp() {
  47         cambriaPublisher = mock(CambriaBatchingPublisher.class);
  48         DMaaPPublishersCache = mock(DMaaPPublishersCache.class);
  49         when(DMaaPPublishersCache.getPublisher(anyString())).thenReturn(Option(cambriaPublisher));
  50         eventPublisher = new DMaaPEventPublisher(DMaaPPublishersCache);
  51     }
  52
  53     @Test
  54     public void shouldSendEventToTopic() throws Exception {
  55         // given
  56         JSONObject event = new JSONObject("{\"event\":{\"commonEventHeader\":{\"startEpochMicrosec\":1537562659253019,\"sourceId\":\"79e90d76-513a-4f79-886d-470a0037c5cf\",\"eventId\":\"Heartbeat_vDNS_100.100.10.10\",\"nfcNamingCode\":\"DNS\",\"reportingEntityId\":\"79e90d76-513a-4f79-886d-470a0037c5cf\",\"eventType\":\"applicationVnf\",\"priority\":\"Normal\",\"version\":3,\"reportingEntityName\":\"dns01cmd004\",\"sequence\":36312,\"domain\":\"heartbeat\",\"lastEpochMicrosec\":1537562659253019,\"eventName\":\"Heartbeat_vDNS\",\"sourceName\":\"dns01cmd004\",\"nfNamingCode\":\"MDNS\"}}}");
  57
  58
  59         // when
  60         eventPublisher.sendEvent(event, STREAM_ID);
  61
  62         // then
  63         verify(cambriaPublisher).send("dns01cmd004", event.toString());
  64     }
  65
  66
  67     @Test
  68     public void shouldRemoveInternalVESUIDBeforeSending() throws Exception {
  69         // given
  70         JSONObject event = new JSONObject(
  71             "{\"VESuniqueId\": \"362e0146-ec5f-45f3-8d8f-bfe877c3f58e\",\"event\":{\"commonEventHeader\":{\"startEpochMicrosec\":1537562659253019,\"sourceId\":\"79e90d76-513a-4f79-886d-470a0037c5cf\",\"eventId\":\"Heartbeat_vDNS_100.100.10.10\",\"nfcNamingCode\":\"DNS\",\"reportingEntityId\":\"79e90d76-513a-4f79-886d-470a0037c5cf\",\"eventType\":\"applicationVnf\",\"priority\":\"Normal\",\"version\":3,\"reportingEntityName\":\"dns01cmd004\",\"sequence\":36312,\"domain\":\"heartbeat\",\"lastEpochMicrosec\":1537562659253019,\"eventName\":\"Heartbeat_vDNS\",\"sourceName\":\"dns01cmd004\",\"nfNamingCode\":\"MDNS\"}}}");
  72
  73         // when
  74         eventPublisher.sendEvent(event, STREAM_ID);
  75
  76         // then
  77         verify(cambriaPublisher).send("dns01cmd004", new JSONObject("{\"event\":{\"commonEventHeader\":{\"startEpochMicrosec\":1537562659253019,\"sourceId\":\"79e90d76-513a-4f79-886d-470a0037c5cf\",\"eventId\":\"Heartbeat_vDNS_100.100.10.10\",\"nfcNamingCode\":\"DNS\",\"reportingEntityId\":\"79e90d76-513a-4f79-886d-470a0037c5cf\",\"eventType\":\"applicationVnf\",\"priority\":\"Normal\",\"version\":3,\"reportingEntityName\":\"dns01cmd004\",\"sequence\":36312,\"domain\":\"heartbeat\",\"lastEpochMicrosec\":1537562659253019,\"eventName\":\"Heartbeat_vDNS\",\"sourceName\":\"dns01cmd004\",\"nfNamingCode\":\"MDNS\"}}}").toString());
  78     }
  79
  80     @Test
  81     public void shouldCloseConnectionWhenExceptionOccurred() throws Exception {
  82         // given
  83         JSONObject event = new JSONObject("{}");
  84         given(cambriaPublisher.send(anyString(), anyString())).willThrow(new IOException("epic fail"));
  85
  86         // when
  87         eventPublisher.sendEvent(event, STREAM_ID);
  88
  89         // then
  90         verify(DMaaPPublishersCache).closePublisherFor(STREAM_ID);
  91     }
  92 }