package com.atlassian.bitbucket.mesh.grpc;

import com.atlassian.bitbucket.mesh.config.NodeConfig;
import com.atlassian.bitbucket.mesh.junit.LogListener;
import com.atlassian.bitbucket.mesh.rpc.v1.*;
import io.grpc.*;
import org.junit.Before;
import org.junit.Rule;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.mockito.ArgumentCaptor;
import org.mockito.InjectMocks;
import org.mockito.Mock;
import org.mockito.junit.MockitoJUnitRunner;

import static com.atlassian.bitbucket.mesh.rpc.MetadataKeys.KEY_REMOTE_CLIENT;
import static com.atlassian.bitbucket.mesh.rpc.MetadataKeys.KEY_TARGET_ID;
import static org.junit.Assert.*;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.Mockito.*;

@RunWith(MockitoJUnitRunner.class)
public class WrongTargetServerInterceptorTest {

    private static final String NODE_ID = "real-node-id";

    @Rule
    public final LogListener logListener = new LogListener(WrongTargetServerInterceptor.class);

    @Mock
    private ServerCall<RpcGetCommitRequest, RpcGetCommitResponse> call;
    @Mock
    private ServerCallHandler<RpcGetCommitRequest, RpcGetCommitResponse> handler;
    @InjectMocks
    private WrongTargetServerInterceptor interceptor;
    @Mock
    private ServerCall.Listener<RpcGetCommitRequest> listener;
    @Mock
    private MethodDescriptor<RpcGetCommitRequest, RpcGetCommitResponse> methodDescriptor;
    @Mock
    private NodeConfig nodeConfig;

    @Before
    public void setUp() {
        when(call.getMethodDescriptor()).thenReturn(methodDescriptor);
        when(methodDescriptor.getFullMethodName()).thenReturn("CommitService/GetCommit");
        when(nodeConfig.getId()).thenReturn(NODE_ID);
        when(handler.startCall(any(), any())).thenReturn(listener);
    }

    @Test
    public void testAcceptsRequestIfNoTargetIdProvided() {
        Metadata metadata = new Metadata();
        ServerCall.Listener<RpcGetCommitRequest> actualListener = interceptor.interceptCall(call, metadata, handler);

        assertSame(listener, actualListener);
        verify(call, never()).close(any(), any());
    }

    @Test
    public void testAcceptsRequestIfNodeUnconfigured() {
        when(nodeConfig.getId()).thenReturn(NodeConfig.UNCONFIGURED_NODE_ID);

        Metadata metadata = new Metadata();
        metadata.put(KEY_TARGET_ID, "new-node-id");
        ServerCall.Listener<RpcGetCommitRequest> actualListener = interceptor.interceptCall(call, metadata, handler);

        assertSame(listener, actualListener);
        verify(call, never()).close(any(), any());
    }

    @Test
    public void testAcceptsRequestIfTargetIdMatches() {
        Metadata metadata = new Metadata();
        metadata.put(KEY_TARGET_ID, NODE_ID);
        ServerCall.Listener<RpcGetCommitRequest> actualListener = interceptor.interceptCall(call, metadata, handler);

        assertSame(listener, actualListener);
        verify(call, never()).close(any(), any());
    }

    @Test
    public void testRejectsRequestIfTargetIdDoesNotMatch() {
        Metadata metadata = new Metadata();
        metadata.put(KEY_TARGET_ID, "other-node-id");
        metadata.put(KEY_REMOTE_CLIENT, "remote-client");
        ServerCall.Listener<RpcGetCommitRequest> actualListener = interceptor.interceptCall(call, metadata, handler);

        assertNotSame(listener, actualListener);
        ArgumentCaptor<Status> statusCaptor = ArgumentCaptor.forClass(Status.class);
        ArgumentCaptor<Metadata> trailerCaptor = ArgumentCaptor.forClass(Metadata.class);
        verify(call).close(statusCaptor.capture(), trailerCaptor.capture());
        Status status = statusCaptor.getValue();
        assertEquals(Status.Code.FAILED_PRECONDITION, status.getCode());
        RpcError error = RpcErrorUtils.fromMetadata(trailerCaptor.getValue());
        assertNotNull(error);
        assertTrue(error.hasWrongTarget());
        RpcWrongTarget wrongTarget = error.getWrongTarget();
        assertEquals(NODE_ID, wrongTarget.getActualId());
        assertEquals("other-node-id", wrongTarget.getTargetId());

        logListener.assertWarn("Rejecting CommitService/GetCommit call from remote-client intended for " +
                "node with ID other-node-id");
    }
}
