package com.atlassian.stash.internal.maintenance;

import com.atlassian.bitbucket.cluster.ClusterNode;
import com.atlassian.bitbucket.event.cluster.ClusterNodeAddedEvent;
import com.google.common.collect.ImmutableSet;
import com.hazelcast.core.ITopic;
import com.hazelcast.core.Member;
import com.hazelcast.core.Message;
import com.hazelcast.core.MessageListener;
import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.mockito.ArgumentCaptor;
import org.mockito.InOrder;
import org.mockito.Mock;
import org.mockito.runners.MockitoJUnitRunner;

import static org.mockito.Matchers.same;
import static org.mockito.Mockito.*;

@RunWith(MockitoJUnitRunner.class)
public class ClusteredMaintenanceTaskStatusSupplierTest {

    public static final String ADDED_NODE_ID = "added-node-id";
    public static final String LOCAL_NODE_ID = "local-node-id";

    @Mock
    private MaintenanceTaskStatusSupplier delegate;
    @Mock
    private MaintenanceTaskStatus status;
    @Mock
    private ITopic<MaintenanceTaskStatus> topic;
    @Mock
    private ClusterNode localNode;
    @Mock
    private ClusterNode addedNode;
    @Mock
    private Member member;
    private ClusteredMaintenanceTaskStatusSupplier holder;

    @Before
    public void setUp() throws Exception {
        holder = new ClusteredMaintenanceTaskStatusSupplier(delegate, topic);
        when(localNode.getId()).thenReturn(LOCAL_NODE_ID);
        when(localNode.isLocal()).thenReturn(true);
        when(addedNode.getId()).thenReturn(ADDED_NODE_ID);
    }

    @Test
    public void testOnNodeAddedWhenNoStatus() throws Exception {
        holder.onNodeAdded(newEvent());
        verifyZeroInteractions(topic);
    }

    @Test
    public void testOnNodeAddedWhenStatusOwnedByLocal() throws Exception {
        mockStatus(LOCAL_NODE_ID);
        holder.onNodeAdded(newEvent());
        verify(topic).publish(same(status));
    }

    @Test
    public void testOnNodeAddedWhenStatusOwnedByOtherNode() throws Exception {
        mockStatus(ADDED_NODE_ID);
        holder.onNodeAdded(newEvent());
        verifyZeroInteractions(topic);
    }

    @Test
    public void testOnNodeAddedWhenStatusOwnedByNeither() throws Exception {
        mockStatus("some-random-node-id");
        holder.onNodeAdded(newEvent());
        verify(topic).publish(same(status));
    }

    @SuppressWarnings("unchecked")
    @Test
    public void testListener() throws Exception {
        doReturn("listener-id").when(topic).addMessageListener(any(MessageListener.class));
        holder.addListener();

        ArgumentCaptor<MessageListener<MaintenanceTaskStatus>> captor = (ArgumentCaptor) ArgumentCaptor.forClass(MessageListener.class);
        verify(topic).addMessageListener(captor.capture());
        MessageListener<MaintenanceTaskStatus> listener = captor.getValue();

        // Remote node
        when(member.localMember()).thenReturn(false);
        listener.onMessage(new Message<>("topicName", status, 0L, member));
        verify(delegate).set(same(status));
        verifyZeroInteractions(delegate);

        // Local node
        when(member.localMember()).thenReturn(true);
        listener.onMessage(new Message<>("topicName", status, 0L, member));
        verifyZeroInteractions(delegate);

        holder.removeListener();
        verify(topic).removeMessageListener("listener-id");
    }

    @Test
    public void testRemoveListenerWhenNotRegistered() throws Exception {
        holder.removeListener();
        verifyZeroInteractions(topic);
    }

    @Test
    public void testSet() throws Exception {
        holder.set(status);
        InOrder ordered = inOrder(delegate, topic);
        ordered.verify(delegate).set(same(status));
        ordered.verify(topic).publish(same(status));
    }

    private void mockStatus(String ownerId) {
        when(status.getOwnerNodeId()).thenReturn(ownerId);
        when(holder.get()).thenReturn(status);
    }

    private ClusterNodeAddedEvent newEvent() {
        return new ClusterNodeAddedEvent(this, addedNode, ImmutableSet.of(addedNode, localNode));
    }
}