diff options
Diffstat (limited to 'dcae-analytics-cdap-plugins/src/test/java/org/openecomp/dcae/apod/analytics/cdap/plugins/streaming')
5 files changed, 379 insertions, 377 deletions
diff --git a/dcae-analytics-cdap-plugins/src/test/java/org/openecomp/dcae/apod/analytics/cdap/plugins/streaming/dmaap/DMaaPMRReceiverTest.java b/dcae-analytics-cdap-plugins/src/test/java/org/openecomp/dcae/apod/analytics/cdap/plugins/streaming/dmaap/DMaaPMRReceiverTest.java index 036244d..40fadcc 100644 --- a/dcae-analytics-cdap-plugins/src/test/java/org/openecomp/dcae/apod/analytics/cdap/plugins/streaming/dmaap/DMaaPMRReceiverTest.java +++ b/dcae-analytics-cdap-plugins/src/test/java/org/openecomp/dcae/apod/analytics/cdap/plugins/streaming/dmaap/DMaaPMRReceiverTest.java @@ -1,75 +1,75 @@ -/* - * ===============================LICENSE_START====================================== - * dcae-analytics - * ================================================================================ - * Copyright © 2017 AT&T Intellectual Property. All rights reserved. - * ================================================================================ - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - * ============================LICENSE_END=========================================== - */ - -package org.openecomp.dcae.apod.analytics.cdap.plugins.streaming.dmaap; - -import com.google.common.collect.ImmutableList; -import org.apache.spark.storage.StorageLevel; -import org.junit.Test; -import org.mockito.Mockito; -import org.openecomp.dcae.apod.analytics.cdap.plugins.BaseAnalyticsCDAPPluginsUnitTest; -import org.openecomp.dcae.apod.analytics.cdap.plugins.domain.config.dmaap.TestDMaaPMRSourcePluginConfig; -import org.openecomp.dcae.apod.analytics.common.exception.DCAEAnalyticsRuntimeException; -import org.openecomp.dcae.apod.analytics.dmaap.domain.response.DMaaPMRSubscriberResponse; -import org.openecomp.dcae.apod.analytics.dmaap.service.subscriber.DMaaPMRSubscriber; - -import static org.mockito.Mockito.times; -import static org.mockito.Mockito.verify; -import static org.mockito.Mockito.when; - -/** - * @author Rajiv Singla . Creation Date: 1/24/2017. - */ -public class DMaaPMRReceiverTest extends BaseAnalyticsCDAPPluginsUnitTest { - - - @Test - public void testStoreStructuredRecords() throws Exception { - - final TestDMaaPMRSourcePluginConfig testDMaaPMRSourcePluginConfig = getTestDMaaPMRSourcePluginConfig(); - final TestDMaaPMRReceiver dMaaPMRReceiver = - new TestDMaaPMRReceiver(StorageLevel.MEMORY_ONLY(), testDMaaPMRSourcePluginConfig); - - final DMaaPMRSubscriber dMaaPMRSubscriber = Mockito.mock(DMaaPMRSubscriber.class); - final DMaaPMRSubscriberResponse subscriberResponse = Mockito.mock(DMaaPMRSubscriberResponse.class); - when(dMaaPMRSubscriber.fetchMessages()).thenReturn(subscriberResponse); - when(subscriberResponse.getFetchedMessages()).thenReturn(ImmutableList.of("Test Message")); - when(subscriberResponse.getResponseCode()).thenReturn(200); - when(subscriberResponse.getResponseMessage()).thenReturn("OK"); - dMaaPMRReceiver.storeStructuredRecords(dMaaPMRSubscriber); - verify(dMaaPMRSubscriber, times(1)).fetchMessages(); - verify(subscriberResponse, times(1)).getFetchedMessages(); - } - - @Test - public void testStoreStructuredRecordsWhenSubscriberThrowsException() throws Exception { - - final TestDMaaPMRSourcePluginConfig testDMaaPMRSourcePluginConfig = getTestDMaaPMRSourcePluginConfig(); - final TestDMaaPMRReceiver dMaaPMRReceiver = - new TestDMaaPMRReceiver(StorageLevel.MEMORY_ONLY(), testDMaaPMRSourcePluginConfig); - - final DMaaPMRSubscriber dMaaPMRSubscriber = Mockito.mock(DMaaPMRSubscriber.class); - final DMaaPMRSubscriberResponse subscriberResponse = Mockito.mock(DMaaPMRSubscriberResponse.class); - when(dMaaPMRSubscriber.fetchMessages()).thenThrow(DCAEAnalyticsRuntimeException.class); - dMaaPMRReceiver.storeStructuredRecords(dMaaPMRSubscriber); - verify(dMaaPMRSubscriber, times(1)).fetchMessages(); - verify(subscriberResponse, times(0)).getFetchedMessages(); - } -} +/*
+ * ===============================LICENSE_START======================================
+ * dcae-analytics
+ * ================================================================================
+ * Copyright © 2017 AT&T Intellectual Property. All rights reserved.
+ * ================================================================================
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ * ============================LICENSE_END===========================================
+ */
+
+package org.openecomp.dcae.apod.analytics.cdap.plugins.streaming.dmaap;
+
+import com.google.common.collect.ImmutableList;
+import org.apache.spark.storage.StorageLevel;
+import org.junit.Test;
+import org.mockito.Mockito;
+import org.openecomp.dcae.apod.analytics.cdap.plugins.BaseAnalyticsCDAPPluginsUnitTest;
+import org.openecomp.dcae.apod.analytics.cdap.plugins.domain.config.dmaap.TestDMaaPMRSourcePluginConfig;
+import org.openecomp.dcae.apod.analytics.common.exception.DCAEAnalyticsRuntimeException;
+import org.openecomp.dcae.apod.analytics.dmaap.domain.response.DMaaPMRSubscriberResponse;
+import org.openecomp.dcae.apod.analytics.dmaap.service.subscriber.DMaaPMRSubscriber;
+
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+/**
+ * @author Rajiv Singla . Creation Date: 1/24/2017.
+ */
+public class DMaaPMRReceiverTest extends BaseAnalyticsCDAPPluginsUnitTest {
+
+
+ @Test
+ public void testStoreStructuredRecords() throws Exception {
+
+ final TestDMaaPMRSourcePluginConfig testDMaaPMRSourcePluginConfig = getTestDMaaPMRSourcePluginConfig();
+ final TestDMaaPMRReceiver dMaaPMRReceiver =
+ new TestDMaaPMRReceiver(StorageLevel.MEMORY_ONLY(), testDMaaPMRSourcePluginConfig);
+
+ final DMaaPMRSubscriber dMaaPMRSubscriber = Mockito.mock(DMaaPMRSubscriber.class);
+ final DMaaPMRSubscriberResponse subscriberResponse = Mockito.mock(DMaaPMRSubscriberResponse.class);
+ when(dMaaPMRSubscriber.fetchMessages()).thenReturn(subscriberResponse);
+ when(subscriberResponse.getFetchedMessages()).thenReturn(ImmutableList.of("Test Message"));
+ when(subscriberResponse.getResponseCode()).thenReturn(200);
+ when(subscriberResponse.getResponseMessage()).thenReturn("OK");
+ dMaaPMRReceiver.storeStructuredRecords(dMaaPMRSubscriber);
+ verify(dMaaPMRSubscriber, times(1)).fetchMessages();
+ verify(subscriberResponse, times(1)).getFetchedMessages();
+ }
+
+ @Test
+ public void testStoreStructuredRecordsWhenSubscriberThrowsException() throws Exception {
+
+ final TestDMaaPMRSourcePluginConfig testDMaaPMRSourcePluginConfig = getTestDMaaPMRSourcePluginConfig();
+ final TestDMaaPMRReceiver dMaaPMRReceiver =
+ new TestDMaaPMRReceiver(StorageLevel.MEMORY_ONLY(), testDMaaPMRSourcePluginConfig);
+
+ final DMaaPMRSubscriber dMaaPMRSubscriber = Mockito.mock(DMaaPMRSubscriber.class);
+ final DMaaPMRSubscriberResponse subscriberResponse = Mockito.mock(DMaaPMRSubscriberResponse.class);
+ when(dMaaPMRSubscriber.fetchMessages()).thenThrow(DCAEAnalyticsRuntimeException.class);
+ dMaaPMRReceiver.storeStructuredRecords(dMaaPMRSubscriber);
+ verify(dMaaPMRSubscriber, times(1)).fetchMessages();
+ verify(subscriberResponse, times(0)).getFetchedMessages();
+ }
+}
diff --git a/dcae-analytics-cdap-plugins/src/test/java/org/openecomp/dcae/apod/analytics/cdap/plugins/streaming/dmaap/DMaaPMRSourceTest.java b/dcae-analytics-cdap-plugins/src/test/java/org/openecomp/dcae/apod/analytics/cdap/plugins/streaming/dmaap/DMaaPMRSourceTest.java index a86206a..0b5ac87 100644 --- a/dcae-analytics-cdap-plugins/src/test/java/org/openecomp/dcae/apod/analytics/cdap/plugins/streaming/dmaap/DMaaPMRSourceTest.java +++ b/dcae-analytics-cdap-plugins/src/test/java/org/openecomp/dcae/apod/analytics/cdap/plugins/streaming/dmaap/DMaaPMRSourceTest.java @@ -1,90 +1,91 @@ -/* - * ===============================LICENSE_START====================================== - * dcae-analytics - * ================================================================================ - * Copyright © 2017 AT&T Intellectual Property. All rights reserved. - * ================================================================================ - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - * ============================LICENSE_END=========================================== - */ - -package org.openecomp.dcae.apod.analytics.cdap.plugins.streaming.dmaap; - -import co.cask.cdap.api.data.format.StructuredRecord; -import co.cask.cdap.api.data.schema.Schema; -import co.cask.cdap.etl.api.PipelineConfigurer; -import co.cask.cdap.etl.api.StageConfigurer; -import co.cask.cdap.etl.api.streaming.StreamingContext; -import org.apache.spark.streaming.api.java.JavaDStream; -import org.apache.spark.streaming.api.java.JavaReceiverInputDStream; -import org.apache.spark.streaming.api.java.JavaStreamingContext; -import org.apache.spark.streaming.receiver.Receiver; -import org.junit.Before; -import org.junit.Test; -import org.mockito.Mockito; -import org.openecomp.dcae.apod.analytics.cdap.common.exception.CDAPSettingsException; -import org.openecomp.dcae.apod.analytics.cdap.plugins.BaseAnalyticsCDAPPluginsUnitTest; -import org.openecomp.dcae.apod.analytics.cdap.plugins.domain.config.dmaap.TestDMaaPMRSourcePluginConfig; - -import static org.junit.Assert.assertNotNull; -import static org.mockito.ArgumentMatchers.any; -import static org.mockito.Mockito.doNothing; -import static org.mockito.Mockito.when; - -/** - * @author Rajiv Singla . Creation Date: 1/24/2017. - */ -public class DMaaPMRSourceTest extends BaseAnalyticsCDAPPluginsUnitTest { - - private PipelineConfigurer pipelineConfigurer; - - @Before - public void before() { - pipelineConfigurer = Mockito.mock(PipelineConfigurer.class); - final StageConfigurer stageConfigurer = Mockito.mock(StageConfigurer.class); - when(pipelineConfigurer.getStageConfigurer()).thenReturn(stageConfigurer); - doNothing().when(stageConfigurer).setOutputSchema(any(Schema.class)); - } - - @Test - public void testDMaaPMRSourceConfigurePipelineWithValidPluginSettings() throws Exception { - final TestDMaaPMRSourcePluginConfig testDMaaPMRSourcePluginConfig = getTestDMaaPMRSourcePluginConfig(); - final DMaaPMRSource dMaaPMRSource = new DMaaPMRSource(testDMaaPMRSourcePluginConfig); - dMaaPMRSource.configurePipeline(pipelineConfigurer); - assertNotNull(dMaaPMRSource); - } - - @Test(expected = CDAPSettingsException.class) - public void testDMaaPMRSourceConfigurePipelineWithInvalidPluginSettings() throws Exception { - final TestDMaaPMRSourcePluginConfig testDMaaPMRSourcePluginConfig = getTestDMaaPMRSourcePluginConfig(); - // blank out DMaaP MR Source Host - testDMaaPMRSourcePluginConfig.setHostName(null); - final DMaaPMRSource dMaaPMRSource = new DMaaPMRSource(testDMaaPMRSourcePluginConfig); - dMaaPMRSource.configurePipeline(pipelineConfigurer); - } - - - @Test - public void testGetStream() throws Exception { - final StreamingContext streamingContext = Mockito.mock(StreamingContext.class); - final JavaStreamingContext javaStreamingContext = Mockito.mock(JavaStreamingContext.class); - final JavaReceiverInputDStream dMaaPMRReceiver = Mockito.mock(JavaReceiverInputDStream.class); - when(streamingContext.getSparkStreamingContext()).thenReturn(javaStreamingContext); - when(javaStreamingContext.receiverStream(any(Receiver.class))).thenReturn(dMaaPMRReceiver); - - final TestDMaaPMRSourcePluginConfig testDMaaPMRSourcePluginConfig = getTestDMaaPMRSourcePluginConfig(); - final DMaaPMRSource dMaaPMRSource = new DMaaPMRSource(testDMaaPMRSourcePluginConfig); - final JavaDStream<StructuredRecord> stream = dMaaPMRSource.getStream(streamingContext); - assertNotNull(stream); - } -} +/*
+ * ===============================LICENSE_START======================================
+ * dcae-analytics
+ * ================================================================================
+ * Copyright © 2017 AT&T Intellectual Property. All rights reserved.
+ * ================================================================================
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ * ============================LICENSE_END===========================================
+ */
+
+package org.openecomp.dcae.apod.analytics.cdap.plugins.streaming.dmaap;
+
+import co.cask.cdap.api.data.format.StructuredRecord;
+import co.cask.cdap.api.data.schema.Schema;
+import co.cask.cdap.etl.api.PipelineConfigurer;
+import co.cask.cdap.etl.api.StageConfigurer;
+import co.cask.cdap.etl.api.streaming.StreamingContext;
+import org.apache.spark.streaming.api.java.JavaDStream;
+import org.apache.spark.streaming.api.java.JavaReceiverInputDStream;
+import org.apache.spark.streaming.api.java.JavaStreamingContext;
+import org.apache.spark.streaming.receiver.Receiver;
+import org.junit.Before;
+import org.junit.Test;
+import org.mockito.Mockito;
+import org.openecomp.dcae.apod.analytics.cdap.common.exception.CDAPSettingsException;
+import org.openecomp.dcae.apod.analytics.cdap.plugins.BaseAnalyticsCDAPPluginsUnitTest;
+import org.openecomp.dcae.apod.analytics.cdap.plugins.domain.config.dmaap.TestDMaaPMRSourcePluginConfig;
+
+import static org.junit.Assert.assertNotNull;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.doNothing;
+import static org.mockito.Mockito.when;
+
+/**
+ * @author Rajiv Singla . Creation Date: 1/24/2017.
+ */
+public class DMaaPMRSourceTest extends BaseAnalyticsCDAPPluginsUnitTest {
+
+ private PipelineConfigurer pipelineConfigurer;
+
+ @Before
+ public void before() {
+ pipelineConfigurer = Mockito.mock(PipelineConfigurer.class);
+ final StageConfigurer stageConfigurer = Mockito.mock(StageConfigurer.class);
+ when(pipelineConfigurer.getStageConfigurer()).thenReturn(stageConfigurer);
+ doNothing().when(stageConfigurer).setOutputSchema(any(Schema.class));
+ }
+
+ @Test
+ public void testDMaaPMRSourceConfigurePipelineWithValidPluginSettings() throws Exception {
+ final TestDMaaPMRSourcePluginConfig testDMaaPMRSourcePluginConfig = getTestDMaaPMRSourcePluginConfig();
+ final DMaaPMRSource dMaaPMRSource = new DMaaPMRSource(testDMaaPMRSourcePluginConfig);
+ dMaaPMRSource.configurePipeline(pipelineConfigurer);
+ assertNotNull(dMaaPMRSource);
+ }
+
+ @Test(expected = CDAPSettingsException.class)
+ public void testDMaaPMRSourceConfigurePipelineWithInvalidPluginSettings() throws Exception {
+ final TestDMaaPMRSourcePluginConfig testDMaaPMRSourcePluginConfig = getTestDMaaPMRSourcePluginConfig();
+ // blank out DMaaP MR Source Host
+ testDMaaPMRSourcePluginConfig.setHostName(null);
+ final DMaaPMRSource dMaaPMRSource = new DMaaPMRSource(testDMaaPMRSourcePluginConfig);
+ dMaaPMRSource.configurePipeline(pipelineConfigurer);
+ }
+
+
+ @Test
+ @SuppressWarnings("unchecked")
+ public void testGetStream() throws Exception {
+ final StreamingContext streamingContext = Mockito.mock(StreamingContext.class);
+ final JavaStreamingContext javaStreamingContext = Mockito.mock(JavaStreamingContext.class);
+ final JavaReceiverInputDStream dMaaPMRReceiver = Mockito.mock(JavaReceiverInputDStream.class);
+ when(streamingContext.getSparkStreamingContext()).thenReturn(javaStreamingContext);
+ when(javaStreamingContext.receiverStream(any(Receiver.class))).thenReturn(dMaaPMRReceiver);
+
+ final TestDMaaPMRSourcePluginConfig testDMaaPMRSourcePluginConfig = getTestDMaaPMRSourcePluginConfig();
+ final DMaaPMRSource dMaaPMRSource = new DMaaPMRSource(testDMaaPMRSourcePluginConfig);
+ final JavaDStream<StructuredRecord> stream = dMaaPMRSource.getStream(streamingContext);
+ assertNotNull(stream);
+ }
+}
diff --git a/dcae-analytics-cdap-plugins/src/test/java/org/openecomp/dcae/apod/analytics/cdap/plugins/streaming/dmaap/MockDMaaPMRReceiverTest.java b/dcae-analytics-cdap-plugins/src/test/java/org/openecomp/dcae/apod/analytics/cdap/plugins/streaming/dmaap/MockDMaaPMRReceiverTest.java index 183c90d..d2221ea 100644 --- a/dcae-analytics-cdap-plugins/src/test/java/org/openecomp/dcae/apod/analytics/cdap/plugins/streaming/dmaap/MockDMaaPMRReceiverTest.java +++ b/dcae-analytics-cdap-plugins/src/test/java/org/openecomp/dcae/apod/analytics/cdap/plugins/streaming/dmaap/MockDMaaPMRReceiverTest.java @@ -1,81 +1,81 @@ -/* - * ===============================LICENSE_START====================================== - * dcae-analytics - * ================================================================================ - * Copyright © 2017 AT&T Intellectual Property. All rights reserved. - * ================================================================================ - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - * ============================LICENSE_END=========================================== - */ - -package org.openecomp.dcae.apod.analytics.cdap.plugins.streaming.dmaap; - -import co.cask.cdap.api.data.format.StructuredRecord; -import org.apache.spark.storage.StorageLevel; -import org.junit.Test; -import org.openecomp.dcae.apod.analytics.cdap.plugins.BaseAnalyticsCDAPPluginsUnitTest; -import org.openecomp.dcae.apod.analytics.cdap.plugins.domain.config.dmaap.DMaaPMRSourcePluginConfig; -import org.openecomp.dcae.apod.analytics.cdap.plugins.domain.config.dmaap.TestDMaaPMRSourcePluginConfig; -import org.openecomp.dcae.apod.analytics.cdap.plugins.domain.schema.dmaap.DMaaPSourceOutputSchema; - -import java.util.concurrent.TimeUnit; - -/** - * @author Rajiv Singla . Creation Date: 2/20/2017. - */ -public class MockDMaaPMRReceiverTest extends BaseAnalyticsCDAPPluginsUnitTest { - - protected class TestMockDMaaPMRReceiverTest extends MockDMaaPMRReceiver { - - private boolean canStop = false; - - public TestMockDMaaPMRReceiverTest(StorageLevel storageLevel, DMaaPMRSourcePluginConfig pluginConfig) { - super(storageLevel, pluginConfig); - } - - @Override - public boolean isStopped() { - return canStop; - } - - @Override - public void store(StructuredRecord dataItem) { - LOG.debug("Mocking storing dataItem - {}", - dataItem.get(DMaaPSourceOutputSchema.FETCHED_MESSAGE.getSchemaColumnName())); - } - - public void setCanStop(boolean canStop) { - this.canStop = canStop; - } - } - - @Test - public void testStoreStructuredRecords() throws Exception { - final TestDMaaPMRSourcePluginConfig testDMaaPMRSourcePluginConfig = getTestDMaaPMRSourcePluginConfig(); - testDMaaPMRSourcePluginConfig.setPollingInterval(100); - final TestMockDMaaPMRReceiverTest mockDMaaPMRReceiver = new TestMockDMaaPMRReceiverTest(StorageLevel - .MEMORY_ONLY(), - testDMaaPMRSourcePluginConfig); - new Thread(new Runnable() { - @Override - public void run() { - mockDMaaPMRReceiver.storeStructuredRecords(null); - } - }).start(); - TimeUnit.MILLISECONDS.sleep(1000); - LOG.info("Killing Mock Subscriber after 1 ms"); - mockDMaaPMRReceiver.setCanStop(true); - - } - -} +/*
+ * ===============================LICENSE_START======================================
+ * dcae-analytics
+ * ================================================================================
+ * Copyright © 2017 AT&T Intellectual Property. All rights reserved.
+ * ================================================================================
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ * ============================LICENSE_END===========================================
+ */
+
+package org.openecomp.dcae.apod.analytics.cdap.plugins.streaming.dmaap;
+
+import co.cask.cdap.api.data.format.StructuredRecord;
+import org.apache.spark.storage.StorageLevel;
+import org.junit.Test;
+import org.openecomp.dcae.apod.analytics.cdap.plugins.BaseAnalyticsCDAPPluginsUnitTest;
+import org.openecomp.dcae.apod.analytics.cdap.plugins.domain.config.dmaap.DMaaPMRSourcePluginConfig;
+import org.openecomp.dcae.apod.analytics.cdap.plugins.domain.config.dmaap.TestDMaaPMRSourcePluginConfig;
+import org.openecomp.dcae.apod.analytics.cdap.plugins.domain.schema.dmaap.DMaaPSourceOutputSchema;
+
+import java.util.concurrent.TimeUnit;
+
+/**
+ * @author Rajiv Singla . Creation Date: 2/20/2017.
+ */
+public class MockDMaaPMRReceiverTest extends BaseAnalyticsCDAPPluginsUnitTest {
+
+ protected class TestMockDMaaPMRReceiverTest extends MockDMaaPMRReceiver {
+
+ private boolean canStop = false;
+
+ public TestMockDMaaPMRReceiverTest(StorageLevel storageLevel, DMaaPMRSourcePluginConfig pluginConfig) {
+ super(storageLevel, pluginConfig);
+ }
+
+ @Override
+ public boolean isStopped() {
+ return canStop;
+ }
+
+ @Override
+ public void store(StructuredRecord dataItem) {
+ LOG.debug("Mocking storing dataItem - {}",
+ dataItem.get(DMaaPSourceOutputSchema.FETCHED_MESSAGE.getSchemaColumnName()));
+ }
+
+ public void setCanStop(boolean canStop) {
+ this.canStop = canStop;
+ }
+ }
+
+ @Test
+ public void testStoreStructuredRecords() throws Exception {
+ final TestDMaaPMRSourcePluginConfig testDMaaPMRSourcePluginConfig = getTestDMaaPMRSourcePluginConfig();
+ testDMaaPMRSourcePluginConfig.setPollingInterval(100);
+ final TestMockDMaaPMRReceiverTest mockDMaaPMRReceiver = new TestMockDMaaPMRReceiverTest(StorageLevel
+ .MEMORY_ONLY(),
+ testDMaaPMRSourcePluginConfig);
+ new Thread(new Runnable() {
+ @Override
+ public void run() {
+ mockDMaaPMRReceiver.storeStructuredRecords(null);
+ }
+ }).start();
+ TimeUnit.MILLISECONDS.sleep(1000);
+ LOG.info("Killing Mock Subscriber after 1 ms");
+ mockDMaaPMRReceiver.setCanStop(true);
+
+ }
+
+}
diff --git a/dcae-analytics-cdap-plugins/src/test/java/org/openecomp/dcae/apod/analytics/cdap/plugins/streaming/dmaap/MockDMaaPMRSourceTest.java b/dcae-analytics-cdap-plugins/src/test/java/org/openecomp/dcae/apod/analytics/cdap/plugins/streaming/dmaap/MockDMaaPMRSourceTest.java index 11e09c5..7888d59 100644 --- a/dcae-analytics-cdap-plugins/src/test/java/org/openecomp/dcae/apod/analytics/cdap/plugins/streaming/dmaap/MockDMaaPMRSourceTest.java +++ b/dcae-analytics-cdap-plugins/src/test/java/org/openecomp/dcae/apod/analytics/cdap/plugins/streaming/dmaap/MockDMaaPMRSourceTest.java @@ -1,73 +1,74 @@ -/* - * ===============================LICENSE_START====================================== - * dcae-analytics - * ================================================================================ - * Copyright © 2017 AT&T Intellectual Property. All rights reserved. - * ================================================================================ - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - * ============================LICENSE_END=========================================== - */ - -package org.openecomp.dcae.apod.analytics.cdap.plugins.streaming.dmaap; - -import co.cask.cdap.api.data.format.StructuredRecord; -import co.cask.cdap.etl.api.streaming.StreamingContext; -import org.apache.spark.streaming.api.java.JavaDStream; -import org.apache.spark.streaming.api.java.JavaReceiverInputDStream; -import org.apache.spark.streaming.api.java.JavaStreamingContext; -import org.apache.spark.streaming.receiver.Receiver; -import org.junit.Test; -import org.mockito.Mockito; -import org.openecomp.dcae.apod.analytics.cdap.common.exception.CDAPSettingsException; -import org.openecomp.dcae.apod.analytics.cdap.plugins.BaseAnalyticsCDAPPluginsUnitTest; -import org.openecomp.dcae.apod.analytics.cdap.plugins.domain.config.dmaap.TestDMaaPMRSourcePluginConfig; - -import static org.junit.Assert.assertNotNull; -import static org.mockito.ArgumentMatchers.any; -import static org.mockito.Mockito.when; - -/** - * @author Rajiv Singla . Creation Date: 2/20/2017. - */ -public class MockDMaaPMRSourceTest extends BaseAnalyticsCDAPPluginsUnitTest { - - @Test - public void testGetStream() throws Exception { - final StreamingContext streamingContext = Mockito.mock(StreamingContext.class); - final JavaStreamingContext javaStreamingContext = Mockito.mock(JavaStreamingContext.class); - final JavaReceiverInputDStream dMaaPMRReceiver = Mockito.mock(JavaReceiverInputDStream.class); - when(streamingContext.getSparkStreamingContext()).thenReturn(javaStreamingContext); - when(javaStreamingContext.receiverStream(any(Receiver.class))).thenReturn(dMaaPMRReceiver); - - MockDMaaPMRSource mockDMaaPMRSource = new MockDMaaPMRSource(getTestDMaaPMRSourcePluginConfig()); - final JavaDStream<StructuredRecord> stream = mockDMaaPMRSource.getStream(streamingContext); - assertNotNull(stream); - } - - @Test(expected = CDAPSettingsException.class) - public void testConfigurePipelineWhenPollingIntervalNotPresent() throws Exception { - final TestDMaaPMRSourcePluginConfig testDMaaPMRSourcePluginConfig = getTestDMaaPMRSourcePluginConfig(); - testDMaaPMRSourcePluginConfig.setPollingInterval(null); - final MockDMaaPMRSource mockDMaaPMRSource = new MockDMaaPMRSource(testDMaaPMRSourcePluginConfig); - mockDMaaPMRSource.configurePipeline(null); - } - - @Test - public void testConfigurePipelineWhenPollingIntervalIsPresent() throws Exception { - final TestDMaaPMRSourcePluginConfig testDMaaPMRSourcePluginConfig = getTestDMaaPMRSourcePluginConfig(); - final MockDMaaPMRSource mockDMaaPMRSource = new MockDMaaPMRSource(testDMaaPMRSourcePluginConfig); - mockDMaaPMRSource.configurePipeline(null); - assertNotNull(mockDMaaPMRSource); - } - -} +/*
+ * ===============================LICENSE_START======================================
+ * dcae-analytics
+ * ================================================================================
+ * Copyright © 2017 AT&T Intellectual Property. All rights reserved.
+ * ================================================================================
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ * ============================LICENSE_END===========================================
+ */
+
+package org.openecomp.dcae.apod.analytics.cdap.plugins.streaming.dmaap;
+
+import co.cask.cdap.api.data.format.StructuredRecord;
+import co.cask.cdap.etl.api.streaming.StreamingContext;
+import org.apache.spark.streaming.api.java.JavaDStream;
+import org.apache.spark.streaming.api.java.JavaReceiverInputDStream;
+import org.apache.spark.streaming.api.java.JavaStreamingContext;
+import org.apache.spark.streaming.receiver.Receiver;
+import org.junit.Test;
+import org.mockito.Mockito;
+import org.openecomp.dcae.apod.analytics.cdap.common.exception.CDAPSettingsException;
+import org.openecomp.dcae.apod.analytics.cdap.plugins.BaseAnalyticsCDAPPluginsUnitTest;
+import org.openecomp.dcae.apod.analytics.cdap.plugins.domain.config.dmaap.TestDMaaPMRSourcePluginConfig;
+
+import static org.junit.Assert.assertNotNull;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.when;
+
+/**
+ * @author Rajiv Singla . Creation Date: 2/20/2017.
+ */
+@SuppressWarnings("unchecked")
+public class MockDMaaPMRSourceTest extends BaseAnalyticsCDAPPluginsUnitTest {
+
+ @Test
+ public void testGetStream() throws Exception {
+ final StreamingContext streamingContext = Mockito.mock(StreamingContext.class);
+ final JavaStreamingContext javaStreamingContext = Mockito.mock(JavaStreamingContext.class);
+ final JavaReceiverInputDStream dMaaPMRReceiver = Mockito.mock(JavaReceiverInputDStream.class);
+ when(streamingContext.getSparkStreamingContext()).thenReturn(javaStreamingContext);
+ when(javaStreamingContext.receiverStream(any(Receiver.class))).thenReturn(dMaaPMRReceiver);
+
+ MockDMaaPMRSource mockDMaaPMRSource = new MockDMaaPMRSource(getTestDMaaPMRSourcePluginConfig());
+ final JavaDStream<StructuredRecord> stream = mockDMaaPMRSource.getStream(streamingContext);
+ assertNotNull(stream);
+ }
+
+ @Test(expected = CDAPSettingsException.class)
+ public void testConfigurePipelineWhenPollingIntervalNotPresent() throws Exception {
+ final TestDMaaPMRSourcePluginConfig testDMaaPMRSourcePluginConfig = getTestDMaaPMRSourcePluginConfig();
+ testDMaaPMRSourcePluginConfig.setPollingInterval(null);
+ final MockDMaaPMRSource mockDMaaPMRSource = new MockDMaaPMRSource(testDMaaPMRSourcePluginConfig);
+ mockDMaaPMRSource.configurePipeline(null);
+ }
+
+ @Test
+ public void testConfigurePipelineWhenPollingIntervalIsPresent() throws Exception {
+ final TestDMaaPMRSourcePluginConfig testDMaaPMRSourcePluginConfig = getTestDMaaPMRSourcePluginConfig();
+ final MockDMaaPMRSource mockDMaaPMRSource = new MockDMaaPMRSource(testDMaaPMRSourcePluginConfig);
+ mockDMaaPMRSource.configurePipeline(null);
+ assertNotNull(mockDMaaPMRSource);
+ }
+
+}
diff --git a/dcae-analytics-cdap-plugins/src/test/java/org/openecomp/dcae/apod/analytics/cdap/plugins/streaming/dmaap/TestDMaaPMRReceiver.java b/dcae-analytics-cdap-plugins/src/test/java/org/openecomp/dcae/apod/analytics/cdap/plugins/streaming/dmaap/TestDMaaPMRReceiver.java index 1baf29c..2eab27b 100644 --- a/dcae-analytics-cdap-plugins/src/test/java/org/openecomp/dcae/apod/analytics/cdap/plugins/streaming/dmaap/TestDMaaPMRReceiver.java +++ b/dcae-analytics-cdap-plugins/src/test/java/org/openecomp/dcae/apod/analytics/cdap/plugins/streaming/dmaap/TestDMaaPMRReceiver.java @@ -1,58 +1,58 @@ -/* - * ===============================LICENSE_START====================================== - * dcae-analytics - * ================================================================================ - * Copyright © 2017 AT&T Intellectual Property. All rights reserved. - * ================================================================================ - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - * ============================LICENSE_END=========================================== - */ - -package org.openecomp.dcae.apod.analytics.cdap.plugins.streaming.dmaap; - -import co.cask.cdap.api.data.format.StructuredRecord; -import co.cask.cdap.api.metrics.Metrics; -import org.apache.spark.storage.StorageLevel; -import org.mockito.Mockito; -import org.openecomp.dcae.apod.analytics.cdap.plugins.domain.config.dmaap.DMaaPMRSourcePluginConfig; - -import static org.mockito.ArgumentMatchers.anyInt; -import static org.mockito.ArgumentMatchers.anyString; -import static org.mockito.Mockito.doNothing; - -/** - * Test implementation for {@link DMaaPMRReceiver} - * <p> - * @author Rajiv Singla . Creation Date: 1/24/2017. - */ -public class TestDMaaPMRReceiver extends DMaaPMRReceiver { - - protected static Metrics metrics; - - static { - metrics = Mockito.mock(Metrics.class); - doNothing().when(metrics).count(anyString(), anyInt()); - doNothing().when(metrics).gauge(anyString(), anyInt()); - } - - - public TestDMaaPMRReceiver(StorageLevel storageLevel, DMaaPMRSourcePluginConfig pluginConfig) { - - super(storageLevel, pluginConfig, metrics); - } - - @Override - public void store(StructuredRecord dataItem) { - // do nothing - } -} +/*
+ * ===============================LICENSE_START======================================
+ * dcae-analytics
+ * ================================================================================
+ * Copyright © 2017 AT&T Intellectual Property. All rights reserved.
+ * ================================================================================
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ * ============================LICENSE_END===========================================
+ */
+
+package org.openecomp.dcae.apod.analytics.cdap.plugins.streaming.dmaap;
+
+import co.cask.cdap.api.data.format.StructuredRecord;
+import co.cask.cdap.api.metrics.Metrics;
+import org.apache.spark.storage.StorageLevel;
+import org.mockito.Mockito;
+import org.openecomp.dcae.apod.analytics.cdap.plugins.domain.config.dmaap.DMaaPMRSourcePluginConfig;
+
+import static org.mockito.ArgumentMatchers.anyInt;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.Mockito.doNothing;
+
+/**
+ * Test implementation for {@link DMaaPMRReceiver}
+ * <p>
+ * @author Rajiv Singla . Creation Date: 1/24/2017.
+ */
+public class TestDMaaPMRReceiver extends DMaaPMRReceiver {
+
+ protected static Metrics metrics;
+
+ static {
+ metrics = Mockito.mock(Metrics.class);
+ doNothing().when(metrics).count(anyString(), anyInt());
+ doNothing().when(metrics).gauge(anyString(), anyInt());
+ }
+
+
+ public TestDMaaPMRReceiver(StorageLevel storageLevel, DMaaPMRSourcePluginConfig pluginConfig) {
+
+ super(storageLevel, pluginConfig, metrics);
+ }
+
+ @Override
+ public void store(StructuredRecord dataItem) {
+ // do nothing
+ }
+}
|