add more tests

Signed-off-by: Jackie <[email protected]>

Author: jackiehanyang

src/test/java/org/opensearch/ad/InsightsJobProcessorTests.java

@@ -181,6 +181,56 @@ public void setUp() throws Exception {
         }).when(client).execute(any(), any(MLExecuteTaskRequest.class), any());
     }
 
+    @Test
+    public void testSkipCorrelationWhenNoDetectorIdsInAnomalies() throws IOException {
+        // An anomaly without detector_id
+        String anomalyJson = TestHelpers
+            .builder()
+            .startObject()
+            .field("anomaly_grade", 0.7)
+            .field("anomaly_score", 2.3)
+            .field("data_start_time", Instant.now().minus(30, ChronoUnit.MINUTES).toEpochMilli())
+            .field("data_end_time", Instant.now().minus(29, ChronoUnit.MINUTES).toEpochMilli())
+            .endObject()
+            .toString();
+
+        SearchHit anomalyHit = new SearchHit(1);
+        anomalyHit.sourceRef(new BytesArray(anomalyJson));
+        anomalyHit.score(1.0f);
+        anomalyHit.shard(new SearchShardTarget("node", new ShardId("test", "uuid", 0), null, null));
+
+        SearchHits anomalySearchHits = new SearchHits(new SearchHit[] { anomalyHit }, new TotalHits(1, TotalHits.Relation.EQUAL_TO), 1.0f);
+
+        // Return anomalies from results index; we won't reach detector config enrichment
+        doAnswer(invocation -> {
+            ActionListener<SearchResponse> listener = invocation.getArgument(1);
+            SearchResponse searchResponse = mock(SearchResponse.class);
+            when(searchResponse.getHits()).thenReturn(anomalySearchHits);
+            listener.onResponse(searchResponse);
+            return null;
+        }).when(client).search(any(SearchRequest.class), any());
+
+        // Lock lifecycle
+        doAnswer(invocation -> {
+            ActionListener<LockModel> listener = invocation.getArgument(2);
+            listener.onResponse(lockModel);
+            return null;
+        }).when(lockService).acquireLock(any(), any(), any());
+        doAnswer(invocation -> {
+            ActionListener<Boolean> listener = invocation.getArgument(1);
+            listener.onResponse(true);
+            return null;
+        }).when(lockService).release(any(), any());
+
+        insightsJobProcessor.process(insightsJob, jobExecutionContext);
+
+        // One search for anomalies, then skip correlation and release lock
+        verify(client, times(1)).search(any(SearchRequest.class), any());
+        verify(lockService, times(1)).release(any(), any());
+        // No index write occurs
+        verify(client, never()).index(any(IndexRequest.class), any());
+    }
+
     @Test
     public void testProcessWithIntervalSchedule() {
         // Mock lock acquisition

src/test/java/org/opensearch/ad/ml/MLCommonsClientTests.java

@@ -430,6 +430,28 @@ public void testExecutionWithVariableTimestamps() {
         verify(listener, times(1)).onResponse(any(MLMetricsCorrelationOutput.class));
     }
 
+    public void testUnexpectedErrorDuringConversionGracefullyHandled() {
+        // Matrix containing a null value to trigger NPE during float conversion
+        List<List<Double>> matrix = Arrays.asList(Arrays.asList(1.0, null, 3.0));
+
+        MLMetricsCorrelationInput input = new MLMetricsCorrelationInput(
+            matrix,
+            Arrays.asList("m1"),
+            Collections.emptyMap(),
+            Instant.now(),
+            Instant.now(),
+            60000L,
+            Collections.emptyList()
+        );
+
+        @SuppressWarnings("unchecked")
+        ActionListener<MLMetricsCorrelationOutput> listener = mock(ActionListener.class);
+
+        // Should catch exception and degrade to empty output
+        mlCommonsClient.executeMetricsCorrelation(input, listener);
+        verify(listener, times(1)).onResponse(any(MLMetricsCorrelationOutput.class));
+    }
+
     public void testClientReuseability() {
         // Test that the same client can be used multiple times
         List<List<Double>> matrix = Arrays.asList(Arrays.asList(1.0));