+ assertThat(this.eiProducers.getProducerIdsForType(TYPE_ID)).doesNotContain("eiProducerId");
+ verifyJobStatus("jobId", "ENABLED");
+
+ deleteEiProducer("eiProducerId2");
+ assertThat(this.eiProducers.size()).isZero();
+ assertThat(this.eiTypes.size()).isEqualTo(1);
+ verifyJobStatus("jobId", "DISABLED");
+ }
+
+ @Test
+ void a1eJobStatusNotifications() throws JsonMappingException, JsonProcessingException, ServiceException {
+ ConsumerSimulatorController.TestResults consumerCalls = this.consumerSimulator.getTestResults();
+ ProducerSimulatorController.TestResults producerCalls = this.producerSimulator.getTestResults();
+
+ putEiProducerWithOneType("eiProducerId", TYPE_ID);
+ putEiJob(TYPE_ID, "jobId");
+ putEiProducerWithOneType("eiProducerId2", TYPE_ID);
+ await().untilAsserted(() -> assertThat(producerCalls.jobsStarted.size()).isEqualTo(2));
+
+ deleteEiProducer("eiProducerId2");
+ assertThat(this.eiTypes.size()).isEqualTo(1); // The type remains, one producer left
+ deleteEiProducer("eiProducerId");
+ assertThat(this.eiTypes.size()).isEqualTo(1); // The type remains
+ assertThat(this.eiJobs.size()).isEqualTo(1); // The job remains
+ await().untilAsserted(() -> assertThat(consumerCalls.status.size()).isEqualTo(1));
+ assertThat(consumerCalls.status.get(0).state).isEqualTo(A1eEiJobStatus.EiJobStatusValues.DISABLED);
+
+ putEiProducerWithOneType("eiProducerId", TYPE_ID);
+ await().untilAsserted(() -> assertThat(consumerCalls.status.size()).isEqualTo(2));
+ assertThat(consumerCalls.status.get(1).state).isEqualTo(A1eEiJobStatus.EiJobStatusValues.ENABLED);
+ }
+
+ @Test
+ void a1eJobStatusNotifications2() throws JsonMappingException, JsonProcessingException, ServiceException {
+ // Test replacing a producer with new and removed types
+
+ // Create a job
+ putEiProducerWithOneType(PRODUCER_ID, TYPE_ID);
+ putEiJob(TYPE_ID, EI_JOB_ID);
+
+ // change the type for the producer, the job shall be disabled
+ putEiProducerWithOneType(PRODUCER_ID, "junk");
+ verifyJobStatus(EI_JOB_ID, "DISABLED");
+ ConsumerSimulatorController.TestResults consumerCalls = this.consumerSimulator.getTestResults();
+ await().untilAsserted(() -> assertThat(consumerCalls.status.size()).isEqualTo(1));
+ assertThat(consumerCalls.status.get(0).state).isEqualTo(A1eEiJobStatus.EiJobStatusValues.DISABLED);
+
+ putEiProducerWithOneType(PRODUCER_ID, TYPE_ID);
+ verifyJobStatus(EI_JOB_ID, "ENABLED");
+ await().untilAsserted(() -> assertThat(consumerCalls.status.size()).isEqualTo(2));
+ assertThat(consumerCalls.status.get(1).state).isEqualTo(A1eEiJobStatus.EiJobStatusValues.ENABLED);
+ }
+
+ @Test
+ void producerGetProducerEiType() throws JsonMappingException, JsonProcessingException, ServiceException {
+ putEiProducerWithOneType(PRODUCER_ID, TYPE_ID);
+ String url = ProducerConsts.API_ROOT + "/eitypes/" + TYPE_ID;
+ ResponseEntity<String> resp = restClient().getForEntity(url).block();
+ ProducerEiTypeInfo info = gson.fromJson(resp.getBody(), ProducerEiTypeInfo.class);
+ assertThat(info.jobDataSchema).isNotNull();
+ }
+
+ @Test
+ void producerGetProducerIdentifiers() throws JsonMappingException, JsonProcessingException, ServiceException {
+ putEiProducerWithOneType(PRODUCER_ID, TYPE_ID);
+ String url = ProducerConsts.API_ROOT + "/eiproducers";
+ ResponseEntity<String> resp = restClient().getForEntity(url).block();
+ assertThat(resp.getBody()).contains(PRODUCER_ID);
+
+ url = ProducerConsts.API_ROOT + "/eiproducers?ei_type_id=" + TYPE_ID;
+ resp = restClient().getForEntity(url).block();
+ assertThat(resp.getBody()).contains(PRODUCER_ID);
+
+ url = ProducerConsts.API_ROOT + "/eiproducers?ei_type_id=junk";
+ resp = restClient().getForEntity(url).block();
+ assertThat(resp.getBody()).isEqualTo("[]");
+ }
+
+ @Test
+ void producerSupervision() throws JsonMappingException, JsonProcessingException, ServiceException {
+
+ ConsumerSimulatorController.TestResults consumerResults = this.consumerSimulator.getTestResults();
+ putEiProducerWithOneTypeRejecting("simulateProducerError", TYPE_ID);
+
+ {
+ // Create a job
+ putEiProducerWithOneType(PRODUCER_ID, TYPE_ID);
+ putEiJob(TYPE_ID, EI_JOB_ID);
+ verifyJobStatus(EI_JOB_ID, "ENABLED");
+ deleteEiProducer(PRODUCER_ID);
+ // A Job disabled status notification shall now be received
+ await().untilAsserted(() -> assertThat(consumerResults.status.size()).isEqualTo(1));
+ assertThat(consumerResults.status.get(0).state).isEqualTo(A1eEiJobStatus.EiJobStatusValues.DISABLED);
+ verifyJobStatus(EI_JOB_ID, "DISABLED");
+ }
+
+ assertThat(this.eiProducers.size()).isEqualTo(1);
+ assertThat(this.eiTypes.size()).isEqualTo(1);
+ assertProducerOpState("simulateProducerError", ProducerStatusInfo.OperationalState.ENABLED);
+
+ this.producerSupervision.createTask().blockLast();
+ this.producerSupervision.createTask().blockLast();
+
+ // Now we have one producer that is disabled
+ assertThat(this.eiProducers.size()).isEqualTo(1);
+ assertProducerOpState("simulateProducerError", ProducerStatusInfo.OperationalState.DISABLED);
+
+ // After 3 failed checks, the producer shall be deregisterred
+ this.producerSupervision.createTask().blockLast();
+ assertThat(this.eiProducers.size()).isEqualTo(0); // The producer is removed
+ assertThat(this.eiTypes.size()).isEqualTo(1); // The type remains
+
+ // Now we have one disabled job, and no producer.
+ // PUT a producer, then a Job ENABLED status notification shall be received
+ putEiProducerWithOneType(PRODUCER_ID, TYPE_ID);
+ await().untilAsserted(() -> assertThat(consumerResults.status.size()).isEqualTo(2));
+ assertThat(consumerResults.status.get(1).state).isEqualTo(A1eEiJobStatus.EiJobStatusValues.ENABLED);
+ verifyJobStatus(EI_JOB_ID, "ENABLED");
+ }
+
+ @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);