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

   1 /*
   2  * ============LICENSE_START=======================================================
   3  * BBS-RELOCATION-CPE-AUTHENTICATION-HANDLER
   4  * ================================================================================
   5  * Copyright (C) 2019 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.bbs.event.processor.pipelines;
  22
  23 import static org.junit.jupiter.api.Assertions.assertEquals;
  24 import static org.mockito.ArgumentMatchers.any;
  25 import static org.mockito.ArgumentMatchers.anyString;
  26 import static org.mockito.Mockito.times;
  27 import static org.mockito.Mockito.verify;
  28 import static org.mockito.Mockito.verifyNoMoreInteractions;
  29 import static org.mockito.Mockito.verifyZeroInteractions;
  30 import static org.mockito.Mockito.when;
  31 import static org.onap.bbs.event.processor.config.ApplicationConstants.CONSUME_REREGISTRATION_TASK_NAME;
  32 import static org.onap.bbs.event.processor.config.ApplicationConstants.RETRIEVE_HSI_CFS_SERVICE_INSTANCE_TASK_NAME;
  33 import static org.onap.bbs.event.processor.config.ApplicationConstants.RETRIEVE_PNF_TASK_NAME;
  34
  35 import java.util.Arrays;
  36 import java.util.Collections;
  37 import java.util.HashMap;
  38 import java.util.UUID;
  39
  40 import javax.net.ssl.SSLException;
  41
  42 import org.junit.jupiter.api.BeforeEach;
  43 import org.junit.jupiter.api.DisplayName;
  44 import org.junit.jupiter.api.Test;
  45 import org.mockito.Mockito;
  46 import org.onap.bbs.event.processor.config.ApplicationConfiguration;
  47 import org.onap.bbs.event.processor.exceptions.AaiTaskException;
  48 import org.onap.bbs.event.processor.exceptions.EmptyDmaapResponseException;
  49 import org.onap.bbs.event.processor.model.ControlLoopPublisherDmaapModel;
  50 import org.onap.bbs.event.processor.model.ImmutableMetadataEntryAaiObject;
  51 import org.onap.bbs.event.processor.model.ImmutableMetadataListAaiObject;
  52 import org.onap.bbs.event.processor.model.ImmutablePnfAaiObject;
  53 import org.onap.bbs.event.processor.model.ImmutablePropertyAaiObject;
  54 import org.onap.bbs.event.processor.model.ImmutableReRegistrationConsumerDmaapModel;
  55 import org.onap.bbs.event.processor.model.ImmutableRelationshipDataEntryAaiObject;
  56 import org.onap.bbs.event.processor.model.ImmutableRelationshipEntryAaiObject;
  57 import org.onap.bbs.event.processor.model.ImmutableRelationshipListAaiObject;
  58 import org.onap.bbs.event.processor.model.ImmutableServiceInstanceAaiObject;
  59 import org.onap.bbs.event.processor.model.MetadataListAaiObject;
  60 import org.onap.bbs.event.processor.model.PnfAaiObject;
  61 import org.onap.bbs.event.processor.model.ReRegistrationConsumerDmaapModel;
  62 import org.onap.bbs.event.processor.model.RelationshipListAaiObject;
  63 import org.onap.bbs.event.processor.model.ServiceInstanceAaiObject;
  64 import org.onap.bbs.event.processor.tasks.AaiClientTask;
  65 import org.onap.bbs.event.processor.tasks.DmaapPublisherTask;
  66 import org.onap.bbs.event.processor.tasks.DmaapReRegistrationConsumerTask;
  67 import org.onap.dcaegen2.services.sdk.rest.services.adapters.http.HttpResponse;
  68 import org.springframework.http.HttpStatus;
  69
  70 import reactor.core.publisher.Flux;
  71 import reactor.core.publisher.Mono;
  72 import reactor.test.StepVerifier;
  73
  74 @DisplayName("PNF Re-registration Pipeline Unit-Tests")
  75 class ReRegistrationPipelineTest {
  76
  77     private ReRegistrationPipeline pipeline;
  78     private ApplicationConfiguration configuration;
  79     private DmaapReRegistrationConsumerTask consumerTask;
  80     private DmaapPublisherTask publisherTask;
  81     private AaiClientTask aaiClientTask;
  82
  83     private HttpResponse httpResponse;
  84
  85     @BeforeEach
  86     void setup() {
  87
  88         httpResponse = Mockito.mock(HttpResponse.class);
  89
  90         configuration = Mockito.mock(ApplicationConfiguration.class);
  91         consumerTask = Mockito.mock(DmaapReRegistrationConsumerTask.class);
  92         publisherTask = Mockito.mock(DmaapPublisherTask.class);
  93         aaiClientTask = Mockito.mock(AaiClientTask.class);
  94
  95         when(configuration.getReRegistrationCloseLoopControlName())
  96                 .thenReturn("controlName");
  97         when(configuration.getReRegistrationCloseLoopPolicyScope())
  98                 .thenReturn("policyScope");
  99         when(configuration.getPolicyVersion())
 100                 .thenReturn("1.0.0");
 101         when(configuration.getCloseLoopTargetType())
 102                 .thenReturn("VM");
 103         when(configuration.getCloseLoopEventStatus())
 104                 .thenReturn("ONSET");
 105         when(configuration.getCloseLoopVersion())
 106                 .thenReturn("1.0.2");
 107         when(configuration.getCloseLoopTarget())
 108                 .thenReturn("CL-Target");
 109         when(configuration.getCloseLoopOriginator())
 110                 .thenReturn("DCAE-BBS-ep");
 111
 112         pipeline = new ReRegistrationPipeline(configuration, consumerTask,
 113                 publisherTask, aaiClientTask, new HashMap<>());
 114     }
 115
 116     @Test
 117     void handleEmptyResponseFromDmaap() throws SSLException {
 118
 119         when(configuration.getPipelinesTimeoutInSeconds()).thenReturn(10);
 120         when(consumerTask.execute(anyString()))
 121                 .thenReturn(Flux.error(new EmptyDmaapResponseException("Mock empty")));
 122
 123         StepVerifier.create(pipeline.executePipeline())
 124                 .expectSubscription()
 125                 .verifyComplete();
 126
 127         verifyZeroInteractions(aaiClientTask);
 128         verifyZeroInteractions(publisherTask);
 129     }
 130
 131     @Test
 132     void noResponseFromDmaap_PipelineTimesOut() throws SSLException {
 133
 134         // Prepare mocks
 135         when(configuration.getPipelinesTimeoutInSeconds()).thenReturn(1);
 136         when(consumerTask.execute(CONSUME_REREGISTRATION_TASK_NAME))
 137                 .thenReturn(Flux.never());
 138
 139         // Execute pipeline
 140         StepVerifier.create(pipeline.executePipeline())
 141                 .expectSubscription()
 142                 .verifyComplete();
 143
 144         verifyZeroInteractions(aaiClientTask);
 145         verifyZeroInteractions(publisherTask);
 146     }
 147
 148     @Test
 149     void noResponseFromAai_PipelineTimesOut() throws SSLException {
 150
 151         String pnfName = "olt1";
 152         String attachmentPoint = "olt2-2-2";
 153         String remoteId = "newRemoteId";
 154         String cvlan = "1005";
 155         String svlan = "100";
 156
 157         // Prepare stubbed replies
 158         ReRegistrationConsumerDmaapModel event = ImmutableReRegistrationConsumerDmaapModel.builder()
 159                 .correlationId(pnfName)
 160                 .attachmentPoint(attachmentPoint)
 161                 .remoteId(remoteId)
 162                 .cVlan(cvlan)
 163                 .sVlan(svlan)
 164                 .build();
 165
 166         // Prepare mocks
 167         when(configuration.getPipelinesTimeoutInSeconds()).thenReturn(1);
 168         when(consumerTask.execute(CONSUME_REREGISTRATION_TASK_NAME)).thenReturn(Flux.just(event));
 169         when(aaiClientTask.executePnfRetrieval(anyString(), anyString())).thenReturn(Mono.never());
 170
 171         // Execute pipeline
 172         StepVerifier.create(pipeline.executePipeline())
 173                 .expectSubscription()
 174                 .verifyComplete();
 175
 176         verifyZeroInteractions(publisherTask);
 177     }
 178
 179     @Test
 180     void noResponseWhilePublishing_PipelineTimesOut() throws SSLException {
 181
 182         String pnfName = "olt1";
 183         String attachmentPoint = "olt2-2-2";
 184         String remoteId = "newRemoteId";
 185         String cvlan = "1005";
 186         String svlan = "100";
 187         String hsiCfsServiceInstanceId = UUID.randomUUID().toString();
 188
 189         // Prepare stubbed replies
 190         ReRegistrationConsumerDmaapModel event = ImmutableReRegistrationConsumerDmaapModel.builder()
 191                 .correlationId(pnfName)
 192                 .attachmentPoint(attachmentPoint)
 193                 .remoteId(remoteId)
 194                 .cVlan(cvlan)
 195                 .sVlan(svlan)
 196                 .build();
 197
 198         PnfAaiObject pnfAaiObject = constructPnfObject(pnfName, "olt1-1-1", hsiCfsServiceInstanceId);
 199         ServiceInstanceAaiObject hsiCfsServiceInstance =
 200                 constructHsiCfsServiceInstanceObject(hsiCfsServiceInstanceId, pnfName, cvlan);
 201
 202         // Prepare Mocks
 203         String cfsUrl = String.format("/aai/v14/nodes/service-instances/service-instance/%s?depth=all",
 204                 hsiCfsServiceInstance.getServiceInstanceId());
 205
 206         when(configuration.getPipelinesTimeoutInSeconds()).thenReturn(1);
 207         when(consumerTask.execute(CONSUME_REREGISTRATION_TASK_NAME)).thenReturn(Flux.just(event));
 208
 209         when(aaiClientTask.executePnfRetrieval(anyString(), anyString()))
 210                 .thenReturn(Mono.just(pnfAaiObject));
 211
 212         when(aaiClientTask
 213                 .executeServiceInstanceRetrieval(RETRIEVE_HSI_CFS_SERVICE_INSTANCE_TASK_NAME, cfsUrl))
 214                 .thenReturn(Mono.just(hsiCfsServiceInstance));
 215
 216         when(publisherTask.execute(any(ControlLoopPublisherDmaapModel.class))).thenReturn(Mono.never());
 217
 218         // Execute the pipeline
 219         StepVerifier.create(pipeline.executePipeline())
 220                 .expectSubscription()
 221                 .verifyComplete();
 222
 223         verify(publisherTask).execute(any(ControlLoopPublisherDmaapModel.class));
 224     }
 225
 226     @Test
 227     void singleCorrectEvent_PnfHavingNoLogicalLink_handleGracefully() throws SSLException {
 228
 229         String pnfName = "olt1";
 230         String attachmentPoint = "olt2-2-2";
 231         String remoteId = "newRemoteId";
 232         String cvlan = "1005";
 233         String svlan = "100";
 234         String hsiCfsServiceInstanceId = UUID.randomUUID().toString();
 235
 236         // Prepare stubbed replies
 237         ReRegistrationConsumerDmaapModel event = ImmutableReRegistrationConsumerDmaapModel.builder()
 238                 .correlationId(pnfName)
 239                 .attachmentPoint(attachmentPoint)
 240                 .remoteId(remoteId)
 241                 .cVlan(cvlan)
 242                 .sVlan(svlan)
 243                 .build();
 244
 245         PnfAaiObject pnfAaiObject = constructPnfObjectWithoutLogicalLink(pnfName, hsiCfsServiceInstanceId);
 246         ServiceInstanceAaiObject hsiCfsServiceInstance =
 247                 constructHsiCfsServiceInstanceObject(hsiCfsServiceInstanceId, pnfName, cvlan);
 248
 249         // Prepare Mocks
 250         String cfsUrl = String.format("/aai/v14/nodes/service-instances/service-instance/%s?depth=all",
 251                 hsiCfsServiceInstance.getServiceInstanceId());
 252
 253         when(configuration.getPipelinesTimeoutInSeconds()).thenReturn(10);
 254         when(consumerTask.execute(CONSUME_REREGISTRATION_TASK_NAME)).thenReturn(Flux.just(event));
 255
 256         when(aaiClientTask.executePnfRetrieval(anyString(), anyString()))
 257                 .thenReturn(Mono.just(pnfAaiObject));
 258
 259         when(aaiClientTask
 260                 .executeServiceInstanceRetrieval(RETRIEVE_HSI_CFS_SERVICE_INSTANCE_TASK_NAME, cfsUrl))
 261                 .thenReturn(Mono.just(hsiCfsServiceInstance));
 262
 263         when(httpResponse.statusCode()).thenReturn(HttpStatus.OK.value());
 264         when(publisherTask.execute(any(ControlLoopPublisherDmaapModel.class))).thenReturn(Mono.just(httpResponse));
 265
 266         // Execute the pipeline
 267         StepVerifier.create(pipeline.executePipeline())
 268                 .expectSubscription()
 269                 .verifyComplete();
 270
 271         verify(aaiClientTask).executePnfRetrieval(anyString(), anyString());
 272         verifyNoMoreInteractions(aaiClientTask);
 273         verifyZeroInteractions(publisherTask);
 274     }
 275
 276     @Test
 277     void singleCorrectEvent_handleSuccessfully() throws SSLException {
 278
 279         String pnfName = "olt1";
 280         String attachmentPoint = "olt2-2-2";
 281         String remoteId = "newRemoteId";
 282         String cvlan = "1005";
 283         String svlan = "100";
 284         String hsiCfsServiceInstanceId = UUID.randomUUID().toString();
 285
 286         // Prepare stubbed replies
 287         ReRegistrationConsumerDmaapModel event = ImmutableReRegistrationConsumerDmaapModel.builder()
 288                 .correlationId(pnfName)
 289                 .attachmentPoint(attachmentPoint)
 290                 .remoteId(remoteId)
 291                 .cVlan(cvlan)
 292                 .sVlan(svlan)
 293                 .build();
 294
 295         PnfAaiObject pnfAaiObject = constructPnfObject(pnfName, "old-attachment-point", hsiCfsServiceInstanceId);
 296         ServiceInstanceAaiObject hsiCfsServiceInstance =
 297                 constructHsiCfsServiceInstanceObject(hsiCfsServiceInstanceId, pnfName, cvlan);
 298
 299         // Prepare Mocks
 300         String cfsUrl = String.format("/aai/v14/nodes/service-instances/service-instance/%s?depth=all",
 301                 hsiCfsServiceInstance.getServiceInstanceId());
 302
 303         when(configuration.getPipelinesTimeoutInSeconds()).thenReturn(10);
 304         when(consumerTask.execute(CONSUME_REREGISTRATION_TASK_NAME)).thenReturn(Flux.just(event));
 305
 306         when(aaiClientTask.executePnfRetrieval(anyString(), anyString()))
 307                 .thenReturn(Mono.just(pnfAaiObject));
 308
 309         when(aaiClientTask
 310                 .executeServiceInstanceRetrieval(RETRIEVE_HSI_CFS_SERVICE_INSTANCE_TASK_NAME, cfsUrl))
 311                 .thenReturn(Mono.just(hsiCfsServiceInstance));
 312
 313         when(httpResponse.statusCode()).thenReturn(HttpStatus.OK.value());
 314         when(publisherTask.execute(any(ControlLoopPublisherDmaapModel.class))).thenReturn(Mono.just(httpResponse));
 315
 316         // Execute the pipeline
 317         StepVerifier.create(pipeline.executePipeline())
 318                 .expectSubscription()
 319                 .assertNext(r -> assertEquals(HttpStatus.OK.value(), r.statusCode()))
 320                 .verifyComplete();
 321
 322         verify(publisherTask).execute(any(ControlLoopPublisherDmaapModel.class));
 323     }
 324
 325     @Test
 326     void twoCorrectEvents_handleSuccessfully() throws SSLException {
 327
 328         String pnfName1 = "olt1";
 329         String pnfName2 = "olt2";
 330         String attachmentPoint1 = "olt1-1-1";
 331         String attachmentPoint2 = "olt2-2-2";
 332         String remoteId1 = "newRemoteId1";
 333         String remoteId2 = "newRemoteId2";
 334         String cvlan1 = "1005";
 335         String cvlan2 = "1006";
 336         String svlan = "100";
 337         String hsiCfsServiceInstanceId1 = UUID.randomUUID().toString();
 338         String hsiCfsServiceInstanceId2 = UUID.randomUUID().toString();
 339
 340         // Prepare stubbed replies
 341         ReRegistrationConsumerDmaapModel firstEvent = ImmutableReRegistrationConsumerDmaapModel.builder()
 342                 .correlationId(pnfName1)
 343                 .attachmentPoint(attachmentPoint1)
 344                 .remoteId(remoteId1)
 345                 .cVlan(cvlan1)
 346                 .sVlan(svlan)
 347                 .build();
 348         ReRegistrationConsumerDmaapModel secondEvent = ImmutableReRegistrationConsumerDmaapModel.builder()
 349                 .correlationId(pnfName2)
 350                 .attachmentPoint(attachmentPoint2)
 351                 .remoteId(remoteId2)
 352                 .cVlan(cvlan2)
 353                 .sVlan(svlan)
 354                 .build();
 355
 356         PnfAaiObject pnfAaiObject1 = constructPnfObject(pnfName1, "olt1-1-0", hsiCfsServiceInstanceId1);
 357         PnfAaiObject pnfAaiObject2 = constructPnfObject(pnfName2, "olt2-2-0", hsiCfsServiceInstanceId2);
 358         ServiceInstanceAaiObject hsiCfsServiceInstance1 =
 359                 constructHsiCfsServiceInstanceObject(hsiCfsServiceInstanceId1, pnfName1, cvlan1);
 360         ServiceInstanceAaiObject hsiCfsServiceInstance2 =
 361                 constructHsiCfsServiceInstanceObject(hsiCfsServiceInstanceId2, pnfName2, cvlan2);
 362
 363         // Prepare Mocks
 364         String pnfUrl1 = String.format("/aai/v14/network/pnfs/pnf/%s?depth=all", pnfName1);
 365         String pnfUrl2 = String.format("/aai/v14/network/pnfs/pnf/%s?depth=all", pnfName2);
 366         String cfsUrl1 = String.format("/aai/v14/nodes/service-instances/service-instance/%s?depth=all",
 367                 hsiCfsServiceInstance1.getServiceInstanceId());
 368         String cfsUrl2 = String.format("/aai/v14/nodes/service-instances/service-instance/%s?depth=all",
 369                 hsiCfsServiceInstance2.getServiceInstanceId());
 370
 371         when(configuration.getPipelinesTimeoutInSeconds()).thenReturn(10);
 372         when(consumerTask.execute(CONSUME_REREGISTRATION_TASK_NAME))
 373                 .thenReturn(Flux.fromIterable(Arrays.asList(firstEvent, secondEvent)));
 374
 375         when(aaiClientTask.executePnfRetrieval(RETRIEVE_PNF_TASK_NAME, pnfUrl1)).thenReturn(Mono.just(pnfAaiObject1));
 376         when(aaiClientTask.executePnfRetrieval(RETRIEVE_PNF_TASK_NAME, pnfUrl2)).thenReturn(Mono.just(pnfAaiObject2));
 377
 378         when(aaiClientTask
 379                 .executeServiceInstanceRetrieval(RETRIEVE_HSI_CFS_SERVICE_INSTANCE_TASK_NAME, cfsUrl1))
 380                 .thenReturn(Mono.just(hsiCfsServiceInstance1));
 381         when(aaiClientTask
 382                 .executeServiceInstanceRetrieval(RETRIEVE_HSI_CFS_SERVICE_INSTANCE_TASK_NAME, cfsUrl2))
 383                 .thenReturn(Mono.just(hsiCfsServiceInstance2));
 384
 385         when(httpResponse.statusCode()).thenReturn(HttpStatus.OK.value());
 386         when(publisherTask.execute(any(ControlLoopPublisherDmaapModel.class))).thenReturn(Mono.just(httpResponse));
 387
 388         // Execute the pipeline
 389         StepVerifier.create(pipeline.executePipeline())
 390                 .expectSubscription()
 391                 .assertNext(r -> assertEquals(HttpStatus.OK.value(), r.statusCode()))
 392                 .assertNext(r -> assertEquals(HttpStatus.OK.value(), r.statusCode()))
 393                 .verifyComplete();
 394
 395         verify(publisherTask, times(2)).execute(any(ControlLoopPublisherDmaapModel.class));
 396     }
 397
 398     @Test
 399     void singleEvent_withPnfErrorReply_handleGracefully() throws SSLException {
 400
 401         String pnfName = "olt1";
 402         String attachmentPoint = "olt2-2-2";
 403         String remoteId = "newRemoteId";
 404         String cvlan = "1005";
 405         String svlan = "100";
 406
 407         // Prepare stubbed replies
 408         ReRegistrationConsumerDmaapModel event = ImmutableReRegistrationConsumerDmaapModel.builder()
 409                 .correlationId(pnfName)
 410                 .attachmentPoint(attachmentPoint)
 411                 .remoteId(remoteId)
 412                 .cVlan(cvlan)
 413                 .sVlan(svlan)
 414                 .build();
 415
 416         // Prepare Mocks
 417         when(configuration.getPipelinesTimeoutInSeconds()).thenReturn(10);
 418         when(consumerTask.execute(CONSUME_REREGISTRATION_TASK_NAME)).thenReturn(Flux.just(event));
 419         when(aaiClientTask.executePnfRetrieval(anyString(), anyString()))
 420                 .thenReturn(Mono.error(new AaiTaskException("Mock A&AI exception")));
 421
 422         // Execute the pipeline
 423         StepVerifier.create(pipeline.executePipeline())
 424                 .expectSubscription()
 425                 .verifyComplete();
 426
 427         verify(aaiClientTask).executePnfRetrieval(anyString(), anyString());
 428         verifyNoMoreInteractions(aaiClientTask);
 429         verifyZeroInteractions(publisherTask);
 430     }
 431
 432     @Test
 433     void twoEvents_FirstOk_SecondNotRelocation_handleCorrectOnly() throws SSLException {
 434
 435         String pnfName1 = "olt1";
 436         String pnfName2 = "olt2";
 437         String attachmentPoint1 = "olt1-1-1";
 438         String attachmentPoint2 = "olt2-2-2";
 439         String remoteId1 = "newRemoteId1";
 440         String remoteId2 = "newRemoteId2";
 441         String cvlan1 = "1005";
 442         String cvlan2 = "1006";
 443         String svlan = "100";
 444         String hsiCfsServiceInstanceId1 = UUID.randomUUID().toString();
 445         String hsiCfsServiceInstanceId2 = UUID.randomUUID().toString();
 446
 447         // Prepare stubbed replies
 448         ReRegistrationConsumerDmaapModel firstEvent = ImmutableReRegistrationConsumerDmaapModel.builder()
 449                 .correlationId(pnfName1)
 450                 .attachmentPoint(attachmentPoint1)
 451                 .remoteId(remoteId1)
 452                 .cVlan(cvlan1)
 453                 .sVlan(svlan)
 454                 .build();
 455         ReRegistrationConsumerDmaapModel secondEvent = ImmutableReRegistrationConsumerDmaapModel.builder()
 456                 .correlationId(pnfName2)
 457                 .attachmentPoint(attachmentPoint2)
 458                 .remoteId(remoteId2)
 459                 .cVlan(cvlan2)
 460                 .sVlan(svlan)
 461                 .build();
 462
 463         PnfAaiObject pnfAaiObject1 = constructPnfObject(pnfName1, "olt1-1-0", hsiCfsServiceInstanceId1);
 464         PnfAaiObject pnfAaiObject2 = constructPnfObject(pnfName2, attachmentPoint2, hsiCfsServiceInstanceId2);
 465         ServiceInstanceAaiObject hsiCfsServiceInstance1 =
 466                 constructHsiCfsServiceInstanceObject(hsiCfsServiceInstanceId1, pnfName1, cvlan1);
 467         ServiceInstanceAaiObject hsiCfsServiceInstance2 =
 468                 constructHsiCfsServiceInstanceObject(hsiCfsServiceInstanceId2, pnfName2, cvlan2);
 469
 470         // Prepare Mocks
 471         String pnfUrl1 = String.format("/aai/v14/network/pnfs/pnf/%s?depth=all", pnfName1);
 472         String pnfUrl2 = String.format("/aai/v14/network/pnfs/pnf/%s?depth=all", pnfName2);
 473         String cfsUrl1 = String.format("/aai/v14/nodes/service-instances/service-instance/%s?depth=all",
 474                 hsiCfsServiceInstance1.getServiceInstanceId());
 475         String cfsUrl2 = String.format("/aai/v14/nodes/service-instances/service-instance/%s?depth=all",
 476                 hsiCfsServiceInstance2.getServiceInstanceId());
 477
 478         when(configuration.getPipelinesTimeoutInSeconds()).thenReturn(10);
 479         when(consumerTask.execute(CONSUME_REREGISTRATION_TASK_NAME))
 480                 .thenReturn(Flux.fromIterable(Arrays.asList(firstEvent, secondEvent)));
 481
 482         when(aaiClientTask.executePnfRetrieval(RETRIEVE_PNF_TASK_NAME, pnfUrl1)).thenReturn(Mono.just(pnfAaiObject1));
 483         when(aaiClientTask.executePnfRetrieval(RETRIEVE_PNF_TASK_NAME, pnfUrl2)).thenReturn(Mono.just(pnfAaiObject2));
 484
 485         when(aaiClientTask
 486                 .executeServiceInstanceRetrieval(RETRIEVE_HSI_CFS_SERVICE_INSTANCE_TASK_NAME, cfsUrl1))
 487                 .thenReturn(Mono.just(hsiCfsServiceInstance1));
 488         when(aaiClientTask
 489                 .executeServiceInstanceRetrieval(RETRIEVE_HSI_CFS_SERVICE_INSTANCE_TASK_NAME, cfsUrl2))
 490                 .thenReturn(Mono.just(hsiCfsServiceInstance2));
 491
 492         when(httpResponse.statusCode()).thenReturn(HttpStatus.OK.value());
 493         when(publisherTask.execute(any(ControlLoopPublisherDmaapModel.class))).thenReturn(Mono.just(httpResponse));
 494
 495         // Execute the pipeline
 496         StepVerifier.create(pipeline.executePipeline())
 497                 .expectSubscription()
 498                 .assertNext(r -> assertEquals(HttpStatus.OK.value(), r.statusCode()))
 499                 .verifyComplete();
 500
 501         verify(publisherTask).execute(any(ControlLoopPublisherDmaapModel.class));
 502     }
 503
 504     @Test
 505     void twoEvents_firstOk_secondWithPnfErrorReply_handleCorrectOnly() throws SSLException {
 506
 507         String pnfName1 = "olt1";
 508         String pnfName2 = "olt2";
 509         String attachmentPoint1 = "olt1-1-1";
 510         String attachmentPoint2 = "olt2-2-2";
 511         String remoteId1 = "newRemoteId1";
 512         String remoteId2 = "newRemoteId2";
 513         String cvlan1 = "1005";
 514         String cvlan2 = "1006";
 515         String svlan = "100";
 516         String hsiCfsServiceInstanceId = UUID.randomUUID().toString();
 517
 518         // Prepare stubbed replies
 519         ReRegistrationConsumerDmaapModel firstEvent = ImmutableReRegistrationConsumerDmaapModel.builder()
 520                 .correlationId(pnfName1)
 521                 .attachmentPoint(attachmentPoint1)
 522                 .remoteId(remoteId1)
 523                 .cVlan(cvlan1)
 524                 .sVlan(svlan)
 525                 .build();
 526         ReRegistrationConsumerDmaapModel secondEvent = ImmutableReRegistrationConsumerDmaapModel.builder()
 527                 .correlationId(pnfName2)
 528                 .attachmentPoint(attachmentPoint2)
 529                 .remoteId(remoteId2)
 530                 .cVlan(cvlan2)
 531                 .sVlan(svlan)
 532                 .build();
 533
 534         PnfAaiObject pnfAaiObject = constructPnfObject(pnfName1, "old-attachment-point", hsiCfsServiceInstanceId);
 535         ServiceInstanceAaiObject hsiCfsServiceInstance =
 536                 constructHsiCfsServiceInstanceObject(hsiCfsServiceInstanceId, pnfName1, cvlan1);
 537
 538         // Prepare Mocks
 539         String cfsUrl = String.format("/aai/v14/nodes/service-instances/service-instance/%s?depth=all",
 540                 hsiCfsServiceInstance.getServiceInstanceId());
 541
 542         when(configuration.getPipelinesTimeoutInSeconds()).thenReturn(10);
 543         when(consumerTask.execute(CONSUME_REREGISTRATION_TASK_NAME))
 544                 .thenReturn(Flux.fromIterable(Arrays.asList(firstEvent, secondEvent)));
 545         when(aaiClientTask.executePnfRetrieval(anyString(), anyString()))
 546                 .thenReturn(Mono.just(pnfAaiObject))
 547                 .thenReturn(Mono.error(new AaiTaskException("Mock A&AI exception")));
 548         when(aaiClientTask
 549                 .executeServiceInstanceRetrieval(RETRIEVE_HSI_CFS_SERVICE_INSTANCE_TASK_NAME, cfsUrl))
 550                 .thenReturn(Mono.just(hsiCfsServiceInstance));
 551
 552         when(httpResponse.statusCode()).thenReturn(HttpStatus.OK.value());
 553         when(publisherTask.execute(any(ControlLoopPublisherDmaapModel.class))).thenReturn(Mono.just(httpResponse));
 554
 555         // Execute the pipeline
 556         StepVerifier.create(pipeline.executePipeline())
 557                 .expectSubscription()
 558                 .assertNext(r -> assertEquals(HttpStatus.OK.value(), r.statusCode()))
 559                 .verifyComplete();
 560
 561         verify(aaiClientTask, times(2)).executePnfRetrieval(anyString(), anyString());
 562         verify(aaiClientTask).executeServiceInstanceRetrieval(anyString(), anyString());
 563         verify(publisherTask).execute(any(ControlLoopPublisherDmaapModel.class));
 564     }
 565
 566     @Test
 567     void twoEvents_firstWithPnfErrorReply_secondOk_handleCorrectOnly() throws SSLException {
 568
 569         String pnfName1 = "olt1";
 570         String pnfName2 = "olt2";
 571         String attachmentPoint1 = "olt1-1-1";
 572         String attachmentPoint2 = "olt2-2-2";
 573         String remoteId1 = "newRemoteId1";
 574         String remoteId2 = "newRemoteId2";
 575         String cvlan1 = "1005";
 576         String cvlan2 = "1006";
 577         String svlan = "100";
 578         String hsiCfsServiceInstanceId = UUID.randomUUID().toString();
 579
 580         // Prepare stubbed replies
 581         ReRegistrationConsumerDmaapModel firstEvent = ImmutableReRegistrationConsumerDmaapModel.builder()
 582                 .correlationId(pnfName1)
 583                 .attachmentPoint(attachmentPoint1)
 584                 .remoteId(remoteId1)
 585                 .cVlan(cvlan1)
 586                 .sVlan(svlan)
 587                 .build();
 588         ReRegistrationConsumerDmaapModel secondEvent = ImmutableReRegistrationConsumerDmaapModel.builder()
 589                 .correlationId(pnfName2)
 590                 .attachmentPoint(attachmentPoint2)
 591                 .remoteId(remoteId2)
 592                 .cVlan(cvlan2)
 593                 .sVlan(svlan)
 594                 .build();
 595
 596         PnfAaiObject pnfAaiObject = constructPnfObject(pnfName2, "old-attachment-point", hsiCfsServiceInstanceId);
 597         ServiceInstanceAaiObject hsiCfsServiceInstance =
 598                 constructHsiCfsServiceInstanceObject(hsiCfsServiceInstanceId, pnfName2, cvlan2);
 599
 600         // Prepare Mocks
 601         String cfsUrl = String.format("/aai/v14/nodes/service-instances/service-instance/%s?depth=all",
 602                 hsiCfsServiceInstance.getServiceInstanceId());
 603
 604         when(configuration.getPipelinesTimeoutInSeconds()).thenReturn(10);
 605         when(consumerTask.execute(CONSUME_REREGISTRATION_TASK_NAME))
 606                 .thenReturn(Flux.fromIterable(Arrays.asList(firstEvent, secondEvent)));
 607         when(aaiClientTask.executePnfRetrieval(anyString(), anyString()))
 608                 .thenReturn(Mono.error(new AaiTaskException("Mock A&AI exception")))
 609                 .thenReturn(Mono.just(pnfAaiObject));
 610         when(aaiClientTask
 611                 .executeServiceInstanceRetrieval(RETRIEVE_HSI_CFS_SERVICE_INSTANCE_TASK_NAME, cfsUrl))
 612                 .thenReturn(Mono.just(hsiCfsServiceInstance));
 613
 614         when(httpResponse.statusCode()).thenReturn(HttpStatus.OK.value());
 615         when(publisherTask.execute(any(ControlLoopPublisherDmaapModel.class))).thenReturn(Mono.just(httpResponse));
 616
 617         // Execute the pipeline
 618         StepVerifier.create(pipeline.executePipeline())
 619                 .expectSubscription()
 620                 .assertNext(r -> assertEquals(HttpStatus.OK.value(), r.statusCode()))
 621                 .verifyComplete();
 622
 623         verify(aaiClientTask, times(2))
 624                 .executePnfRetrieval(anyString(), anyString());
 625         verify(aaiClientTask).executeServiceInstanceRetrieval(anyString(), anyString());
 626         verify(publisherTask).execute(any(ControlLoopPublisherDmaapModel.class));
 627     }
 628
 629     private PnfAaiObject constructPnfObject(String pnfName, String attachmentPoint,
 630                                             String hsiCfsServiceInstanceId) {
 631
 632         // Build Relationship Data
 633         RelationshipListAaiObject.RelationshipEntryAaiObject firstRelationshipEntry =
 634                 ImmutableRelationshipEntryAaiObject.builder()
 635                         .relatedTo("service-instance")
 636                         .relatedLink("/aai/v14/business/customers/customer/Demonstration/service-subscriptions"
 637                                 + "/service-subscription/BBS-CFS/service-instances"
 638                                 + "/service-instance/" + hsiCfsServiceInstanceId)
 639                         .relationshipLabel("org.onap.relationships.inventory.ComposedOf")
 640                         .relationshipData(Arrays.asList(
 641                                 ImmutableRelationshipDataEntryAaiObject.builder()
 642                                         .relationshipKey("customer.global-customer-id")
 643                                         .relationshipValue("Demonstration").build(),
 644                                 ImmutableRelationshipDataEntryAaiObject.builder()
 645                                         .relationshipKey("service-subscription.service-type")
 646                                         .relationshipValue("BBS-CFS").build(),
 647                                 ImmutableRelationshipDataEntryAaiObject.builder()
 648                                         .relationshipKey("service-instance.service-instance-id")
 649                                         .relationshipValue(hsiCfsServiceInstanceId).build())
 650                         )
 651                         .relatedToProperties(Collections.singletonList(
 652                                 ImmutablePropertyAaiObject.builder()
 653                                         .propertyKey("service-instance.service-instance-name")
 654                                         .propertyValue("bbs-instance").build())
 655                         )
 656                         .build();
 657
 658         RelationshipListAaiObject.RelationshipEntryAaiObject secondRelationshipEntry =
 659                 ImmutableRelationshipEntryAaiObject.builder()
 660                         .relatedTo("logical-link")
 661                         .relatedLink("/network/logical-links/logical-link/" + attachmentPoint)
 662                         .relationshipData(Collections.singletonList(ImmutableRelationshipDataEntryAaiObject.builder()
 663                                 .relationshipKey("logical-link.link-name")
 664                                 .relationshipValue(attachmentPoint).build()))
 665                         .build();
 666
 667         RelationshipListAaiObject relationshipListAaiObject = ImmutableRelationshipListAaiObject.builder()
 668                 .relationshipEntries(Arrays.asList(firstRelationshipEntry, secondRelationshipEntry))
 669                 .build();
 670
 671         // Finally construct PNF object data
 672         return ImmutablePnfAaiObject.builder()
 673                 .pnfName(pnfName)
 674                 .isInMaintenance(true)
 675                 .relationshipListAaiObject(relationshipListAaiObject)
 676                 .build();
 677     }
 678
 679     private PnfAaiObject constructPnfObjectWithoutLogicalLink(String pnfName, String hsiCfsServiceInstanceId) {
 680
 681         // Build Relationship Data
 682         RelationshipListAaiObject.RelationshipEntryAaiObject relationshipEntry =
 683                 ImmutableRelationshipEntryAaiObject.builder()
 684                         .relatedTo("service-instance")
 685                         .relatedLink("/aai/v14/business/customers/customer/Demonstration/service-subscriptions"
 686                                 + "/service-subscription/BBS-CFS/service-instances"
 687                                 + "/service-instance/" + hsiCfsServiceInstanceId)
 688                         .relationshipLabel("org.onap.relationships.inventory.ComposedOf")
 689                         .relationshipData(Arrays.asList(
 690                                 ImmutableRelationshipDataEntryAaiObject.builder()
 691                                         .relationshipKey("customer.global-customer-id")
 692                                         .relationshipValue("Demonstration").build(),
 693                                 ImmutableRelationshipDataEntryAaiObject.builder()
 694                                         .relationshipKey("service-subscription.service-type")
 695                                         .relationshipValue("BBS-CFS").build(),
 696                                 ImmutableRelationshipDataEntryAaiObject.builder()
 697                                         .relationshipKey("service-instance.service-instance-id")
 698                                         .relationshipValue(hsiCfsServiceInstanceId).build())
 699                         )
 700                         .relatedToProperties(Collections.singletonList(
 701                                 ImmutablePropertyAaiObject.builder()
 702                                         .propertyKey("service-instance.service-instance-name")
 703                                         .propertyValue("bbs-instance").build())
 704                         )
 705                         .build();
 706
 707         RelationshipListAaiObject relationshipListAaiObject = ImmutableRelationshipListAaiObject.builder()
 708                 .relationshipEntries(Collections.singletonList(relationshipEntry))
 709                 .build();
 710
 711         // Finally construct PNF object data
 712         return ImmutablePnfAaiObject.builder()
 713                 .pnfName(pnfName)
 714                 .isInMaintenance(true)
 715                 .relationshipListAaiObject(relationshipListAaiObject)
 716                 .build();
 717     }
 718
 719     private ServiceInstanceAaiObject constructHsiCfsServiceInstanceObject(String hsiCfsServiceInstanceId,
 720                                                                              String pnfName,
 721                                                                              String cvlan) {
 722         String orchestrationStatus = "active";
 723
 724         RelationshipListAaiObject.RelationshipEntryAaiObject relationshipEntry =
 725                 ImmutableRelationshipEntryAaiObject.builder()
 726                         .relatedTo("pnf")
 727                         .relatedLink("/pnfs/pnf/" + pnfName)
 728                         .relationshipData(Collections.singletonList(ImmutableRelationshipDataEntryAaiObject.builder()
 729                                 .relationshipKey("pnf.pnf-name")
 730                                 .relationshipValue(pnfName).build()))
 731                         .build();
 732
 733         RelationshipListAaiObject relationshipListAaiObject = ImmutableRelationshipListAaiObject.builder()
 734                 .relationshipEntries(Collections.singletonList(relationshipEntry))
 735                 .build();
 736
 737         MetadataListAaiObject.MetadataEntryAaiObject metadataEntry =
 738                 ImmutableMetadataEntryAaiObject.builder()
 739                         .metaname("cvlan")
 740                         .metavalue(cvlan)
 741                         .build();
 742
 743         MetadataListAaiObject metadataListAaiObject = ImmutableMetadataListAaiObject.builder()
 744                 .metadataEntries(Collections.singletonList(metadataEntry))
 745                 .build();
 746
 747         // Finally construct Service Instance object data
 748         return ImmutableServiceInstanceAaiObject.builder()
 749                 .serviceInstanceId(hsiCfsServiceInstanceId)
 750                 .orchestrationStatus(orchestrationStatus)
 751                 .relationshipListAaiObject(relationshipListAaiObject)
 752                 .metadataListAaiObject(metadataListAaiObject)
 753                 .build();
 754     }
 755 }