gerrit.onap Code Review - dcaegen2/services/prh.git/blob

   1 /*
   2  * ============LICENSE_START=======================================================
   3  * PNF-REGISTRATION-HANDLER
   4  * ================================================================================
   5  * Copyright (C) 2018 NOKIA Intellectual Property. All rights reserved.
   6  * ================================================================================
   7  * Licensed under the Apache License, Version 2.0 (the "License");
   8  * you may not use this file except in compliance with the License.
   9  * You may obtain a copy of the License at
  10  *
  11  *      http://www.apache.org/licenses/LICENSE-2.0
  12  *
  13  * Unless required by applicable law or agreed to in writing, software
  14  * distributed under the License is distributed on an "AS IS" BASIS,
  15  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  16  * See the License for the specific language governing permissions and
  17  * limitations under the License.
  18  * ============LICENSE_END=========================================================
  19  */
  20
  21 package org.onap.dcaegen2.services.prh.service;
  22
  23 import static org.mockito.Mockito.spy;
  24
  25 import com.google.gson.JsonElement;
  26 import com.google.gson.JsonParser;
  27 import java.util.Optional;
  28 import org.junit.jupiter.api.Assertions;
  29 import org.junit.jupiter.api.Test;
  30 import org.mockito.Mockito;
  31 import org.onap.dcaegen2.services.sdk.rest.services.model.ConsumerDmaapModel;
  32 import org.onap.dcaegen2.services.sdk.rest.services.model.ImmutableConsumerDmaapModel;
  33 import reactor.core.publisher.Mono;
  34 import reactor.test.StepVerifier;
  35
  36 /**
  37  * @author <a href="mailto:przemyslaw.wasala@nokia.com">Przemysław Wąsala</a> on 5/8/18
  38  */
  39 class DmaapConsumerJsonParserTest {
  40
  41     @Test
  42     void whenPassingCorrectJson_validationNotThrowingAnException() {
  43         //given
  44         String message = "[{\"event\": {"
  45             + "\"commonEventHeader\": { \"sourceName\":\"NOKQTFCOC540002E\"},"
  46             + "\"pnfRegistrationFields\": {"
  47             + " \"unitType\": \"AirScale\","
  48             + " \"serialNumber\": \"QTFCOC540002E\","
  49             + " \"pnfRegistrationFieldsVersion\": \"2.0\","
  50             + " \"manufactureDate\": \"1535014037024\","
  51             + " \"modelNumber\": \"7BEA\",\n"
  52             + " \"lastServiceDate\": \"1535014037024\","
  53             + " \"unitFamily\": \"BBU\","
  54             + " \"vendorName\": \"Nokia\","
  55             + " \"oamV4IpAddress\": \"10.16.123.234\","
  56             + " \"softwareVersion\": \"v4.5.0.1\","
  57             + " \"oamV6IpAddress\": \"0:0:0:0:0:FFFF:0A10:7BEA\""
  58             + "}}}]";
  59
  60         String parsed = "{\"event\": {"
  61             + "\"commonEventHeader\": { \"sourceName\":\"NOKQTFCOC540002E\"},"
  62             + "\"pnfRegistrationFields\": {"
  63             + " \"unitType\": \"AirScale\","
  64             + " \"serialNumber\": \"QTFCOC540002E\","
  65             + " \"pnfRegistrationFieldsVersion\": \"2.0\","
  66             + " \"manufactureDate\": \"1535014037024\","
  67             + " \"modelNumber\": \"7BEA\",\n"
  68             + " \"lastServiceDate\": \"1535014037024\","
  69             + " \"unitFamily\": \"BBU\","
  70             + " \"vendorName\": \"Nokia\","
  71             + " \"oamV4IpAddress\": \"10.16.123.234\","
  72             + " \"softwareVersion\": \"v4.5.0.1\","
  73             + " \"oamV6IpAddress\": \"0:0:0:0:0:FFFF:0A10:7BEA\""
  74             + "}}}";
  75
  76         ConsumerDmaapModel expectedObject = ImmutableConsumerDmaapModel.builder().ipv4("10.16.123.234")
  77             .ipv6("0:0:0:0:0:FFFF:0A10:7BEA")
  78             .correlationId("NOKQTFCOC540002E").build();
  79         //when
  80         DmaapConsumerJsonParser dmaapConsumerJsonParser = spy(new DmaapConsumerJsonParser());
  81         JsonElement jsonElement = new JsonParser().parse(parsed);
  82         Mockito.doReturn(Optional.of(jsonElement.getAsJsonObject()))
  83             .when(dmaapConsumerJsonParser).getJsonObjectFromAnArray(jsonElement);
  84         ConsumerDmaapModel consumerDmaapModel = dmaapConsumerJsonParser
  85             .getJsonObject(Mono.just((message))).blockFirst();
  86         //then
  87         Assertions.assertNotNull(consumerDmaapModel);
  88         Assertions.assertEquals(expectedObject, consumerDmaapModel);
  89     }
  90
  91     @Test
  92     void whenPassingCorrectJsonWithoutIpv4_validationNotThrowingAnException() {
  93         //given
  94         String message = "[{\"event\": {"
  95             + "\"commonEventHeader\": { \"sourceName\":\"NOKQTFCOC540002E\"},"
  96             + "\"pnfRegistrationFields\": {"
  97             + " \"unitType\": \"AirScale\","
  98             + " \"serialNumber\": \"QTFCOC540002E\","
  99             + " \"pnfRegistrationFieldsVersion\": \"2.0\","
 100             + " \"manufactureDate\": \"1535014037024\","
 101             + " \"modelNumber\": \"7BEA\",\n"
 102             + " \"lastServiceDate\": \"1535014037024\","
 103             + " \"unitFamily\": \"BBU\","
 104             + " \"vendorName\": \"Nokia\","
 105             + " \"softwareVersion\": \"v4.5.0.1\","
 106             + " \"oamV6IpAddress\": \"0:0:0:0:0:FFFF:0A10:7BEA\""
 107             + "}}}]";
 108
 109         String parsed = "{\"event\": {"
 110             + "\"commonEventHeader\": { \"sourceName\":\"NOKQTFCOC540002E\"},"
 111             + "\"pnfRegistrationFields\": {"
 112             + " \"unitType\": \"AirScale\","
 113             + " \"serialNumber\": \"QTFCOC540002E\","
 114             + " \"pnfRegistrationFieldsVersion\": \"2.0\","
 115             + " \"manufactureDate\": \"1535014037024\","
 116             + " \"modelNumber\": \"7BEA\",\n"
 117             + " \"lastServiceDate\": \"1535014037024\","
 118             + " \"unitFamily\": \"BBU\","
 119             + " \"vendorName\": \"Nokia\","
 120             + " \"softwareVersion\": \"v4.5.0.1\","
 121             + " \"oamV6IpAddress\": \"0:0:0:0:0:FFFF:0A10:7BEA\""
 122             + "}}}";
 123
 124         //when
 125         DmaapConsumerJsonParser dmaapConsumerJsonParser = spy(new DmaapConsumerJsonParser());
 126         JsonElement jsonElement = new JsonParser().parse(parsed);
 127         Mockito.doReturn(Optional.of(jsonElement.getAsJsonObject()))
 128             .when(dmaapConsumerJsonParser).getJsonObjectFromAnArray(jsonElement);
 129         dmaapConsumerJsonParser.getJsonObject(Mono.just((message)));
 130         ConsumerDmaapModel consumerDmaapModel = dmaapConsumerJsonParser.getJsonObject(Mono.just((message)))
 131             .blockFirst();
 132         //then
 133         ConsumerDmaapModel expectedObject = ImmutableConsumerDmaapModel.builder().ipv4("")
 134             .ipv6("0:0:0:0:0:FFFF:0A10:7BEA")
 135             .correlationId("NOKQTFCOC540002E").build();
 136         Assertions.assertNotNull(consumerDmaapModel);
 137         Assertions.assertEquals(expectedObject, consumerDmaapModel);
 138     }
 139
 140     @Test
 141     void whenPassingCorrectJsonWithoutIpv6_validationNotThrowingAnException() {
 142         //given
 143         String message = "[{\"event\": {"
 144             + "\"commonEventHeader\": { \"sourceName\":\"NOKQTFCOC540002E\"},"
 145             + "\"pnfRegistrationFields\": {"
 146             + " \"unitType\": \"AirScale\","
 147             + " \"serialNumber\": \"QTFCOC540002E\","
 148             + " \"pnfRegistrationFieldsVersion\": \"2.0\","
 149             + " \"manufactureDate\": \"1535014037024\","
 150             + " \"modelNumber\": \"7BEA\",\n"
 151             + " \"lastServiceDate\": \"1535014037024\","
 152             + " \"unitFamily\": \"BBU\","
 153             + " \"vendorName\": \"Nokia\","
 154             + " \"oamV4IpAddress\": \"10.16.123.234\","
 155             + " \"softwareVersion\": \"v4.5.0.1\""
 156             + "}}}]";
 157
 158         String parsed = "{\"event\": {"
 159             + "\"commonEventHeader\": { \"sourceName\":\"NOKQTFCOC540002E\"},"
 160             + "\"pnfRegistrationFields\": {"
 161             + " \"unitType\": \"AirScale\","
 162             + " \"serialNumber\": \"QTFCOC540002E\","
 163             + " \"pnfRegistrationFieldsVersion\": \"2.0\","
 164             + " \"manufactureDate\": \"1535014037024\","
 165             + " \"modelNumber\": \"7BEA\",\n"
 166             + " \"lastServiceDate\": \"1535014037024\","
 167             + " \"unitFamily\": \"BBU\","
 168             + " \"vendorName\": \"Nokia\","
 169             + " \"oamV4IpAddress\": \"10.16.123.234\","
 170             + " \"softwareVersion\": \"v4.5.0.1\""
 171             + "}}}";
 172
 173         ConsumerDmaapModel expectedObject = ImmutableConsumerDmaapModel.builder().ipv4("10.16.123.234").ipv6("")
 174             .correlationId("NOKQTFCOC540002E").build();
 175         //when
 176         DmaapConsumerJsonParser dmaapConsumerJsonParser = spy(new DmaapConsumerJsonParser());
 177         JsonElement jsonElement = new JsonParser().parse(parsed);
 178         Mockito.doReturn(Optional.of(jsonElement.getAsJsonObject()))
 179             .when(dmaapConsumerJsonParser).getJsonObjectFromAnArray(jsonElement);
 180         ConsumerDmaapModel consumerDmaapModel = dmaapConsumerJsonParser.getJsonObject(Mono.just((message)))
 181             .blockFirst();
 182         //then
 183         Assertions.assertNotNull(consumerDmaapModel);
 184         Assertions.assertEquals(expectedObject, consumerDmaapModel);
 185     }
 186
 187     @Test
 188     void whenPassingCorrectJsonWithoutIpv4andIpv6_validationAddingAnException() {
 189         //given
 190         String message = "[{\"event\": {"
 191             + "\"commonEventHeader\": { \"sourceName\":\"NOKQTFCOC540002E\"},"
 192             + "\"pnfRegistrationFields\": {"
 193             + " \"unitType\": \"AirScale\","
 194             + " \"serialNumber\": \"QTFCOC540002E\","
 195             + " \"pnfRegistrationFieldsVersion\": \"2.0\","
 196             + " \"manufactureDate\": \"1535014037024\","
 197             + " \"modelNumber\": \"7BEA\",\n"
 198             + " \"lastServiceDate\": \"1535014037024\","
 199             + " \"unitFamily\": \"BBU\","
 200             + " \"vendorName\": \"Nokia\","
 201             + " \"softwareVersion\": \"v4.5.0.1\""
 202             + "}}}]";
 203
 204         String parsed = "{\"event\": {"
 205             + "\"commonEventHeader\": { \"sourceName\":\"NOKQTFCOC540002E\"},"
 206             + "\"pnfRegistrationFields\": {"
 207             + " \"unitType\": \"AirScale\","
 208             + " \"serialNumber\": \"QTFCOC540002E\","
 209             + " \"pnfRegistrationFieldsVersion\": \"2.0\","
 210             + " \"manufactureDate\": \"1535014037024\","
 211             + " \"modelNumber\": \"7BEA\",\n"
 212             + " \"lastServiceDate\": \"1535014037024\","
 213             + " \"unitFamily\": \"BBU\","
 214             + " \"vendorName\": \"Nokia\","
 215             + " \"softwareVersion\": \"v4.5.0.1\""
 216             + "}}}";
 217
 218         DmaapConsumerJsonParser dmaapConsumerJsonParser = spy(new DmaapConsumerJsonParser());
 219         JsonElement jsonElement = new JsonParser().parse(parsed);
 220         Mockito.doReturn(Optional.of(jsonElement.getAsJsonObject()))
 221             .when(dmaapConsumerJsonParser).getJsonObjectFromAnArray(jsonElement);
 222         StepVerifier.create(dmaapConsumerJsonParser.getJsonObject(Mono.just(message)))
 223             .expectSubscription().thenRequest(1).verifyComplete();
 224
 225     }
 226
 227     @Test
 228     void whenPassingJsonWithoutMandatoryHeaderInformation_validationAddingAnException() {
 229         String parsed = "{\"event\": {"
 230             + "\"commonEventHeader\": {},"
 231             + "\"pnfRegistrationFields\": {"
 232             + " \"unitType\": \"AirScale\","
 233             + " \"serialNumber\": \"QTFCOC540002E\","
 234             + " \"pnfRegistrationFieldsVersion\": \"2.0\","
 235             + " \"manufactureDate\": \"1535014037024\","
 236             + " \"modelNumber\": \"7BEA\",\n"
 237             + " \"lastServiceDate\": \"1535014037024\","
 238             + " \"unitFamily\": \"BBU\","
 239             + " \"vendorName\": \"Nokia\","
 240             + " \"softwareVersion\": \"v4.5.0.1\""
 241             + "}}}";
 242
 243         DmaapConsumerJsonParser dmaapConsumerJsonParser = spy(new DmaapConsumerJsonParser());
 244         JsonElement jsonElement = new JsonParser().parse(parsed);
 245         Mockito.doReturn(Optional.of(jsonElement.getAsJsonObject()))
 246             .when(dmaapConsumerJsonParser).getJsonObjectFromAnArray(jsonElement);
 247         String incorrectMessage = "[{\"event\": {"
 248             + "\"commonEventHeader\": {},"
 249             + "\"pnfRegistrationFields\": {"
 250             + " \"unitType\": \"AirScale\","
 251             + " \"serialNumber\": \"QTFCOC540002E\","
 252             + " \"pnfRegistrationFieldsVersion\": \"2.0\","
 253             + " \"manufactureDate\": \"1535014037024\","
 254             + " \"modelNumber\": \"7BEA\",\n"
 255             + " \"lastServiceDate\": \"1535014037024\","
 256             + " \"unitFamily\": \"BBU\","
 257             + " \"vendorName\": \"Nokia\","
 258             + " \"softwareVersion\": \"v4.5.0.1\""
 259             + "}}}]";
 260         StepVerifier.create(dmaapConsumerJsonParser.getJsonObject(Mono.just(incorrectMessage)))
 261             .expectSubscription().thenRequest(1).verifyComplete();
 262     }
 263
 264     @Test
 265     void whenPassingJsonWithoutSourceName_validationAddingAnException() {
 266         String parsed = "{\"event\": {"
 267             + "\"commonEventHeader\": {},"
 268             + "\"pnfRegistrationFields\": {"
 269             + " \"unitType\": \"AirScale\","
 270             + " \"serialNumber\": \"QTFCOC540002E\","
 271             + " \"pnfRegistrationFieldsVersion\": \"2.0\","
 272             + " \"manufactureDate\": \"1535014037024\","
 273             + " \"modelNumber\": \"7BEA\",\n"
 274             + " \"lastServiceDate\": \"1535014037024\","
 275             + " \"unitFamily\": \"BBU\","
 276             + " \"vendorName\": \"Nokia\","
 277             + " \"softwareVersion\": \"v4.5.0.1\","
 278             + " \"oamV6IpAddress\": \"0:0:0:0:0:FFFF:0A10:7BEA\""
 279             + "}}}";
 280
 281         DmaapConsumerJsonParser dmaapConsumerJsonParser = spy(new DmaapConsumerJsonParser());
 282         JsonElement jsonElement = new JsonParser().parse(parsed);
 283         Mockito.doReturn(Optional.of(jsonElement.getAsJsonObject()))
 284             .when(dmaapConsumerJsonParser).getJsonObjectFromAnArray(jsonElement);
 285         String jsonWithoutSourceName =
 286             "[{\"event\": {"
 287                 + "\"commonEventHeader\": {},"
 288                 + "\"pnfRegistrationFields\": {"
 289                 + " \"unitType\": \"AirScale\","
 290                 + " \"serialNumber\": \"QTFCOC540002E\","
 291                 + " \"pnfRegistrationFieldsVersion\": \"2.0\","
 292                 + " \"manufactureDate\": \"1535014037024\","
 293                 + " \"modelNumber\": \"7BEA\",\n"
 294                 + " \"lastServiceDate\": \"1535014037024\","
 295                 + " \"unitFamily\": \"BBU\","
 296                 + " \"vendorName\": \"Nokia\","
 297                 + " \"softwareVersion\": \"v4.5.0.1\","
 298                 + " \"oamV6IpAddress\": \"0:0:0:0:0:FFFF:0A10:7BEA\""
 299                 + "}}}]";
 300         StepVerifier
 301             .create(dmaapConsumerJsonParser.getJsonObject(Mono.just(jsonWithoutSourceName)))
 302             .expectSubscription().thenRequest(1)
 303             .verifyComplete();
 304     }
 305
 306     @Test
 307     void whenPassingJsonWithoutIpInformation_validationAddingAnException() {
 308         String parsed =
 309             "{\"event\": {"
 310                 + "\"commonEventHeader\": {\"sourceName\": \"NOKQTFCOC540002E\"},"
 311                 + "\"pnfRegistrationFields\": {"
 312                 + " \"unitType\": \"AirScale\","
 313                 + " \"serialNumber\": \"QTFCOC540002E\","
 314                 + " \"pnfRegistrationFieldsVersion\": \"2.0\","
 315                 + " \"manufactureDate\": \"1535014037024\","
 316                 + " \"modelNumber\": \"7BEA\",\n"
 317                 + " \"lastServiceDate\": \"1535014037024\","
 318                 + " \"unitFamily\": \"BBU\","
 319                 + " \"vendorName\": \"Nokia\","
 320                 + " \"softwareVersion\": \"v4.5.0.1\","
 321                 + " \"oamV4IpAddress\": \"\","
 322                 + " \"oamV6IpAddress\": \"\""
 323                 + "}}}";
 324
 325         DmaapConsumerJsonParser dmaapConsumerJsonParser = spy(new DmaapConsumerJsonParser());
 326         JsonElement jsonElement = new JsonParser().parse(parsed);
 327         Mockito.doReturn(Optional.of(jsonElement.getAsJsonObject()))
 328             .when(dmaapConsumerJsonParser).getJsonObjectFromAnArray(jsonElement);
 329         String jsonWithoutIpInformation =
 330             "[{\"event\": {"
 331                 + "\"commonEventHeader\": {\"sourceName\": \"NOKQTFCOC540002E\"},"
 332                 + "\"pnfRegistrationFields\": {"
 333                 + " \"unitType\": \"AirScale\","
 334                 + " \"serialNumber\": \"QTFCOC540002E\","
 335                 + " \"pnfRegistrationFieldsVersion\": \"2.0\","
 336                 + " \"manufactureDate\": \"1535014037024\","
 337                 + " \"modelNumber\": \"7BEA\",\n"
 338                 + " \"lastServiceDate\": \"1535014037024\","
 339                 + " \"unitFamily\": \"BBU\","
 340                 + " \"vendorName\": \"Nokia\","
 341                 + " \"softwareVersion\": \"v4.5.0.1\","
 342                 + " \"oamV4IpAddress\": \"\","
 343                 + " \"oamV6IpAddress\": \"\""
 344                 + "}}}]";
 345         StepVerifier.create(dmaapConsumerJsonParser.getJsonObject(Mono.just(jsonWithoutIpInformation)))
 346             .expectSubscription().thenRequest(1).verifyComplete();
 347     }
 348
 349     @Test
 350     void whenPassingJsonWithoutSourceNameValue_validationAddingAnException() {
 351         String parsed =
 352             "{\"event\": {"
 353                 + "\"commonEventHeader\": {\"sourceName\": \"\"},"
 354                 + "\"pnfRegistrationFields\": {"
 355                 + " \"unitType\": \"AirScale\","
 356                 + " \"serialNumber\": \"QTFCOC540002E\","
 357                 + " \"pnfRegistrationFieldsVersion\": \"2.0\","
 358                 + " \"manufactureDate\": \"1535014037024\","
 359                 + " \"modelNumber\": \"7BEA\",\n"
 360                 + " \"lastServiceDate\": \"1535014037024\","
 361                 + " \"unitFamily\": \"BBU\","
 362                 + " \"vendorName\": \"Nokia\","
 363                 + " \"softwareVersion\": \"v4.5.0.1\","
 364                 + " \"oamV4IpAddress\": \"10.16.123.234\","
 365                 + " \"oamV6IpAddress\": \"0:0:0:0:0:FFFF:0A10:7BEA\""
 366                 + "}}}";
 367
 368         DmaapConsumerJsonParser dmaapConsumerJsonParser = spy(new DmaapConsumerJsonParser());
 369         JsonElement jsonElement = new JsonParser().parse(parsed);
 370         Mockito.doReturn(Optional.of(jsonElement.getAsJsonObject()))
 371             .when(dmaapConsumerJsonParser).getJsonObjectFromAnArray(jsonElement);
 372         String jsonWithoutIpInformation =
 373             "[{\"event\": {"
 374                 + "\"commonEventHeader\": {\"sourceName\": \"\"},"
 375                 + "\"pnfRegistrationFields\": {"
 376                 + " \"unitType\": \"AirScale\","
 377                 + " \"serialNumber\": \"QTFCOC540002E\","
 378                 + " \"pnfRegistrationFieldsVersion\": \"2.0\","
 379                 + " \"manufactureDate\": \"1535014037024\","
 380                 + " \"modelNumber\": \"7BEA\",\n"
 381                 + " \"lastServiceDate\": \"1535014037024\","
 382                 + " \"unitFamily\": \"BBU\","
 383                 + " \"vendorName\": \"Nokia\","
 384                 + " \"softwareVersion\": \"v4.5.0.1\","
 385                 + " \"oamV4IpAddress\": \"10.16.123.234\","
 386                 + " \"oamV6IpAddress\": \"0:0:0:0:0:FFFF:0A10:7BEA\""
 387                 + "}}}]";
 388         StepVerifier.create(dmaapConsumerJsonParser.getJsonObject(Mono.just(jsonWithoutIpInformation)))
 389             .expectSubscription().thenRequest(1).verifyComplete();
 390     }
 391
 392     @Test
 393     void whenPassingCorrectJsoArraynWithoutIpv4_validationNotThrowingAnException() {
 394         //given
 395         String message = "[{\"event\": {"
 396             + "\"commonEventHeader\": { \"sourceName\":\"NOKQTFCOC540002E\"},"
 397             + "\"pnfRegistrationFields\": {"
 398             + " \"unitType\": \"AirScale\","
 399             + " \"serialNumber\": \"QTFCOC540002E\","
 400             + " \"pnfRegistrationFieldsVersion\": \"2.0\","
 401             + " \"manufactureDate\": \"1535014037024\","
 402             + " \"modelNumber\": \"7BEA\",\n"
 403             + " \"lastServiceDate\": \"1535014037024\","
 404             + " \"unitFamily\": \"BBU\","
 405             + " \"vendorName\": \"Nokia\","
 406             + " \"oamV4IpAddress\": \"10.16.123.234\","
 407             + " \"softwareVersion\": \"v4.5.0.1\","
 408             + " \"oamV6IpAddress\": \"0:0:0:0:0:FFFF:0A10:7BEA\""
 409             + "}}},"
 410             + "{\"event\": {"
 411             + "\"commonEventHeader\": { \"sourceName\":\"NOKQTFCOC540002E\"},"
 412             + "\"pnfRegistrationFields\": {"
 413             + " \"unitType\": \"AirScale\","
 414             + " \"serialNumber\": \"QTFCOC540002E\","
 415             + " \"pnfRegistrationFieldsVersion\": \"2.0\","
 416             + " \"manufactureDate\": \"1535014037024\","
 417             + " \"modelNumber\": \"7BEA\",\n"
 418             + " \"lastServiceDate\": \"1535014037024\","
 419             + " \"unitFamily\": \"BBU\","
 420             + " \"vendorName\": \"Nokia\","
 421             + " \"oamV4IpAddress\": \"10.16.123.234\","
 422             + " \"softwareVersion\": \"v4.5.0.1\","
 423             + " \"oamV6IpAddress\": \"0:0:0:0:0:FFFF:0A10:7BEA\""
 424             + "}}}"
 425             + "]";
 426
 427         ConsumerDmaapModel expectedObject = ImmutableConsumerDmaapModel.builder().ipv4("10.16.123.234")
 428             .ipv6("0:0:0:0:0:FFFF:0A10:7BEA")
 429             .correlationId("NOKQTFCOC540002E").build();
 430         //when
 431         DmaapConsumerJsonParser dmaapConsumerJsonParser = new DmaapConsumerJsonParser();
 432
 433         //then
 434         StepVerifier.create(dmaapConsumerJsonParser.getJsonObject(Mono.just(message)))
 435             .expectSubscription().expectNext(expectedObject).expectNext(expectedObject).verifyComplete();
 436     }
 437 }