dcae-analytics-cdap-plugins/src/test/java/org/openecomp/dcae/apod/analytics/cdap/plugins/batch/sink/dmaap/DMaaPMRSinkTest.java

   1 /*\r
   2  * ===============================LICENSE_START======================================\r
   3  *  dcae-analytics\r
   4  * ================================================================================\r
   5  *    Copyright © 2017 AT&T Intellectual Property. All rights reserved.\r
   6  * ================================================================================\r
   7  *  Licensed under the Apache License, Version 2.0 (the "License");\r
   8  *  you may not use this file except in compliance with the License.\r
   9  *   You may obtain a copy of the License at\r
  10  *\r
  11  *          http://www.apache.org/licenses/LICENSE-2.0\r
  12  *\r
  13  *  Unless required by applicable law or agreed to in writing, software\r
  14  *  distributed under the License is distributed on an "AS IS" BASIS,\r
  15  *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\r
  16  *  See the License for the specific language governing permissions and\r
  17  *  limitations under the License.\r
  18  *  ============================LICENSE_END===========================================\r
  19  */\r
  20 \r
  21 package org.openecomp.dcae.apod.analytics.cdap.plugins.batch.sink.dmaap;\r
  22 \r
  23 import co.cask.cdap.api.data.format.StructuredRecord;\r
  24 import co.cask.cdap.api.data.schema.Schema;\r
  25 import co.cask.cdap.etl.api.Emitter;\r
  26 import co.cask.cdap.etl.api.PipelineConfigurer;\r
  27 import co.cask.cdap.etl.api.StageConfigurer;\r
  28 import co.cask.cdap.etl.api.batch.BatchSinkContext;\r
  29 import org.junit.Before;\r
  30 import org.junit.Test;\r
  31 import org.mockito.Mockito;\r
  32 import org.openecomp.dcae.apod.analytics.cdap.common.exception.CDAPSettingsException;\r
  33 import org.openecomp.dcae.apod.analytics.cdap.plugins.BaseAnalyticsCDAPPluginsUnitTest;\r
  34 \r
  35 import static org.mockito.ArgumentMatchers.any;\r
  36 import static org.mockito.ArgumentMatchers.eq;\r
  37 import static org.mockito.Mockito.doNothing;\r
  38 import static org.mockito.Mockito.times;\r
  39 import static org.mockito.Mockito.verify;\r
  40 import static org.mockito.Mockito.when;\r
  41 \r
  42 /**\r
  43  * @author Rajiv Singla . Creation Date: 1/30/2017.\r
  44  */\r
  45 public class DMaaPMRSinkTest extends BaseAnalyticsCDAPPluginsUnitTest {\r
  46 \r
  47     private DMaaPMRSink dMaaPMRSink;\r
  48 \r
  49     @Before\r
  50     public void before() {\r
  51         dMaaPMRSink = new DMaaPMRSink(getTestDMaaPMRSinkPluginConfig());\r
  52     }\r
  53 \r
  54     @Test\r
  55     public void testConfigurePipeline() throws Exception {\r
  56         final PipelineConfigurer pipelineConfigurer = Mockito.mock(PipelineConfigurer.class);\r
  57         final StageConfigurer stageConfigurer = Mockito.mock(StageConfigurer.class);\r
  58         when(pipelineConfigurer.getStageConfigurer()).thenReturn(stageConfigurer);\r
  59         when(stageConfigurer.getInputSchema()).thenReturn(getDMaaPMRSinkTestSchema());\r
  60         dMaaPMRSink.configurePipeline(pipelineConfigurer);\r
  61         verify(stageConfigurer, times(1)).getInputSchema();\r
  62     }\r
  63 \r
  64     @Test(expected = CDAPSettingsException.class)\r
  65     public void testConfigurePipelineWithInvalidSchema() throws Exception {\r
  66         final PipelineConfigurer pipelineConfigurer = Mockito.mock(PipelineConfigurer.class);\r
  67         final StageConfigurer stageConfigurer = Mockito.mock(StageConfigurer.class);\r
  68         when(pipelineConfigurer.getStageConfigurer()).thenReturn(stageConfigurer);\r
  69         when(stageConfigurer.getInputSchema()).thenReturn(Schema.recordOf(\r
  70                 "DMaaPMRSinkInvalidSchema",\r
  71                 Schema.Field.of("message1", Schema.of(Schema.Type.STRING)),\r
  72                 Schema.Field.of("field1", Schema.of(Schema.Type.STRING))\r
  73         ));\r
  74         dMaaPMRSink.configurePipeline(pipelineConfigurer);\r
  75     }\r
  76 \r
  77     @Test\r
  78     public void testPrepareRun() throws Exception {\r
  79         final BatchSinkContext batchSinkContext = Mockito.mock(BatchSinkContext.class);\r
  80         dMaaPMRSink.prepareRun(batchSinkContext);\r
  81     }\r
  82 \r
  83     @Test\r
  84     @SuppressWarnings("unchecked")\r
  85     public void testTransform() throws Exception {\r
  86         final StructuredRecord structuredRecord = Mockito.mock(StructuredRecord.class);\r
  87         final Emitter emitter = Mockito.mock(Emitter.class);\r
  88         final String incomingTestMessage = "test message";\r
  89         when(structuredRecord.get(\r
  90                 eq(getTestDMaaPMRSinkPluginConfig().getMessageColumnName()))).thenReturn(incomingTestMessage);\r
  91         doNothing().when(emitter).emit(any());\r
  92         dMaaPMRSink.transform(structuredRecord, emitter);\r
  93     }\r
  94 \r
  95 }\r