+ @Test
+ void producerSupervision2() throws JsonMappingException, JsonProcessingException, ServiceException {
+ // Test that supervision enables not enabled jobs and sends a notification when
+ // suceeded
+
+ putEiProducerWithOneType(PRODUCER_ID, TYPE_ID);
+ putEiJob(TYPE_ID, EI_JOB_ID);
+
+ EiProducer producer = this.eiProducers.getProducer(PRODUCER_ID);
+ EiJob job = this.eiJobs.getJob(EI_JOB_ID);
+ // Pretend that the producer did reject the job and the a DISABLED notification
+ // is sent for the job
+ producer.setJobDisabled(job);
+ job.setLastReportedStatus(false);
+ verifyJobStatus(EI_JOB_ID, "DISABLED");
+
+ // Run the supervision and wait for the job to get started in the producer
+ this.producerSupervision.createTask().blockLast();
+ ConsumerSimulatorController.TestResults consumerResults = this.consumerSimulator.getTestResults();
+ await().untilAsserted(() -> assertThat(consumerResults.status.size()).isEqualTo(1));
+ assertThat(consumerResults.status.get(0).state).isEqualTo(A1eEiJobStatus.EiJobStatusValues.ENABLED);
+ verifyJobStatus(EI_JOB_ID, "ENABLED");
+ }
+
+ @Test
+ void testGetStatus() throws JsonMappingException, JsonProcessingException, ServiceException {
+ putEiProducerWithOneTypeRejecting("simulateProducerError", TYPE_ID);
+ putEiProducerWithOneTypeRejecting("simulateProducerError2", TYPE_ID);
+
+ String url = "/status";
+ ResponseEntity<String> resp = restClient().getForEntity(url).block();
+ assertThat(resp.getBody()).contains("hunky dory");
+ }
+
+ @Test
+ void testEiJobDatabase() throws Exception {
+ putEiProducerWithOneType(PRODUCER_ID, TYPE_ID);
+ putEiJob(TYPE_ID, "jobId1");
+ putEiJob(TYPE_ID, "jobId2");
+
+ assertThat(this.eiJobs.size()).isEqualTo(2);
+
+ {
+ EiJob savedJob = this.eiJobs.getJob("jobId1");
+ // Restore the jobs
+ EiJobs jobs = new EiJobs(this.applicationConfig, this.producerCallbacks);
+ jobs.restoreJobsFromDatabase();
+ assertThat(jobs.size()).isEqualTo(2);
+ EiJob restoredJob = jobs.getJob("jobId1");
+ assertThat(restoredJob.getId()).isEqualTo("jobId1");
+ assertThat(restoredJob.getLastUpdated()).isEqualTo(savedJob.getLastUpdated());
+
+ jobs.remove("jobId1", this.eiProducers);
+ jobs.remove("jobId2", this.eiProducers);
+ }
+ {
+ // Restore the jobs, no jobs in database
+ EiJobs jobs = new EiJobs(this.applicationConfig, this.producerCallbacks);
+ jobs.restoreJobsFromDatabase();
+ assertThat(jobs.size()).isEqualTo(0);
+ }
+ logger.warn("Test removing a job when the db file is gone");
+ this.eiJobs.remove("jobId1", this.eiProducers);
+ assertThat(this.eiJobs.size()).isEqualTo(1);
+
+ ProducerSimulatorController.TestResults simulatorResults = this.producerSimulator.getTestResults();
+ await().untilAsserted(() -> assertThat(simulatorResults.jobsStopped.size()).isEqualTo(3));
+ }
+
+ @Test
+ void testEiTypesDatabase() throws Exception {
+ putEiProducerWithOneType(PRODUCER_ID, TYPE_ID);
+
+ assertThat(this.eiTypes.size()).isEqualTo(1);
+
+ {
+ // Restore the types
+ EiTypes types = new EiTypes(this.applicationConfig);
+ types.restoreTypesFromDatabase();
+ assertThat(types.size()).isEqualTo(1);
+
+ }
+ {
+ // Restore the jobs, no jobs in database
+ EiTypes types = new EiTypes(this.applicationConfig);
+ types.clear();
+ types.restoreTypesFromDatabase();
+ assertThat(types.size()).isEqualTo(0);
+ }
+ logger.warn("Test removing a job when the db file is gone");
+ this.eiTypes.remove(this.eiTypes.getType(TYPE_ID));
+ assertThat(this.eiJobs.size()).isEqualTo(0);
+ }
+
+ private void deleteEiProducer(String eiProducerId) {
+ String url = ProducerConsts.API_ROOT + "/eiproducers/" + eiProducerId;
+ restClient().deleteForEntity(url).block();
+ }
+
+ private void verifyJobStatus(String jobId, String expStatus) {
+ String url = A1eConsts.API_ROOT + "/eijobs/" + jobId + "/status";
+ String rsp = restClient().get(url).block();
+ assertThat(rsp).contains(expStatus);
+ }
+
+ private void assertProducerOpState(String producerId,
+ ProducerStatusInfo.OperationalState expectedOperationalState) {
+ String statusUrl = ProducerConsts.API_ROOT + "/eiproducers/" + producerId + "/status";
+ ResponseEntity<String> resp = restClient().getForEntity(statusUrl).block();
+ ProducerStatusInfo statusInfo = gson.fromJson(resp.getBody(), ProducerStatusInfo.class);
+ assertThat(statusInfo.opState).isEqualTo(expectedOperationalState);
+ }
+
+ ProducerEiTypeInfo producerEiTypeRegistrationInfo(String typeId)
+ throws JsonMappingException, JsonProcessingException {
+ return new ProducerEiTypeInfo(jsonSchemaObject());
+ }
+
+ ProducerRegistrationInfo producerEiRegistratioInfoRejecting(String typeId)
+ throws JsonMappingException, JsonProcessingException {
+ return new ProducerRegistrationInfo(Arrays.asList(typeId), //
+ baseUrl() + ProducerSimulatorController.JOB_ERROR_URL,
+ baseUrl() + ProducerSimulatorController.SUPERVISION_ERROR_URL);
+ }
+
+ ProducerRegistrationInfo producerEiRegistratioInfo(String typeId)
+ throws JsonMappingException, JsonProcessingException {
+ return new ProducerRegistrationInfo(Arrays.asList(typeId), //
+ baseUrl() + ProducerSimulatorController.JOB_URL, baseUrl() + ProducerSimulatorController.SUPERVISION_URL);
+ }
+
+ private ConsumerJobInfo consumerJobInfo() throws JsonMappingException, JsonProcessingException {
+ return consumerJobInfo(TYPE_ID, EI_JOB_ID);
+ }
+
+ ConsumerJobInfo consumerJobInfo(String typeId, String eiJobId)
+ throws JsonMappingException, JsonProcessingException {
+ return new ConsumerJobInfo(typeId, jsonObject(), "owner", "targetUri",
+ baseUrl() + ConsumerSimulatorController.getJobStatusUrl(eiJobId));
+ }
+
+ private A1eEiJobInfo eiJobInfo() throws JsonMappingException, JsonProcessingException {
+ return eiJobInfo(TYPE_ID, EI_JOB_ID);
+ }
+
+ A1eEiJobInfo eiJobInfo(String typeId, String eiJobId) throws JsonMappingException, JsonProcessingException {
+ return new A1eEiJobInfo(typeId, jsonObject(), "owner", "targetUri",
+ baseUrl() + ConsumerSimulatorController.getJobStatusUrl(eiJobId));
+ }
+
+ private Object jsonObject(String json) {
+ try {
+ return JsonParser.parseString(json).getAsJsonObject();
+ } catch (Exception e) {
+ throw new NullPointerException(e.toString());
+ }
+ }
+
+ private Object jsonSchemaObject() {
+ // a json schema with one mandatory property named "string"
+ String schemaStr = "{" //
+ + "\"$schema\": \"http://json-schema.org/draft-04/schema#\"," //
+ + "\"type\": \"object\"," //
+ + "\"properties\": {" //
+ + EI_JOB_PROPERTY + " : {" //
+ + " \"type\": \"string\"" //
+ + " }" //
+ + "}," //
+ + "\"required\": [" //
+ + EI_JOB_PROPERTY //
+ + "]" //
+ + "}"; //
+ return jsonObject(schemaStr);
+ }
+
+ private Object jsonObject() {
+ return jsonObject("{ " + EI_JOB_PROPERTY + " : \"value\" }");
+ }
+
+ private EiJob putEiJob(String eiTypeId, String jobId)
+ throws JsonMappingException, JsonProcessingException, ServiceException {
+
+ String url = A1eConsts.API_ROOT + "/eijobs/" + jobId;
+ String body = gson.toJson(eiJobInfo(eiTypeId, jobId));
+ restClient().putForEntity(url, body).block();
+
+ return this.eiJobs.getJob(jobId);
+ }
+
+ private HttpStatus putEiType(String eiTypeId)
+ throws JsonMappingException, JsonProcessingException, ServiceException {
+ String url = ProducerConsts.API_ROOT + "/eitypes/" + eiTypeId;
+ String body = gson.toJson(producerEiTypeRegistrationInfo(eiTypeId));
+
+ ResponseEntity<String> resp = restClient().putForEntity(url, body).block();
+ this.eiTypes.getType(eiTypeId);
+ return resp.getStatusCode();
+
+ }
+
+ private EiType putEiProducerWithOneTypeRejecting(String producerId, String eiTypeId)
+ throws JsonMappingException, JsonProcessingException, ServiceException {
+ this.putEiType(eiTypeId);
+ String url = ProducerConsts.API_ROOT + "/eiproducers/" + producerId;
+ String body = gson.toJson(producerEiRegistratioInfoRejecting(eiTypeId));
+ restClient().putForEntity(url, body).block();
+ return this.eiTypes.getType(eiTypeId);
+ }
+
+ private EiType putEiProducerWithOneType(String producerId, String eiTypeId)
+ throws JsonMappingException, JsonProcessingException, ServiceException {
+ this.putEiType(eiTypeId);
+
+ String url = ProducerConsts.API_ROOT + "/eiproducers/" + producerId;
+ String body = gson.toJson(producerEiRegistratioInfo(eiTypeId));
+
+ restClient().putForEntity(url, body).block();
+
+ return this.eiTypes.getType(eiTypeId);