package com.atlassian.bitbucket.mesh.grpc;

import com.atlassian.bitbucket.mesh.management.NodeManager;
import com.atlassian.bitbucket.mesh.rpc.v1.RpcError;
import com.atlassian.bitbucket.mesh.rpc.v1.RpcErrorUtils;
import io.grpc.*;
import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.mockito.ArgumentCaptor;
import org.mockito.Mock;
import org.mockito.junit.MockitoJUnitRunner;

import java.time.Duration;

import static com.atlassian.bitbucket.mesh.grpc.RequireConfiguredServerInterceptor.ALWAYS_ALLOWED;
import static org.hamcrest.MatcherAssert.assertThat;
import static org.hamcrest.Matchers.containsString;
import static org.junit.Assert.*;
import static org.mockito.Mockito.*;

@RunWith(MockitoJUnitRunner.class)
public class RequireConfiguredServerInterceptorTest {

    private static final Duration TIMEOUT = Duration.ofMinutes(1L);

    @Mock
    private ServerCall<Object, Object> call;
    @Mock
    private MethodDescriptor<Object, Object> descriptor;
    @Mock
    private ServerCallHandler<Object, Object> handler;
    private RequireConfiguredServerInterceptor interceptor;
    @Mock
    private ServerCall.Listener<Object> listener;
    @Mock
    private NodeManager nodeManager;

    @Before
    public void setup() {
        when(call.getMethodDescriptor()).thenReturn(descriptor);
        when(handler.startCall(same(call), any())).thenReturn(listener);

        interceptor = new RequireConfiguredServerInterceptor(nodeManager, TIMEOUT);
    }

    @Test
    public void testInterceptCall() throws InterruptedException {
        when(nodeManager.awaitConfiguration(same(TIMEOUT))).thenReturn(true);

        assertSame(listener, interceptor.interceptCall(call, null, handler));

        verify(nodeManager).awaitConfiguration(same(TIMEOUT));
        verify(nodeManager).isConfigured();
    }

    @Test
    public void testInterceptCallAlreadyConfigured() {
        when(nodeManager.isConfigured()).thenReturn(true);

        assertSame(listener, interceptor.interceptCall(call, null, handler));

        verify(nodeManager).isConfigured();
        verifyNoInteractions(descriptor); //Should not need to check the descriptor against the allowed list
    }

    @Test
    public void testInterceptCallAlwaysAllowsCertainRequests() {
        for (String allowedMethod : ALWAYS_ALLOWED) {
            when(descriptor.getFullMethodName()).thenReturn(allowedMethod);

            //nodeManager.isConfigured is not trained, so it will return false. The only way the request
            //is allowed is if the method name is in the allowed list.
            assertSame(listener, interceptor.interceptCall(call, null, handler));
        }

        verify(nodeManager, times(ALWAYS_ALLOWED.size())).isConfigured();
    }

    @Test
    public void testInterceptCallRejectsIfInterrupted() throws InterruptedException {
        doThrow(InterruptedException.class).when(nodeManager).awaitConfiguration(same(TIMEOUT));

        assertNotSame(listener, interceptor.interceptCall(call, null, handler)); //Should return a no-op listener

        ArgumentCaptor<Metadata> metadataCaptor = ArgumentCaptor.forClass(Metadata.class);
        ArgumentCaptor<Status> statusCaptor = ArgumentCaptor.forClass(Status.class);

        verify(nodeManager).awaitConfiguration(same(TIMEOUT));
        verify(call).close(statusCaptor.capture(), metadataCaptor.capture());
        verifyNoInteractions(handler); //No call should be started

        Status status = statusCaptor.getValue();
        assertSame(Status.Code.CANCELLED, status.getCode());
        assertThat(status.getDescription(), containsString("Interrupted while awaiting configuration"));
        assertUnconfigured(metadataCaptor.getValue());
    }

    @Test
    public void testInterceptCallRejectsIfUnconfigured() throws InterruptedException {
        assertNotSame(listener, interceptor.interceptCall(call, null, handler)); //Should return a no-op listener

        ArgumentCaptor<Metadata> metadataCaptor = ArgumentCaptor.forClass(Metadata.class);
        ArgumentCaptor<Status> statusCaptor = ArgumentCaptor.forClass(Status.class);

        verify(nodeManager).awaitConfiguration(same(TIMEOUT));
        verify(call).close(statusCaptor.capture(), metadataCaptor.capture());
        verifyNoInteractions(handler); //No call should be started

        Status status = statusCaptor.getValue();
        assertSame(Status.Code.UNAVAILABLE, status.getCode());
        assertThat(status.getDescription(), containsString("has not been configured"));
        assertUnconfigured(metadataCaptor.getValue());
    }

    private static void assertUnconfigured(Metadata metadata) {
        RpcError error = RpcErrorUtils.fromMetadata(metadata);
        assertNotNull(error);
        assertSame(RpcError.ErrorOneofCase.NODE_UNCONFIGURED, error.getErrorOneofCase());
    }
}