enrichment-coordinator-service/src/test/java/org/oransc/enrichment/ApplicationTest.java

   1 /*-
   2  * ========================LICENSE_START=================================
   3  * O-RAN-SC
   4  * %%
   5  * Copyright (C) 2020 Nordix Foundation
   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.oransc.enrichment;
  22
  23 import static org.assertj.core.api.Assertions.assertThat;
  24 import static org.awaitility.Awaitility.await;
  25 import static org.junit.jupiter.api.Assertions.assertTrue;
  26
  27 import com.fasterxml.jackson.core.JsonProcessingException;
  28 import com.fasterxml.jackson.databind.JsonMappingException;
  29 import com.google.gson.Gson;
  30 import com.google.gson.GsonBuilder;
  31 import com.google.gson.JsonParser;
  32
  33 import java.io.FileNotFoundException;
  34 import java.io.FileOutputStream;
  35 import java.io.PrintStream;
  36 import java.lang.invoke.MethodHandles;
  37 import java.util.Arrays;
  38
  39 import org.json.JSONObject;
  40 import org.junit.jupiter.api.AfterEach;
  41 import org.junit.jupiter.api.BeforeEach;
  42 import org.junit.jupiter.api.Test;
  43 import org.junit.jupiter.api.extension.ExtendWith;
  44 import org.oransc.enrichment.clients.AsyncRestClient;
  45 import org.oransc.enrichment.clients.AsyncRestClientFactory;
  46 import org.oransc.enrichment.configuration.ApplicationConfig;
  47 import org.oransc.enrichment.configuration.ImmutableHttpProxyConfig;
  48 import org.oransc.enrichment.configuration.ImmutableWebClientConfig;
  49 import org.oransc.enrichment.configuration.WebClientConfig;
  50 import org.oransc.enrichment.configuration.WebClientConfig.HttpProxyConfig;
  51 import org.oransc.enrichment.controller.ConsumerSimulatorController;
  52 import org.oransc.enrichment.controller.ProducerSimulatorController;
  53 import org.oransc.enrichment.controllers.a1e.A1eConsts;
  54 import org.oransc.enrichment.controllers.a1e.A1eEiJobInfo;
  55 import org.oransc.enrichment.controllers.a1e.A1eEiJobStatus;
  56 import org.oransc.enrichment.controllers.a1e.A1eEiTypeInfo;
  57 import org.oransc.enrichment.controllers.r1consumer.ConsumerConsts;
  58 import org.oransc.enrichment.controllers.r1consumer.ConsumerInfoTypeInfo;
  59 import org.oransc.enrichment.controllers.r1consumer.ConsumerJobInfo;
  60 import org.oransc.enrichment.controllers.r1consumer.ConsumerJobStatus;
  61 import org.oransc.enrichment.controllers.r1consumer.ConsumerTypeRegistrationInfo;
  62 import org.oransc.enrichment.controllers.r1consumer.ConsumerTypeSubscriptionInfo;
  63 import org.oransc.enrichment.controllers.r1producer.ProducerCallbacks;
  64 import org.oransc.enrichment.controllers.r1producer.ProducerConsts;
  65 import org.oransc.enrichment.controllers.r1producer.ProducerInfoTypeInfo;
  66 import org.oransc.enrichment.controllers.r1producer.ProducerJobInfo;
  67 import org.oransc.enrichment.controllers.r1producer.ProducerRegistrationInfo;
  68 import org.oransc.enrichment.controllers.r1producer.ProducerStatusInfo;
  69 import org.oransc.enrichment.exceptions.ServiceException;
  70 import org.oransc.enrichment.repository.InfoJob;
  71 import org.oransc.enrichment.repository.InfoJobs;
  72 import org.oransc.enrichment.repository.InfoProducer;
  73 import org.oransc.enrichment.repository.InfoProducers;
  74 import org.oransc.enrichment.repository.InfoType;
  75 import org.oransc.enrichment.repository.InfoTypeSubscriptions;
  76 import org.oransc.enrichment.repository.InfoTypes;
  77 import org.oransc.enrichment.tasks.ProducerSupervision;
  78 import org.slf4j.Logger;
  79 import org.slf4j.LoggerFactory;
  80 import org.springframework.beans.factory.annotation.Autowired;
  81 import org.springframework.boot.test.context.SpringBootTest;
  82 import org.springframework.boot.test.context.SpringBootTest.WebEnvironment;
  83 import org.springframework.boot.test.context.TestConfiguration;
  84 import org.springframework.boot.web.embedded.tomcat.TomcatServletWebServerFactory;
  85 import org.springframework.boot.web.server.LocalServerPort;
  86 import org.springframework.boot.web.servlet.server.ServletWebServerFactory;
  87 import org.springframework.context.ApplicationContext;
  88 import org.springframework.context.annotation.Bean;
  89 import org.springframework.http.HttpStatus;
  90 import org.springframework.http.MediaType;
  91 import org.springframework.http.ResponseEntity;
  92 import org.springframework.test.context.TestPropertySource;
  93 import org.springframework.test.context.junit.jupiter.SpringExtension;
  94 import org.springframework.web.reactive.function.client.WebClientResponseException;
  95
  96 import reactor.core.publisher.Mono;
  97 import reactor.test.StepVerifier;
  98
  99 @ExtendWith(SpringExtension.class)
 100 @SpringBootTest(webEnvironment = WebEnvironment.RANDOM_PORT)
 101 @TestPropertySource(
 102     properties = { //
 103         "server.ssl.key-store=./config/keystore.jks", //
 104         "app.webclient.trust-store=./config/truststore.jks", //
 105         "app.vardata-directory=./target"})
 106 class ApplicationTest {
 107     private final Logger logger = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
 108
 109     private final String TYPE_ID = "typeId";
 110     private final String PRODUCER_ID = "producerId";
 111     private final String EI_JOB_PROPERTY = "\"property1\"";
 112     private final String EI_JOB_ID = "jobId";
 113
 114     @Autowired
 115     ApplicationContext context;
 116
 117     @Autowired
 118     InfoJobs infoJobs;
 119
 120     @Autowired
 121     InfoTypes infoTypes;
 122
 123     @Autowired
 124     InfoProducers infoProducers;
 125
 126     @Autowired
 127     ApplicationConfig applicationConfig;
 128
 129     @Autowired
 130     ProducerSimulatorController producerSimulator;
 131
 132     @Autowired
 133     ConsumerSimulatorController consumerSimulator;
 134
 135     @Autowired
 136     ProducerSupervision producerSupervision;
 137
 138     @Autowired
 139     ProducerCallbacks producerCallbacks;
 140
 141     @Autowired
 142     InfoTypeSubscriptions infoTypeSubscriptions;
 143
 144     private static Gson gson = new GsonBuilder().create();
 145
 146     /**
 147      * Overrides the BeanFactory.
 148      */
 149     @TestConfiguration
 150     static class TestBeanFactory {
 151         @Bean
 152         public ServletWebServerFactory servletContainer() {
 153             return new TomcatServletWebServerFactory();
 154         }
 155     }
 156
 157     @LocalServerPort
 158     private int port;
 159
 160     @BeforeEach
 161     void reset() {
 162         this.infoJobs.clear();
 163         this.infoTypes.clear();
 164         this.infoProducers.clear();
 165         this.infoTypeSubscriptions.clear();
 166         this.producerSimulator.getTestResults().reset();
 167         this.consumerSimulator.getTestResults().reset();
 168     }
 169
 170     @AfterEach
 171     void check() {
 172         assertThat(this.producerSimulator.getTestResults().errorFound).isFalse();
 173     }
 174
 175     @Test
 176     void generateApiDoc() throws FileNotFoundException {
 177         String url = "/v3/api-docs";
 178         ResponseEntity<String> resp = restClient().getForEntity(url).block();
 179         assertThat(resp.getStatusCode()).isEqualTo(HttpStatus.OK);
 180
 181         JSONObject jsonObj = new JSONObject(resp.getBody());
 182         assertThat(jsonObj.remove("servers")).isNotNull();
 183
 184         String indented = jsonObj.toString(4);
 185         try (PrintStream out = new PrintStream(new FileOutputStream("api/ecs-api.json"))) {
 186             out.print(indented);
 187         }
 188     }
 189
 190     @Test
 191     void a1eGetEiTypes() throws Exception {
 192         putInfoProducerWithOneType(PRODUCER_ID, "test");
 193         String url = A1eConsts.API_ROOT + "/eitypes";
 194         String rsp = restClient().get(url).block();
 195         assertThat(rsp).isEqualTo("[\"test\"]");
 196     }
 197
 198     @Test
 199     void consumerGetInfoTypes() throws Exception {
 200         putInfoProducerWithOneType(PRODUCER_ID, "test");
 201         String url = ConsumerConsts.API_ROOT + "/info-types";
 202         String rsp = restClient().get(url).block();
 203         assertThat(rsp).isEqualTo("[\"test\"]");
 204     }
 205
 206     @Test
 207     void a1eGetEiTypesEmpty() throws Exception {
 208         String url = A1eConsts.API_ROOT + "/eitypes";
 209         String rsp = restClient().get(url).block();
 210         assertThat(rsp).isEqualTo("[]");
 211     }
 212
 213     @Test
 214     void consumerGetEiTypesEmpty() throws Exception {
 215         String url = ConsumerConsts.API_ROOT + "/info-types";
 216         String rsp = restClient().get(url).block();
 217         assertThat(rsp).isEqualTo("[]");
 218     }
 219
 220     @Test
 221     void a1eGetEiType() throws Exception {
 222         putInfoProducerWithOneType(PRODUCER_ID, "test");
 223         String url = A1eConsts.API_ROOT + "/eitypes/test";
 224         String rsp = restClient().get(url).block();
 225         A1eEiTypeInfo info = gson.fromJson(rsp, A1eEiTypeInfo.class);
 226         assertThat(info).isNotNull();
 227     }
 228
 229     @Test
 230     void consumerGetEiType() throws Exception {
 231         putInfoProducerWithOneType(PRODUCER_ID, "test");
 232         String url = ConsumerConsts.API_ROOT + "/info-types/test";
 233         String rsp = restClient().get(url).block();
 234         ConsumerInfoTypeInfo info = gson.fromJson(rsp, ConsumerInfoTypeInfo.class);
 235         assertThat(info).isNotNull();
 236         assertThat(info.jobDataSchema).isNotNull();
 237         assertThat(info.state).isEqualTo(ConsumerInfoTypeInfo.ConsumerTypeStatusValues.ENABLED);
 238         assertThat(info.noOfProducers).isEqualTo(1);
 239     }
 240
 241     @Test
 242     void a1eGetEiTypeNotFound() throws Exception {
 243         String url = A1eConsts.API_ROOT + "/eitypes/junk";
 244         testErrorCode(restClient().get(url), HttpStatus.NOT_FOUND, "Information type not found: junk");
 245     }
 246
 247     @Test
 248     void consumerGetEiTypeNotFound() throws Exception {
 249         String url = ConsumerConsts.API_ROOT + "/info-types/junk";
 250         testErrorCode(restClient().get(url), HttpStatus.NOT_FOUND, "Information type not found: junk");
 251     }
 252
 253     @Test
 254     void a1eGetEiJobsIds() throws Exception {
 255         putInfoProducerWithOneType(PRODUCER_ID, TYPE_ID);
 256         putEiJob(TYPE_ID, "jobId");
 257         final String JOB_ID_JSON = "[\"jobId\"]";
 258         String url = A1eConsts.API_ROOT + "/eijobs?infoTypeId=typeId";
 259         String rsp = restClient().get(url).block();
 260         assertThat(rsp).isEqualTo(JOB_ID_JSON);
 261
 262         url = A1eConsts.API_ROOT + "/eijobs?owner=owner";
 263         rsp = restClient().get(url).block();
 264         assertThat(rsp).isEqualTo(JOB_ID_JSON);
 265
 266         url = A1eConsts.API_ROOT + "/eijobs?owner=JUNK";
 267         rsp = restClient().get(url).block();
 268         assertThat(rsp).isEqualTo("[]");
 269
 270         url = A1eConsts.API_ROOT + "/eijobs";
 271         rsp = restClient().get(url).block();
 272         assertThat(rsp).isEqualTo(JOB_ID_JSON);
 273
 274         url = A1eConsts.API_ROOT + "/eijobs?eiTypeId=typeId&&owner=owner";
 275         rsp = restClient().get(url).block();
 276         assertThat(rsp).isEqualTo(JOB_ID_JSON);
 277
 278         url = A1eConsts.API_ROOT + "/eijobs?eiTypeId=JUNK";
 279         rsp = restClient().get(url).block();
 280         assertThat(rsp).isEqualTo("[]");
 281     }
 282
 283     @Test
 284     void consumerGetInformationJobsIds() throws Exception {
 285         putInfoProducerWithOneType(PRODUCER_ID, TYPE_ID);
 286         putEiJob(TYPE_ID, "jobId");
 287         final String JOB_ID_JSON = "[\"jobId\"]";
 288         String url = ConsumerConsts.API_ROOT + "/info-jobs?infoTypeId=typeId";
 289         String rsp = restClient().get(url).block();
 290         assertThat(rsp).isEqualTo(JOB_ID_JSON);
 291
 292         url = ConsumerConsts.API_ROOT + "/info-jobs?owner=owner";
 293         rsp = restClient().get(url).block();
 294         assertThat(rsp).isEqualTo(JOB_ID_JSON);
 295
 296         url = ConsumerConsts.API_ROOT + "/info-jobs?owner=JUNK";
 297         rsp = restClient().get(url).block();
 298         assertThat(rsp).isEqualTo("[]");
 299
 300         url = ConsumerConsts.API_ROOT + "/info-jobs";
 301         rsp = restClient().get(url).block();
 302         assertThat(rsp).isEqualTo(JOB_ID_JSON);
 303
 304         url = ConsumerConsts.API_ROOT + "/info-jobs?infoTypeId=typeId&&owner=owner";
 305         rsp = restClient().get(url).block();
 306         assertThat(rsp).isEqualTo(JOB_ID_JSON);
 307
 308         url = ConsumerConsts.API_ROOT + "/info-jobs?infoTypeId=JUNK";
 309         rsp = restClient().get(url).block();
 310         assertThat(rsp).isEqualTo("[]");
 311     }
 312
 313     @Test
 314     void a1eGetEiJob() throws Exception {
 315         putInfoProducerWithOneType(PRODUCER_ID, TYPE_ID);
 316         putEiJob(TYPE_ID, "jobId");
 317         String url = A1eConsts.API_ROOT + "/eijobs/jobId";
 318         String rsp = restClient().get(url).block();
 319         A1eEiJobInfo info = gson.fromJson(rsp, A1eEiJobInfo.class);
 320         assertThat(info.owner).isEqualTo("owner");
 321         assertThat(info.eiTypeId).isEqualTo(TYPE_ID);
 322     }
 323
 324     @Test
 325     void consumerGetEiJob() throws Exception {
 326         putInfoProducerWithOneType(PRODUCER_ID, TYPE_ID);
 327         putEiJob(TYPE_ID, "jobId");
 328         String url = ConsumerConsts.API_ROOT + "/info-jobs/jobId";
 329         String rsp = restClient().get(url).block();
 330         ConsumerJobInfo info = gson.fromJson(rsp, ConsumerJobInfo.class);
 331         assertThat(info.owner).isEqualTo("owner");
 332         assertThat(info.infoTypeId).isEqualTo(TYPE_ID);
 333     }
 334
 335     @Test
 336     void a1eGetEiJobNotFound() throws Exception {
 337         putInfoProducerWithOneType(PRODUCER_ID, TYPE_ID);
 338         String url = A1eConsts.API_ROOT + "/eijobs/junk";
 339         testErrorCode(restClient().get(url), HttpStatus.NOT_FOUND, "Could not find Information job: junk");
 340     }
 341
 342     @Test
 343     void consumerGetInfoJobNotFound() throws Exception {
 344         putInfoProducerWithOneType(PRODUCER_ID, TYPE_ID);
 345         String url = ConsumerConsts.API_ROOT + "/info-jobs/junk";
 346         testErrorCode(restClient().get(url), HttpStatus.NOT_FOUND, "Could not find Information job: junk");
 347     }
 348
 349     @Test
 350     void a1eGetEiJobStatus() throws Exception {
 351         putInfoProducerWithOneType(PRODUCER_ID, TYPE_ID);
 352         putEiJob(TYPE_ID, "jobId");
 353
 354         verifyJobStatus("jobId", "ENABLED");
 355     }
 356
 357     @Test
 358     void consumerGetEiJobStatus() throws Exception {
 359         putInfoProducerWithOneType(PRODUCER_ID, TYPE_ID);
 360         putEiJob(TYPE_ID, "jobId");
 361
 362         String url = ConsumerConsts.API_ROOT + "/info-jobs/jobId/status";
 363         String rsp = restClient().get(url).block();
 364         assertThat(rsp) //
 365             .contains("ENABLED") //
 366             .contains(PRODUCER_ID);
 367
 368         ConsumerJobStatus status = gson.fromJson(rsp, ConsumerJobStatus.class);
 369         assertThat(status.producers).contains(PRODUCER_ID);
 370     }
 371
 372     @Test
 373     void a1eDeleteEiJob() throws Exception {
 374         putInfoProducerWithOneType(PRODUCER_ID, TYPE_ID);
 375         putEiJob(TYPE_ID, "jobId");
 376         assertThat(this.infoJobs.size()).isEqualTo(1);
 377         String url = A1eConsts.API_ROOT + "/eijobs/jobId";
 378         restClient().delete(url).block();
 379         assertThat(this.infoJobs.size()).isZero();
 380
 381         ProducerSimulatorController.TestResults simulatorResults = this.producerSimulator.getTestResults();
 382         await().untilAsserted(() -> assertThat(simulatorResults.jobsStopped.size()).isEqualTo(1));
 383         assertThat(simulatorResults.jobsStopped.get(0)).isEqualTo("jobId");
 384     }
 385
 386     @Test
 387     void consumerDeleteEiJob() throws Exception {
 388         putInfoProducerWithOneType(PRODUCER_ID, TYPE_ID);
 389         putEiJob(TYPE_ID, "jobId");
 390         assertThat(this.infoJobs.size()).isEqualTo(1);
 391         String url = ConsumerConsts.API_ROOT + "/info-jobs/jobId";
 392         restClient().delete(url).block();
 393         assertThat(this.infoJobs.size()).isZero();
 394
 395         ProducerSimulatorController.TestResults simulatorResults = this.producerSimulator.getTestResults();
 396         await().untilAsserted(() -> assertThat(simulatorResults.jobsStopped.size()).isEqualTo(1));
 397         assertThat(simulatorResults.jobsStopped.get(0)).isEqualTo("jobId");
 398     }
 399
 400     @Test
 401     void a1eDeleteEiJobNotFound() throws Exception {
 402         putInfoProducerWithOneType(PRODUCER_ID, TYPE_ID);
 403         String url = A1eConsts.API_ROOT + "/eijobs/junk";
 404         testErrorCode(restClient().get(url), HttpStatus.NOT_FOUND, "Could not find Information job: junk");
 405     }
 406
 407     @Test
 408     void consumerDeleteEiJobNotFound() throws Exception {
 409         putInfoProducerWithOneType(PRODUCER_ID, TYPE_ID);
 410         String url = ConsumerConsts.API_ROOT + "/info-jobs/junk";
 411         testErrorCode(restClient().get(url), HttpStatus.NOT_FOUND, "Could not find Information job: junk");
 412     }
 413
 414     @Test
 415     void a1ePutEiJob() throws Exception {
 416         // Test that one producer accepting a job is enough
 417         putInfoProducerWithOneType(PRODUCER_ID, TYPE_ID);
 418         putEiProducerWithOneTypeRejecting("simulateProducerError", TYPE_ID);
 419
 420         String url = A1eConsts.API_ROOT + "/eijobs/jobId";
 421         String body = gson.toJson(infoJobInfo());
 422         ResponseEntity<String> resp = restClient().putForEntity(url, body).block();
 423         assertThat(this.infoJobs.size()).isEqualTo(1);
 424         assertThat(resp.getStatusCode()).isEqualTo(HttpStatus.CREATED);
 425
 426         ProducerSimulatorController.TestResults simulatorResults = this.producerSimulator.getTestResults();
 427         await().untilAsserted(() -> assertThat(simulatorResults.jobsStarted.size()).isEqualTo(1));
 428         ProducerJobInfo request = simulatorResults.jobsStarted.get(0);
 429         assertThat(request.id).isEqualTo("jobId");
 430
 431         // One retry --> two calls
 432         await().untilAsserted(() -> assertThat(simulatorResults.noOfRejectedCreate).isEqualTo(2));
 433         assertThat(simulatorResults.noOfRejectedCreate).isEqualTo(2);
 434
 435         resp = restClient().putForEntity(url, body).block();
 436         assertThat(resp.getStatusCode()).isEqualTo(HttpStatus.OK);
 437         InfoJob job = this.infoJobs.getJob("jobId");
 438         assertThat(job.getOwner()).isEqualTo("owner");
 439
 440         verifyJobStatus(EI_JOB_ID, "ENABLED");
 441     }
 442
 443     @Test
 444     void consumerPutInformationJob() throws Exception {
 445         // Test that one producer accepting a job is enough
 446         putInfoProducerWithOneType(PRODUCER_ID, TYPE_ID);
 447
 448         String url = ConsumerConsts.API_ROOT + "/info-jobs/jobId";
 449         String body = gson.toJson(consumerJobInfo());
 450         ResponseEntity<String> resp = restClient().putForEntity(url, body).block();
 451         assertThat(this.infoJobs.size()).isEqualTo(1);
 452         assertThat(resp.getStatusCode()).isEqualTo(HttpStatus.CREATED);
 453
 454         ProducerSimulatorController.TestResults simulatorResults = this.producerSimulator.getTestResults();
 455         await().untilAsserted(() -> assertThat(simulatorResults.jobsStarted.size()).isEqualTo(1));
 456         ProducerJobInfo request = simulatorResults.jobsStarted.get(0);
 457         assertThat(request.id).isEqualTo("jobId");
 458
 459         resp = restClient().putForEntity(url, body).block();
 460         assertThat(resp.getStatusCode()).isEqualTo(HttpStatus.OK);
 461         InfoJob job = this.infoJobs.getJob("jobId");
 462         assertThat(job.getOwner()).isEqualTo("owner");
 463
 464         verifyJobStatus(EI_JOB_ID, "ENABLED");
 465     }
 466
 467     @Test
 468     void consumerPutInformationJob_noType() throws JsonMappingException, JsonProcessingException, ServiceException {
 469         String url = ConsumerConsts.API_ROOT + "/info-jobs/jobId?typeCheck=false";
 470         String body = gson.toJson(consumerJobInfo());
 471         ResponseEntity<String> resp = restClient().putForEntity(url, body).block();
 472         assertThat(this.infoJobs.size()).isEqualTo(1);
 473         assertThat(resp.getStatusCode()).isEqualTo(HttpStatus.CREATED);
 474         verifyJobStatus(EI_JOB_ID, "DISABLED");
 475
 476         putInfoProducerWithOneType(PRODUCER_ID, TYPE_ID);
 477
 478         verifyJobStatus(EI_JOB_ID, "ENABLED");
 479     }
 480
 481     @Test
 482     void a1ePutEiJob_jsonSchemavalidationError() throws Exception {
 483         putInfoProducerWithOneType(PRODUCER_ID, TYPE_ID);
 484
 485         String url = A1eConsts.API_ROOT + "/eijobs/jobId";
 486         // The element with name "property1" is mandatory in the schema
 487         A1eEiJobInfo jobInfo = new A1eEiJobInfo("typeId", jsonObject("{ \"XXstring\" : \"value\" }"), "owner",
 488             "targetUri", "jobStatusUrl");
 489         String body = gson.toJson(jobInfo);
 490
 491         testErrorCode(restClient().put(url, body), HttpStatus.CONFLICT, "Json validation failure");
 492     }
 493
 494     @Test
 495     void consumerPutJob_jsonSchemavalidationError() throws Exception {
 496         putInfoProducerWithOneType(PRODUCER_ID, TYPE_ID);
 497
 498         String url = ConsumerConsts.API_ROOT + "/info-jobs/jobId?typeCheck=true";
 499         // The element with name "property1" is mandatory in the schema
 500         ConsumerJobInfo jobInfo =
 501             new ConsumerJobInfo("typeId", jsonObject("{ \"XXstring\" : \"value\" }"), "owner", "targetUri", null);
 502         String body = gson.toJson(jobInfo);
 503
 504         testErrorCode(restClient().put(url, body), HttpStatus.CONFLICT, "Json validation failure");
 505     }
 506
 507     @Test
 508     void consumerPutJob_uriError() throws Exception {
 509         putInfoProducerWithOneType(PRODUCER_ID, TYPE_ID);
 510
 511         String url = ConsumerConsts.API_ROOT + "/info-jobs/jobId?typeCheck=true";
 512
 513         ConsumerJobInfo jobInfo = new ConsumerJobInfo(TYPE_ID, jsonObject(), "owner", "junk", null);
 514         String body = gson.toJson(jobInfo);
 515
 516         testErrorCode(restClient().put(url, body), HttpStatus.CONFLICT, "URI: junk is not absolute");
 517     }
 518
 519     @Test
 520     void a1eChangingEiTypeGetRejected() throws Exception {
 521         putInfoProducerWithOneType("producer1", "typeId1");
 522         putInfoProducerWithOneType("producer2", "typeId2");
 523         putEiJob("typeId1", "jobId");
 524
 525         String url = A1eConsts.API_ROOT + "/eijobs/jobId";
 526         String body = gson.toJson(infoJobInfo("typeId2", "jobId"));
 527         testErrorCode(restClient().put(url, body), HttpStatus.CONFLICT,
 528             "Not allowed to change type for existing EI job");
 529     }
 530
 531     @Test
 532     void consumerChangingInfoTypeGetRejected() throws Exception {
 533         putInfoProducerWithOneType("producer1", "typeId1");
 534         putInfoProducerWithOneType("producer2", "typeId2");
 535         putEiJob("typeId1", "jobId");
 536
 537         String url = ConsumerConsts.API_ROOT + "/info-jobs/jobId";
 538         String body = gson.toJson(consumerJobInfo("typeId2", "jobId"));
 539         testErrorCode(restClient().put(url, body), HttpStatus.CONFLICT, "Not allowed to change type for existing job");
 540     }
 541
 542     @Test
 543     void producerPutEiType() throws JsonMappingException, JsonProcessingException, ServiceException {
 544         assertThat(putInfoType(TYPE_ID)).isEqualTo(HttpStatus.CREATED);
 545         assertThat(putInfoType(TYPE_ID)).isEqualTo(HttpStatus.OK);
 546     }
 547
 548     @Test
 549     void producerPutEiType_noSchema() {
 550         String url = ProducerConsts.API_ROOT + "/info-types/" + TYPE_ID;
 551         String body = "{}";
 552         testErrorCode(restClient().put(url, body), HttpStatus.BAD_REQUEST, "No schema provided");
 553     }
 554
 555     @Test
 556     void producerDeleteEiType() throws Exception {
 557         putInfoType(TYPE_ID);
 558         this.putEiJob(TYPE_ID, "job1");
 559         this.putEiJob(TYPE_ID, "job2");
 560         deleteInfoType(TYPE_ID);
 561
 562         assertThat(this.infoTypes.size()).isZero();
 563         assertThat(this.infoJobs.size()).isZero(); // Test that also the job is deleted
 564
 565         testErrorCode(restClient().delete(deleteInfoTypeUrl(TYPE_ID)), HttpStatus.NOT_FOUND,
 566             "Information type not found");
 567     }
 568
 569     @Test
 570     void producerDeleteEiTypeExistingProducer() throws Exception {
 571         putInfoProducerWithOneType(PRODUCER_ID, TYPE_ID);
 572         String url = ProducerConsts.API_ROOT + "/info-types/" + TYPE_ID;
 573         testErrorCode(restClient().delete(url), HttpStatus.NOT_ACCEPTABLE,
 574             "The type has active producers: " + PRODUCER_ID);
 575         assertThat(this.infoTypes.size()).isEqualTo(1);
 576     }
 577
 578     @Test
 579     void producerPutProducerWithOneType_rejecting()
 580         throws JsonMappingException, JsonProcessingException, ServiceException {
 581         putEiProducerWithOneTypeRejecting("simulateProducerError", TYPE_ID);
 582         String url = A1eConsts.API_ROOT + "/eijobs/" + EI_JOB_ID;
 583         String body = gson.toJson(infoJobInfo());
 584         restClient().put(url, body).block();
 585
 586         ProducerSimulatorController.TestResults simulatorResults = this.producerSimulator.getTestResults();
 587         // There is one retry -> 2 calls
 588         await().untilAsserted(() -> assertThat(simulatorResults.noOfRejectedCreate).isEqualTo(2));
 589         assertThat(simulatorResults.noOfRejectedCreate).isEqualTo(2);
 590
 591         verifyJobStatus(EI_JOB_ID, "DISABLED");
 592     }
 593
 594     @Test
 595     void producerGetEiProducerTypes() throws Exception {
 596         final String EI_TYPE_ID_2 = TYPE_ID + "_2";
 597         putInfoProducerWithOneType("producer1", TYPE_ID);
 598         putEiJob(TYPE_ID, "jobId");
 599         putInfoProducerWithOneType("producer2", EI_TYPE_ID_2);
 600         putEiJob(EI_TYPE_ID_2, "jobId2");
 601         String url = ProducerConsts.API_ROOT + "/info-types";
 602
 603         ResponseEntity<String> resp = restClient().getForEntity(url).block();
 604         assertThat(resp.getStatusCode()).isEqualTo(HttpStatus.OK);
 605         assertThat(resp.getBody()).contains(TYPE_ID);
 606         assertThat(resp.getBody()).contains(EI_TYPE_ID_2);
 607     }
 608
 609     @Test
 610     void producerPutEiProducer() throws Exception {
 611         this.putInfoType(TYPE_ID);
 612         String url = ProducerConsts.API_ROOT + "/info-producers/infoProducerId";
 613         String body = gson.toJson(producerInfoRegistratioInfo(TYPE_ID));
 614
 615         ResponseEntity<String> resp = restClient().putForEntity(url, body).block();
 616         assertThat(resp.getStatusCode()).isEqualTo(HttpStatus.CREATED);
 617
 618         assertThat(this.infoTypes.size()).isEqualTo(1);
 619         assertThat(this.infoProducers.getProducersForType(TYPE_ID)).hasSize(1);
 620         assertThat(this.infoProducers.size()).isEqualTo(1);
 621         assertThat(this.infoProducers.get("infoProducerId").getInfoTypes().iterator().next().getId())
 622             .isEqualTo(TYPE_ID);
 623
 624         resp = restClient().putForEntity(url, body).block();
 625         assertThat(resp.getStatusCode()).isEqualTo(HttpStatus.OK);
 626
 627         resp = restClient().getForEntity(url).block();
 628         assertThat(resp.getStatusCode()).isEqualTo(HttpStatus.OK);
 629         assertThat(resp.getBody()).isEqualTo(body);
 630     }
 631
 632     @Test
 633     void producerPutEiProducerExistingJob() throws Exception {
 634         putInfoProducerWithOneType(PRODUCER_ID, TYPE_ID);
 635         putEiJob(TYPE_ID, "jobId");
 636         String url = ProducerConsts.API_ROOT + "/info-producers/infoProducerId";
 637         String body = gson.toJson(producerInfoRegistratioInfo(TYPE_ID));
 638         restClient().putForEntity(url, body).block();
 639
 640         ProducerSimulatorController.TestResults simulatorResults = this.producerSimulator.getTestResults();
 641         await().untilAsserted(() -> assertThat(simulatorResults.jobsStarted.size()).isEqualTo(2));
 642         ProducerJobInfo request = simulatorResults.jobsStarted.get(0);
 643         assertThat(request.id).isEqualTo("jobId");
 644     }
 645
 646     @Test
 647     void testPutEiProducer_noType() throws Exception {
 648         String url = ProducerConsts.API_ROOT + "/info-producers/infoProducerId";
 649         String body = gson.toJson(producerInfoRegistratioInfo(TYPE_ID));
 650         testErrorCode(restClient().put(url, body), HttpStatus.NOT_FOUND, "Information type not found");
 651     }
 652
 653     @Test
 654     void producerPutProducerAndEiJob() throws Exception {
 655         this.putInfoType(TYPE_ID);
 656         String url = ProducerConsts.API_ROOT + "/info-producers/infoProducerId";
 657         String body = gson.toJson(producerInfoRegistratioInfo(TYPE_ID));
 658         restClient().putForEntity(url, body).block();
 659         assertThat(this.infoTypes.size()).isEqualTo(1);
 660         this.infoTypes.getType(TYPE_ID);
 661
 662         url = A1eConsts.API_ROOT + "/eijobs/jobId";
 663         body = gson.toJson(infoJobInfo());
 664         restClient().putForEntity(url, body).block();
 665
 666         ProducerSimulatorController.TestResults simulatorResults = this.producerSimulator.getTestResults();
 667         await().untilAsserted(() -> assertThat(simulatorResults.jobsStarted.size()).isEqualTo(1));
 668         ProducerJobInfo request = simulatorResults.jobsStarted.get(0);
 669         assertThat(request.id).isEqualTo("jobId");
 670     }
 671
 672     @Test
 673     void producerGetEiJobsForProducer() throws JsonMappingException, JsonProcessingException, ServiceException {
 674         putInfoProducerWithOneType(PRODUCER_ID, TYPE_ID);
 675         putEiJob(TYPE_ID, "jobId1");
 676         putEiJob(TYPE_ID, "jobId2");
 677
 678         // PUT a consumerRestApiTestBase.java
 679         String url = ProducerConsts.API_ROOT + "/info-producers/infoProducerId";
 680         String body = gson.toJson(producerInfoRegistratioInfo(TYPE_ID));
 681         restClient().putForEntity(url, body).block();
 682
 683         url = ProducerConsts.API_ROOT + "/info-producers/infoProducerId/info-jobs";
 684         ResponseEntity<String> resp = restClient().getForEntity(url).block();
 685         assertThat(resp.getStatusCode()).isEqualTo(HttpStatus.OK);
 686
 687         ProducerJobInfo[] parsedResp = gson.fromJson(resp.getBody(), ProducerJobInfo[].class);
 688         assertThat(parsedResp[0].typeId).isEqualTo(TYPE_ID);
 689         assertThat(parsedResp[1].typeId).isEqualTo(TYPE_ID);
 690     }
 691
 692     @Test
 693     void producerDeleteEiProducer() throws Exception {
 694         putInfoProducerWithOneType("infoProducerId", TYPE_ID);
 695         putInfoProducerWithOneType("infoProducerId2", TYPE_ID);
 696
 697         assertThat(this.infoProducers.size()).isEqualTo(2);
 698         InfoType type = this.infoTypes.getType(TYPE_ID);
 699         assertThat(this.infoProducers.getProducerIdsForType(type.getId())).contains("infoProducerId");
 700         assertThat(this.infoProducers.getProducerIdsForType(type.getId())).contains("infoProducerId2");
 701         putEiJob(TYPE_ID, "jobId");
 702         assertThat(this.infoJobs.size()).isEqualTo(1);
 703
 704         deleteEiProducer("infoProducerId");
 705         assertThat(this.infoProducers.size()).isEqualTo(1);
 706         assertThat(this.infoProducers.getProducerIdsForType(TYPE_ID)).doesNotContain("infoProducerId");
 707         verifyJobStatus("jobId", "ENABLED");
 708
 709         deleteEiProducer("infoProducerId2");
 710         assertThat(this.infoProducers.size()).isZero();
 711         assertThat(this.infoTypes.size()).isEqualTo(1);
 712         verifyJobStatus("jobId", "DISABLED");
 713     }
 714
 715     @Test
 716     void a1eJobStatusNotifications() throws JsonMappingException, JsonProcessingException, ServiceException {
 717         ConsumerSimulatorController.TestResults consumerCalls = this.consumerSimulator.getTestResults();
 718         ProducerSimulatorController.TestResults producerCalls = this.producerSimulator.getTestResults();
 719
 720         putInfoProducerWithOneType("infoProducerId", TYPE_ID);
 721         putEiJob(TYPE_ID, "jobId");
 722         putInfoProducerWithOneType("infoProducerId2", TYPE_ID);
 723         await().untilAsserted(() -> assertThat(producerCalls.jobsStarted.size()).isEqualTo(2));
 724
 725         deleteEiProducer("infoProducerId2");
 726         assertThat(this.infoTypes.size()).isEqualTo(1); // The type remains, one producer left
 727         deleteEiProducer("infoProducerId");
 728         assertThat(this.infoTypes.size()).isEqualTo(1); // The type remains
 729         assertThat(this.infoJobs.size()).isEqualTo(1); // The job remains
 730         await().untilAsserted(() -> assertThat(consumerCalls.eiJobStatusCallbacks.size()).isEqualTo(1));
 731         assertThat(consumerCalls.eiJobStatusCallbacks.get(0).state)
 732             .isEqualTo(A1eEiJobStatus.EiJobStatusValues.DISABLED);
 733
 734         putInfoProducerWithOneType("infoProducerId", TYPE_ID);
 735         await().untilAsserted(() -> assertThat(consumerCalls.eiJobStatusCallbacks.size()).isEqualTo(2));
 736         assertThat(consumerCalls.eiJobStatusCallbacks.get(1).state).isEqualTo(A1eEiJobStatus.EiJobStatusValues.ENABLED);
 737     }
 738
 739     @Test
 740     void a1eJobStatusNotifications2() throws JsonMappingException, JsonProcessingException, ServiceException {
 741         // Test replacing a producer with new and removed types
 742
 743         // Create a job
 744         putInfoProducerWithOneType(PRODUCER_ID, TYPE_ID);
 745         putEiJob(TYPE_ID, EI_JOB_ID);
 746
 747         // change the type for the producer, the job shall be disabled
 748         putInfoProducerWithOneType(PRODUCER_ID, "junk");
 749         verifyJobStatus(EI_JOB_ID, "DISABLED");
 750         ConsumerSimulatorController.TestResults consumerCalls = this.consumerSimulator.getTestResults();
 751         await().untilAsserted(() -> assertThat(consumerCalls.eiJobStatusCallbacks.size()).isEqualTo(1));
 752         assertThat(consumerCalls.eiJobStatusCallbacks.get(0).state)
 753             .isEqualTo(A1eEiJobStatus.EiJobStatusValues.DISABLED);
 754
 755         putInfoProducerWithOneType(PRODUCER_ID, TYPE_ID);
 756         verifyJobStatus(EI_JOB_ID, "ENABLED");
 757         await().untilAsserted(() -> assertThat(consumerCalls.eiJobStatusCallbacks.size()).isEqualTo(2));
 758         assertThat(consumerCalls.eiJobStatusCallbacks.get(1).state).isEqualTo(A1eEiJobStatus.EiJobStatusValues.ENABLED);
 759     }
 760
 761     @Test
 762     void producerGetProducerEiType() throws JsonMappingException, JsonProcessingException, ServiceException {
 763         putInfoProducerWithOneType(PRODUCER_ID, TYPE_ID);
 764         String url = ProducerConsts.API_ROOT + "/info-types/" + TYPE_ID;
 765         ResponseEntity<String> resp = restClient().getForEntity(url).block();
 766         ProducerInfoTypeInfo info = gson.fromJson(resp.getBody(), ProducerInfoTypeInfo.class);
 767         assertThat(info.jobDataSchema).isNotNull();
 768         assertThat(info.typeSpecificInformation).isNotNull();
 769     }
 770
 771     @Test
 772     void producerGetProducerIdentifiers() throws JsonMappingException, JsonProcessingException, ServiceException {
 773         putInfoProducerWithOneType(PRODUCER_ID, TYPE_ID);
 774         String url = ProducerConsts.API_ROOT + "/info-producers";
 775         ResponseEntity<String> resp = restClient().getForEntity(url).block();
 776         assertThat(resp.getBody()).contains(PRODUCER_ID);
 777
 778         url = ProducerConsts.API_ROOT + "/info-producers?info_type_id=" + TYPE_ID;
 779         resp = restClient().getForEntity(url).block();
 780         assertThat(resp.getBody()).contains(PRODUCER_ID);
 781
 782         url = ProducerConsts.API_ROOT + "/info-producers?info_type_id=junk";
 783         resp = restClient().getForEntity(url).block();
 784         assertThat(resp.getBody()).isEqualTo("[]");
 785     }
 786
 787     @Test
 788     void producerSupervision() throws JsonMappingException, JsonProcessingException, ServiceException {
 789
 790         ConsumerSimulatorController.TestResults consumerResults = this.consumerSimulator.getTestResults();
 791         putEiProducerWithOneTypeRejecting("simulateProducerError", TYPE_ID);
 792
 793         {
 794             // Create a job
 795             putInfoProducerWithOneType(PRODUCER_ID, TYPE_ID);
 796             putEiJob(TYPE_ID, EI_JOB_ID);
 797             verifyJobStatus(EI_JOB_ID, "ENABLED");
 798             deleteEiProducer(PRODUCER_ID);
 799             // A Job disabled status notification shall now be received
 800             await().untilAsserted(() -> assertThat(consumerResults.eiJobStatusCallbacks.size()).isEqualTo(1));
 801             assertThat(consumerResults.eiJobStatusCallbacks.get(0).state)
 802                 .isEqualTo(A1eEiJobStatus.EiJobStatusValues.DISABLED);
 803             verifyJobStatus(EI_JOB_ID, "DISABLED");
 804         }
 805
 806         assertThat(this.infoProducers.size()).isEqualTo(1);
 807         assertThat(this.infoTypes.size()).isEqualTo(1);
 808         assertProducerOpState("simulateProducerError", ProducerStatusInfo.OperationalState.ENABLED);
 809
 810         this.producerSupervision.createTask().blockLast();
 811         this.producerSupervision.createTask().blockLast();
 812
 813         // Now we have one producer that is disabled
 814         assertThat(this.infoProducers.size()).isEqualTo(1);
 815         assertProducerOpState("simulateProducerError", ProducerStatusInfo.OperationalState.DISABLED);
 816
 817         // After 3 failed checks, the producer shall be deregistered
 818         this.producerSupervision.createTask().blockLast();
 819         assertThat(this.infoProducers.size()).isZero(); // The producer is removed
 820         assertThat(this.infoTypes.size()).isEqualTo(1); // The type remains
 821
 822         // Now we have one disabled job, and no producer.
 823         // PUT a producer, then a Job ENABLED status notification shall be received
 824         putInfoProducerWithOneType(PRODUCER_ID, TYPE_ID);
 825         await().untilAsserted(() -> assertThat(consumerResults.eiJobStatusCallbacks.size()).isEqualTo(2));
 826         assertThat(consumerResults.eiJobStatusCallbacks.get(1).state)
 827             .isEqualTo(A1eEiJobStatus.EiJobStatusValues.ENABLED);
 828         verifyJobStatus(EI_JOB_ID, "ENABLED");
 829     }
 830
 831     @Test
 832     void producerSupervision2() throws JsonMappingException, JsonProcessingException, ServiceException {
 833         // Test that supervision enables not enabled jobs and sends a notification when
 834         // suceeded
 835
 836         putInfoProducerWithOneType(PRODUCER_ID, TYPE_ID);
 837         putEiJob(TYPE_ID, EI_JOB_ID);
 838
 839         InfoProducer producer = this.infoProducers.getProducer(PRODUCER_ID);
 840         InfoJob job = this.infoJobs.getJob(EI_JOB_ID);
 841         // Pretend that the producer did reject the job and the a DISABLED notification
 842         // is sent for the job
 843         producer.setJobDisabled(job);
 844         job.setLastReportedStatus(false);
 845         verifyJobStatus(EI_JOB_ID, "DISABLED");
 846
 847         // Run the supervision and wait for the job to get started in the producer
 848         this.producerSupervision.createTask().blockLast();
 849         ConsumerSimulatorController.TestResults consumerResults = this.consumerSimulator.getTestResults();
 850         await().untilAsserted(() -> assertThat(consumerResults.eiJobStatusCallbacks.size()).isEqualTo(1));
 851         assertThat(consumerResults.eiJobStatusCallbacks.get(0).state)
 852             .isEqualTo(A1eEiJobStatus.EiJobStatusValues.ENABLED);
 853         verifyJobStatus(EI_JOB_ID, "ENABLED");
 854     }
 855
 856     @Test
 857     void testGetStatus() throws JsonMappingException, JsonProcessingException, ServiceException {
 858         putEiProducerWithOneTypeRejecting("simulateProducerError", TYPE_ID);
 859         putEiProducerWithOneTypeRejecting("simulateProducerError2", TYPE_ID);
 860
 861         String url = "/status";
 862         ResponseEntity<String> resp = restClient().getForEntity(url).block();
 863         assertThat(resp.getBody()).contains("hunky dory");
 864     }
 865
 866     @Test
 867     void testEiJobDatabase() throws Exception {
 868         putInfoProducerWithOneType(PRODUCER_ID, TYPE_ID);
 869         putEiJob(TYPE_ID, "jobId1");
 870         putEiJob(TYPE_ID, "jobId2");
 871
 872         assertThat(this.infoJobs.size()).isEqualTo(2);
 873
 874         {
 875             InfoJob savedJob = this.infoJobs.getJob("jobId1");
 876             // Restore the jobs
 877             InfoJobs jobs = new InfoJobs(this.applicationConfig, this.producerCallbacks);
 878             jobs.restoreJobsFromDatabase();
 879             assertThat(jobs.size()).isEqualTo(2);
 880             InfoJob restoredJob = jobs.getJob("jobId1");
 881             assertThat(restoredJob.getId()).isEqualTo("jobId1");
 882             assertThat(restoredJob.getLastUpdated()).isEqualTo(savedJob.getLastUpdated());
 883
 884             jobs.remove("jobId1", this.infoProducers);
 885             jobs.remove("jobId2", this.infoProducers);
 886         }
 887         {
 888             // Restore the jobs, no jobs in database
 889             InfoJobs jobs = new InfoJobs(this.applicationConfig, this.producerCallbacks);
 890             jobs.restoreJobsFromDatabase();
 891             assertThat(jobs.size()).isZero();
 892         }
 893         logger.warn("Test removing a job when the db file is gone");
 894         this.infoJobs.remove("jobId1", this.infoProducers);
 895         assertThat(this.infoJobs.size()).isEqualTo(1);
 896
 897         ProducerSimulatorController.TestResults simulatorResults = this.producerSimulator.getTestResults();
 898         await().untilAsserted(() -> assertThat(simulatorResults.jobsStopped.size()).isEqualTo(3));
 899     }
 900
 901     @Test
 902     void testEiTypesDatabase() throws Exception {
 903         putInfoProducerWithOneType(PRODUCER_ID, TYPE_ID);
 904
 905         assertThat(this.infoTypes.size()).isEqualTo(1);
 906
 907         {
 908             // Restore the types
 909             InfoTypes types = new InfoTypes(this.applicationConfig);
 910             types.restoreTypesFromDatabase();
 911             assertThat(types.size()).isEqualTo(1);
 912         }
 913         {
 914             // Restore the jobs, no jobs in database
 915             InfoTypes types = new InfoTypes(this.applicationConfig);
 916             types.clear();
 917             types.restoreTypesFromDatabase();
 918             assertThat(types.size()).isZero();
 919         }
 920         logger.warn("Test removing a job when the db file is gone");
 921         this.infoTypes.remove(this.infoTypes.getType(TYPE_ID));
 922         assertThat(this.infoJobs.size()).isZero();
 923     }
 924
 925     @Test
 926     void testConsumerTypeSubscriptionDatabase() {
 927         final String callbackUrl = baseUrl() + ConsumerSimulatorController.getTypeStatusCallbackUrl();
 928         final ConsumerTypeSubscriptionInfo info = new ConsumerTypeSubscriptionInfo(callbackUrl, "owner");
 929         // PUT a subscription
 930         String body = gson.toJson(info);
 931         restClient().putForEntity(typeSubscriptionUrl() + "/subscriptionId", body).block();
 932         assertThat(this.infoTypeSubscriptions.size()).isEqualTo(1);
 933
 934         InfoTypeSubscriptions restoredSubscriptions = new InfoTypeSubscriptions(this.applicationConfig);
 935         assertThat(restoredSubscriptions.size()).isEqualTo(1);
 936         assertThat(restoredSubscriptions.getSubscriptionsForOwner("owner")).hasSize(1);
 937
 938         // Delete the subscription
 939         restClient().deleteForEntity(typeSubscriptionUrl() + "/subscriptionId").block();
 940         restoredSubscriptions = new InfoTypeSubscriptions(this.applicationConfig);
 941         assertThat(restoredSubscriptions.size()).isZero();
 942     }
 943
 944     @Test
 945     void testConsumerTypeSubscription() throws Exception {
 946
 947         final String callbackUrl = baseUrl() + ConsumerSimulatorController.getTypeStatusCallbackUrl();
 948         final ConsumerTypeSubscriptionInfo info = new ConsumerTypeSubscriptionInfo(callbackUrl, "owner");
 949
 950         {
 951             // PUT a subscription
 952             String body = gson.toJson(info);
 953             ResponseEntity<String> resp =
 954                 restClient().putForEntity(typeSubscriptionUrl() + "/subscriptionId", body).block();
 955             assertThat(this.infoTypeSubscriptions.size()).isEqualTo(1);
 956             assertThat(resp.getStatusCode()).isEqualTo(HttpStatus.CREATED);
 957             resp = restClient().putForEntity(typeSubscriptionUrl() + "/subscriptionId", body).block();
 958             assertThat(resp.getStatusCode()).isEqualTo(HttpStatus.OK);
 959         }
 960         {
 961             // GET IDs
 962             ResponseEntity<String> resp = restClient().getForEntity(typeSubscriptionUrl()).block();
 963             assertThat(resp.getBody()).isEqualTo("[\"subscriptionId\"]");
 964             resp = restClient().getForEntity(typeSubscriptionUrl() + "?owner=owner").block();
 965             assertThat(resp.getBody()).isEqualTo("[\"subscriptionId\"]");
 966             resp = restClient().getForEntity(typeSubscriptionUrl() + "?owner=junk").block();
 967             assertThat(resp.getBody()).isEqualTo("[]");
 968         }
 969
 970         {
 971             // GET the individual subscription
 972             ResponseEntity<String> resp = restClient().getForEntity(typeSubscriptionUrl() + "/subscriptionId").block();
 973             ConsumerTypeSubscriptionInfo respInfo = gson.fromJson(resp.getBody(), ConsumerTypeSubscriptionInfo.class);
 974             assertThat(respInfo).isEqualTo(info);
 975         }
 976
 977         {
 978             // Test the callbacks
 979             final ConsumerSimulatorController.TestResults consumerCalls = this.consumerSimulator.getTestResults();
 980
 981             // Test callback for PUT type
 982             this.putInfoType(TYPE_ID);
 983             await().untilAsserted(() -> assertThat(consumerCalls.typeRegistrationInfoCallbacks.size()).isEqualTo(1));
 984             assertThat(consumerCalls.typeRegistrationInfoCallbacks.get(0).state)
 985                 .isEqualTo(ConsumerTypeRegistrationInfo.ConsumerTypeStatusValues.REGISTERED);
 986
 987             // Test callback for DELETE type
 988             this.deleteInfoType(TYPE_ID);
 989             await().untilAsserted(() -> assertThat(consumerCalls.typeRegistrationInfoCallbacks.size()).isEqualTo(2));
 990             assertThat(consumerCalls.typeRegistrationInfoCallbacks.get(1).state)
 991                 .isEqualTo(ConsumerTypeRegistrationInfo.ConsumerTypeStatusValues.DEREGISTERED);
 992         }
 993
 994         {
 995             // DELETE the subscription
 996             ResponseEntity<String> resp =
 997                 restClient().deleteForEntity(typeSubscriptionUrl() + "/subscriptionId").block();
 998             assertThat(resp.getStatusCode()).isEqualTo(HttpStatus.NO_CONTENT);
 999             assertThat(this.infoTypeSubscriptions.size()).isZero();
1000             resp = restClient().getForEntity(typeSubscriptionUrl()).block();
1001             assertThat(resp.getBody()).isEqualTo("[]");
1002         }
1003     }
1004
1005     @Test
1006     void testRemovingNonWorkingSubscription() throws Exception {
1007         // Test that subscriptions are removed for a unresponsive consumer
1008
1009         // PUT a subscription with a junk callback
1010         final ConsumerTypeSubscriptionInfo info = new ConsumerTypeSubscriptionInfo(baseUrl() + "JUNK", "owner");
1011         String body = gson.toJson(info);
1012         restClient().putForEntity(typeSubscriptionUrl() + "/subscriptionId", body).block();
1013         assertThat(this.infoTypeSubscriptions.size()).isEqualTo(1);
1014
1015         this.putInfoType(TYPE_ID);
1016         // The callback will fail and the subscription will be removed
1017         await().untilAsserted(() -> assertThat(this.infoTypeSubscriptions.size()).isZero());
1018     }
1019
1020     @Test
1021     void testTypeSubscriptionErrorCodes() throws Exception {
1022
1023         testErrorCode(restClient().get(typeSubscriptionUrl() + "/junk"), HttpStatus.NOT_FOUND,
1024             "Could not find Information subscription: junk");
1025
1026         testErrorCode(restClient().delete(typeSubscriptionUrl() + "/junk"), HttpStatus.NOT_FOUND,
1027             "Could not find Information subscription: junk");
1028     }
1029
1030     private String typeSubscriptionUrl() {
1031         return ConsumerConsts.API_ROOT + "/info-type-subscription";
1032     }
1033
1034     private void deleteEiProducer(String infoProducerId) {
1035         String url = ProducerConsts.API_ROOT + "/info-producers/" + infoProducerId;
1036         restClient().deleteForEntity(url).block();
1037     }
1038
1039     private void verifyJobStatus(String jobId, String expStatus) {
1040         String url = A1eConsts.API_ROOT + "/eijobs/" + jobId + "/status";
1041         String rsp = restClient().get(url).block();
1042         assertThat(rsp).contains(expStatus);
1043     }
1044
1045     private void assertProducerOpState(String producerId,
1046         ProducerStatusInfo.OperationalState expectedOperationalState) {
1047         String statusUrl = ProducerConsts.API_ROOT + "/info-producers/" + producerId + "/status";
1048         ResponseEntity<String> resp = restClient().getForEntity(statusUrl).block();
1049         ProducerStatusInfo statusInfo = gson.fromJson(resp.getBody(), ProducerStatusInfo.class);
1050         assertThat(statusInfo.opState).isEqualTo(expectedOperationalState);
1051     }
1052
1053     ProducerInfoTypeInfo producerEiTypeRegistrationInfo(String typeId)
1054         throws JsonMappingException, JsonProcessingException {
1055         return new ProducerInfoTypeInfo(jsonSchemaObject(), typeSpecifcInfoObject());
1056     }
1057
1058     ProducerRegistrationInfo producerEiRegistratioInfoRejecting(String typeId)
1059         throws JsonMappingException, JsonProcessingException {
1060         return new ProducerRegistrationInfo(Arrays.asList(typeId), //
1061             baseUrl() + ProducerSimulatorController.JOB_ERROR_URL,
1062             baseUrl() + ProducerSimulatorController.SUPERVISION_ERROR_URL);
1063     }
1064
1065     ProducerRegistrationInfo producerInfoRegistratioInfo(String typeId)
1066         throws JsonMappingException, JsonProcessingException {
1067         return new ProducerRegistrationInfo(Arrays.asList(typeId), //
1068             baseUrl() + ProducerSimulatorController.JOB_URL, baseUrl() + ProducerSimulatorController.SUPERVISION_URL);
1069     }
1070
1071     private ConsumerJobInfo consumerJobInfo() throws JsonMappingException, JsonProcessingException {
1072         return consumerJobInfo(TYPE_ID, EI_JOB_ID);
1073     }
1074
1075     ConsumerJobInfo consumerJobInfo(String typeId, String infoJobId)
1076         throws JsonMappingException, JsonProcessingException {
1077         return new ConsumerJobInfo(typeId, jsonObject(), "owner", "https://junk.com",
1078             baseUrl() + ConsumerSimulatorController.getJobStatusUrl(infoJobId));
1079     }
1080
1081     private A1eEiJobInfo infoJobInfo() throws JsonMappingException, JsonProcessingException {
1082         return infoJobInfo(TYPE_ID, EI_JOB_ID);
1083     }
1084
1085     A1eEiJobInfo infoJobInfo(String typeId, String infoJobId) throws JsonMappingException, JsonProcessingException {
1086         return new A1eEiJobInfo(typeId, jsonObject(), "owner", "https://junk.com",
1087             baseUrl() + ConsumerSimulatorController.getJobStatusUrl(infoJobId));
1088     }
1089
1090     private Object jsonObject(String json) {
1091         try {
1092             return JsonParser.parseString(json).getAsJsonObject();
1093         } catch (Exception e) {
1094             throw new NullPointerException(e.toString());
1095         }
1096     }
1097
1098     private Object typeSpecifcInfoObject() {
1099         return jsonObject("{ \"propertyName\" : \"value\" }");
1100     }
1101
1102     private Object jsonSchemaObject() {
1103         // a json schema with one mandatory property named "string"
1104         String schemaStr = "{" //
1105             + "\"$schema\": \"http://json-schema.org/draft-04/schema#\"," //
1106             + "\"type\": \"object\"," //
1107             + "\"properties\": {" //
1108             + EI_JOB_PROPERTY + " : {" //
1109             + "    \"type\": \"string\"" //
1110             + "  }" //
1111             + "}," //
1112             + "\"required\": [" //
1113             + EI_JOB_PROPERTY //
1114             + "]" //
1115             + "}"; //
1116         return jsonObject(schemaStr);
1117     }
1118
1119     private Object jsonObject() {
1120         return jsonObject("{ " + EI_JOB_PROPERTY + " : \"value\" }");
1121     }
1122
1123     private InfoJob putEiJob(String infoTypeId, String jobId)
1124         throws JsonMappingException, JsonProcessingException, ServiceException {
1125
1126         String url = A1eConsts.API_ROOT + "/eijobs/" + jobId;
1127         String body = gson.toJson(infoJobInfo(infoTypeId, jobId));
1128         restClient().putForEntity(url, body).block();
1129
1130         return this.infoJobs.getJob(jobId);
1131     }
1132
1133     private HttpStatus putInfoType(String infoTypeId)
1134         throws JsonMappingException, JsonProcessingException, ServiceException {
1135         String url = ProducerConsts.API_ROOT + "/info-types/" + infoTypeId;
1136         String body = gson.toJson(producerEiTypeRegistrationInfo(infoTypeId));
1137
1138         ResponseEntity<String> resp = restClient().putForEntity(url, body).block();
1139         this.infoTypes.getType(infoTypeId);
1140         return resp.getStatusCode();
1141     }
1142
1143     private String deleteInfoTypeUrl(String typeId) {
1144         return ProducerConsts.API_ROOT + "/info-types/" + typeId;
1145     }
1146
1147     private void deleteInfoType(String typeId) {
1148         restClient().delete(deleteInfoTypeUrl(typeId)).block();
1149     }
1150
1151     private InfoType putEiProducerWithOneTypeRejecting(String producerId, String infoTypeId)
1152         throws JsonMappingException, JsonProcessingException, ServiceException {
1153         this.putInfoType(infoTypeId);
1154         String url = ProducerConsts.API_ROOT + "/info-producers/" + producerId;
1155         String body = gson.toJson(producerEiRegistratioInfoRejecting(infoTypeId));
1156         restClient().putForEntity(url, body).block();
1157         return this.infoTypes.getType(infoTypeId);
1158     }
1159
1160     private InfoType putInfoProducerWithOneType(String producerId, String infoTypeId)
1161         throws JsonMappingException, JsonProcessingException, ServiceException {
1162         this.putInfoType(infoTypeId);
1163
1164         String url = ProducerConsts.API_ROOT + "/info-producers/" + producerId;
1165         String body = gson.toJson(producerInfoRegistratioInfo(infoTypeId));
1166
1167         restClient().putForEntity(url, body).block();
1168
1169         return this.infoTypes.getType(infoTypeId);
1170     }
1171
1172     private String baseUrl() {
1173         return "https://localhost:" + this.port;
1174     }
1175
1176     private AsyncRestClient restClient(boolean useTrustValidation) {
1177         WebClientConfig config = this.applicationConfig.getWebClientConfig();
1178         HttpProxyConfig httpProxyConfig = ImmutableHttpProxyConfig.builder() //
1179             .httpProxyHost("") //
1180             .httpProxyPort(0) //
1181             .build();
1182         config = ImmutableWebClientConfig.builder() //
1183             .keyStoreType(config.keyStoreType()) //
1184             .keyStorePassword(config.keyStorePassword()) //
1185             .keyStore(config.keyStore()) //
1186             .keyPassword(config.keyPassword()) //
1187             .isTrustStoreUsed(useTrustValidation) //
1188             .trustStore(config.trustStore()) //
1189             .trustStorePassword(config.trustStorePassword()) //
1190             .httpProxyConfig(httpProxyConfig).build();
1191
1192         AsyncRestClientFactory restClientFactory = new AsyncRestClientFactory(config);
1193         return restClientFactory.createRestClientNoHttpProxy(baseUrl());
1194     }
1195
1196     private AsyncRestClient restClient() {
1197         return restClient(false);
1198     }
1199
1200     private void testErrorCode(Mono<?> request, HttpStatus expStatus, String responseContains) {
1201         testErrorCode(request, expStatus, responseContains, true);
1202     }
1203
1204     private void testErrorCode(Mono<?> request, HttpStatus expStatus, String responseContains,
1205         boolean expectApplicationProblemJsonMediaType) {
1206         StepVerifier.create(request) //
1207             .expectSubscription() //
1208             .expectErrorMatches(
1209                 t -> checkWebClientError(t, expStatus, responseContains, expectApplicationProblemJsonMediaType)) //
1210             .verify();
1211     }
1212
1213     private boolean checkWebClientError(Throwable throwable, HttpStatus expStatus, String responseContains,
1214         boolean expectApplicationProblemJsonMediaType) {
1215         assertTrue(throwable instanceof WebClientResponseException);
1216         WebClientResponseException responseException = (WebClientResponseException) throwable;
1217         assertThat(responseException.getStatusCode()).isEqualTo(expStatus);
1218         assertThat(responseException.getResponseBodyAsString()).contains(responseContains);
1219         if (expectApplicationProblemJsonMediaType) {
1220             assertThat(responseException.getHeaders().getContentType()).isEqualTo(MediaType.APPLICATION_PROBLEM_JSON);
1221         }
1222         return true;
1223     }
1224
1225 }