/*
 * 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.watcher.transport.actions.put;

import org.elasticsearch.action.ActionListener;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.index.IndexResponse;
import org.elasticsearch.action.support.ActionFilters;
import org.elasticsearch.client.Client;
import org.elasticsearch.cluster.metadata.IndexNameExpressionResolver;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.util.concurrent.ThreadContext;
import org.elasticsearch.index.Index;
import org.elasticsearch.index.shard.ShardId;
import org.elasticsearch.license.XPackLicenseState;
import org.elasticsearch.test.ESTestCase;
import org.elasticsearch.threadpool.ThreadPool;
import org.elasticsearch.transport.TransportService;
import org.elasticsearch.xpack.watcher.Watcher;
import org.elasticsearch.xpack.watcher.test.WatchExecutionContextMockBuilder;
import org.elasticsearch.xpack.watcher.watch.Watch;
import org.elasticsearch.xpack.watcher.watch.clock.ClockMock;
import org.junit.Before;
import org.mockito.ArgumentCaptor;

import java.util.Collections;
import java.util.Map;

import static org.hamcrest.Matchers.hasKey;
import static org.hamcrest.Matchers.hasSize;
import static org.hamcrest.Matchers.is;
import static org.hamcrest.Matchers.nullValue;
import static org.mockito.Matchers.any;
import static org.mockito.Matchers.anyObject;
import static org.mockito.Matchers.eq;
import static org.mockito.Mockito.doAnswer;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;

public class TransportPutWatchActionTests extends ESTestCase {

    private TransportPutWatchAction action;
    private Watch watch = new WatchExecutionContextMockBuilder("_id").buildMock().watch();
    private ThreadContext threadContext = new ThreadContext(Settings.EMPTY);

    @Before
    public void setupAction() throws Exception {
        ThreadPool threadPool = mock(ThreadPool.class);
        when(threadPool.getThreadContext()).thenReturn(threadContext);

        TransportService transportService = mock(TransportService.class);

        Watch.Parser parser = mock(Watch.Parser.class);
        when(parser.parseWithSecrets(eq("_id"), eq(false), anyObject(), anyObject(), anyObject())).thenReturn(watch);

        Client client = mock(Client.class);
        when(client.threadPool()).thenReturn(threadPool);
        // mock an index response that calls the listener
        doAnswer(invocation -> {
            IndexRequest request = (IndexRequest) invocation.getArguments()[1];
            ActionListener<IndexResponse> listener = (ActionListener) invocation.getArguments()[2];

            ShardId shardId = new ShardId(new Index(Watch.INDEX, "uuid"), 0);
            listener.onResponse(new IndexResponse(shardId, request.type(), request.id(), 1, 1, 1, true));

            return null;
        }).when(client).execute(any(), any(), any());

        action = new TransportPutWatchAction(Settings.EMPTY, transportService, threadPool,
                new ActionFilters(Collections.emptySet()), new IndexNameExpressionResolver(Settings.EMPTY), new ClockMock(),
                new XPackLicenseState(), parser, client);
    }

    public void testHeadersAreFilteredWhenPuttingWatches() throws Exception {
        // set up threadcontext with some arbitrary info
        String headerName = randomFrom(Watcher.HEADER_FILTERS);
        threadContext.putHeader(headerName, randomAlphaOfLength(10));
        threadContext.putHeader(randomAlphaOfLength(10), "doesntmatter");

        PutWatchRequest putWatchRequest = new PutWatchRequest();
        putWatchRequest.setId("_id");
        action.doExecute(putWatchRequest, ActionListener.wrap(r -> {}, e -> assertThat(e, is(nullValue()))));

        ArgumentCaptor<Map> captor = ArgumentCaptor.forClass(Map.class);
        verify(watch.status()).setHeaders(captor.capture());
        Map<String, String> capturedHeaders = captor.getValue();
        assertThat(capturedHeaders.keySet(), hasSize(1));
        assertThat(capturedHeaders, hasKey(headerName));
    }
}