/*
 * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
 * or more contributor license agreements. Licensed under the Elastic License;
 * you may not use this file except in compliance with the Elastic License.
 */

package org.elasticsearch.xpack.dataframe.integration;

import org.elasticsearch.client.Request;
import org.elasticsearch.xpack.dataframe.persistence.DataFrameInternalIndex;
import org.junit.Before;

import java.io.IOException;
import java.util.Arrays;
import java.util.List;
import java.util.Map;

import static org.elasticsearch.xpack.core.security.authc.support.UsernamePasswordToken.basicAuthHeaderValue;
import static org.hamcrest.CoreMatchers.equalTo;
import static org.hamcrest.CoreMatchers.notNullValue;
import static org.hamcrest.Matchers.is;

public class DataFrameAuditorIT extends DataFrameRestTestCase {

    private static final String TEST_USER_NAME = "df_admin_plus_data";
    private static final String DATA_ACCESS_ROLE = "test_data_access";
    private static final String BASIC_AUTH_VALUE_DATA_FRAME_ADMIN_WITH_SOME_DATA_ACCESS =
        basicAuthHeaderValue(TEST_USER_NAME, TEST_PASSWORD_SECURE_STRING);

    private static boolean indicesCreated = false;

    // preserve indices in order to reuse source indices in several test cases
    @Override
    protected boolean preserveIndicesUponCompletion() {
        return true;
    }

    @Before
    public void createIndexes() throws IOException {

        // it's not possible to run it as @BeforeClass as clients aren't initialized then, so we need this little hack
        if (indicesCreated) {
            return;
        }

        createReviewsIndex();
        indicesCreated = true;
        setupDataAccessRole(DATA_ACCESS_ROLE, REVIEWS_INDEX_NAME);
        setupUser(TEST_USER_NAME, Arrays.asList("data_frame_transforms_admin", DATA_ACCESS_ROLE));
    }

    @AwaitsFix(bugUrl = "https://github.com/elastic/elasticsearch/issues/40594")
    @SuppressWarnings("unchecked")
    public void testAuditorWritesAudits() throws Exception {
        String transformId = "simplePivotForAudit";
        String dataFrameIndex = "pivot_reviews_user_id_above_20";
        setupDataAccessRole(DATA_ACCESS_ROLE, REVIEWS_INDEX_NAME, dataFrameIndex);
        String query = "\"match\": {\"user_id\": \"user_26\"}";

        createPivotReviewsTransform(transformId, dataFrameIndex, query, BASIC_AUTH_VALUE_DATA_FRAME_ADMIN_WITH_SOME_DATA_ACCESS);

        startAndWaitForTransform(transformId, dataFrameIndex, BASIC_AUTH_VALUE_DATA_FRAME_ADMIN_WITH_SOME_DATA_ACCESS);

        // Make sure we wrote to the audit
        assertTrue(indexExists(DataFrameInternalIndex.AUDIT_INDEX));
        refreshIndex(DataFrameInternalIndex.AUDIT_INDEX);
        Request request = new Request("GET", DataFrameInternalIndex.AUDIT_INDEX + "/_search");
        request.setJsonEntity("{\"query\":{\"term\":{\"transform_id\":\"simplePivotForAudit\"}}}");
        Map<String, Object> response = entityAsMap(client().performRequest(request));
        Map<?, ?> hitRsp = (Map<?, ?>) ((List<?>) ((Map<?, ?>)response.get("hits")).get("hits")).get(0);
        Map<String, Object> source = (Map<String, Object>)hitRsp.get("_source");
        assertThat(source.get("transform_id"), equalTo(transformId));
        assertThat(source.get("level"), equalTo("info"));
        assertThat(source.get("message"), is(notNullValue()));
        assertThat(source.get("node_name"), is(notNullValue()));
        assertThat(source.get("timestamp"), is(notNullValue()));
    }
}

@@ -737,25 +737,113 @@
 hers.not
-NullValue
+;%0Aimport static org.hamcrest.CoreMatchers.notNullValue;%0Aimport static org.hamcrest.Matchers.empty
 ;%0Aimport
@@ -1963,89 +1963,8 @@
  %7D%0A%0A
-    @AwaitsFix(bugUrl = %22https://github.com/elastic/elasticsearch/issues/40594%22)%0A
     
@@ -2628,134 +2628,13 @@
     
-assertTrue(indexExists(DataFrameInternalIndex.AUDIT_INDEX));%0A        refreshIndex(DataFrameInternalIndex.AUDIT_INDEX);%0A       
+final
  Req
@@ -2813,24 +2813,445 @@
 dit%5C%22%7D%7D%7D%22);%0A
+        assertBusy(() -%3E %7B%0A            assertTrue(indexExists(DataFrameInternalIndex.AUDIT_INDEX));%0A        %7D);%0A        // Since calls to write the Auditor are sent and forgot (async) we could have returned from the start,%0A        // finished the job (as this is a very short DF job), all without the audit being fully written.%0A        assertBusy(() -%3E %7B%0A            refreshIndex(DataFrameInternalIndex.AUDIT_INDEX);%0A    
         Map%3C
@@ -3336,38 +3336,29 @@
     
-Map%3C?, ?%3E hitRsp = (Map%3C?, ?%3E)
+    List%3C?%3E hitList =
  ((L
@@ -3407,24 +3407,127 @@
 get(%22hits%22))
+;%0A            assertThat(hitList, is(not(empty())));%0A            Map%3C?, ?%3E hitRsp = (Map%3C?, ?%3E) hitList
 .get(0);%0A   
@@ -3523,16 +3523,20 @@
 get(0);%0A
+    
         
@@ -3608,32 +3608,36 @@
 urce%22);%0A        
+    
 assertThat(sourc
@@ -3682,32 +3682,36 @@
 rmId));%0A        
+    
 assertThat(sourc
@@ -3744,32 +3744,36 @@
 nfo%22));%0A        
+   
+ 
 assertThat(sourc
@@ -3803,32 +3803,36 @@
 tNullValue()));%0A
+    
         assertTh
@@ -3880,32 +3880,36 @@
 ue()));%0A        
+ 
+   
 assertThat(sourc
@@ -3945,20 +3945,33 @@
 lValue()));%0A
+        %7D);%0A%0A
     %7D%0A%7D%0A

fix test, should account for async nature of audit (#40637)