/*
 * Copyright 1999-2018 Alibaba Group Holding Ltd.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package com.alibaba.nacos.config.server.service.notify;

import com.alibaba.nacos.api.config.remote.request.cluster.ConfigChangeClusterSyncRequest;
import com.alibaba.nacos.api.config.remote.response.cluster.ConfigChangeClusterSyncResponse;
import com.alibaba.nacos.api.exception.NacosException;
import com.alibaba.nacos.api.remote.RequestCallBack;
import com.alibaba.nacos.config.server.model.event.ConfigDataChangeEvent;
import com.alibaba.nacos.config.server.remote.ConfigClusterRpcClientProxy;
import com.alibaba.nacos.config.server.service.notify.AsyncNotifyService.AsyncRpcNotifyCallBack;
import com.alibaba.nacos.config.server.utils.ConfigExecutor;
import com.alibaba.nacos.core.cluster.Member;
import com.alibaba.nacos.api.common.NodeState;
import com.alibaba.nacos.core.cluster.ServerMemberManager;
import com.alibaba.nacos.sys.env.EnvUtil;
import com.alibaba.nacos.sys.utils.InetUtils;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.ExtendWith;
import org.mockito.Mock;
import org.mockito.MockedStatic;
import org.mockito.Mockito;
import org.springframework.test.context.junit.jupiter.SpringExtension;
import org.springframework.test.util.ReflectionTestUtils;

import java.util.ArrayList;
import java.util.LinkedList;
import java.util.List;
import java.util.Queue;
import java.util.concurrent.TimeUnit;

import static com.alibaba.nacos.config.server.service.notify.AsyncNotifyService.HEALTHY_CHECK_STATUS;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyLong;
import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.times;

/**
 * AsyncNotifyServiceTest.
 *
 * @author shiyiyue
 */
@ExtendWith(SpringExtension.class)
class AsyncNotifyServiceTest {
    
    @Mock
    ServerMemberManager serverMemberManager;
    
    MockedStatic<EnvUtil> envUtilMocked;
    
    MockedStatic<ConfigExecutor> configExecutorMocked;
    
    MockedStatic<InetUtils> inetUtilsMocked;
    
    @Mock
    private ConfigClusterRpcClientProxy configClusterRpcClientProxy;
    
    @BeforeEach
    void setUp() {
        envUtilMocked = Mockito.mockStatic(EnvUtil.class);
        configExecutorMocked = Mockito.mockStatic(ConfigExecutor.class);
        inetUtilsMocked = Mockito.mockStatic(InetUtils.class);
        inetUtilsMocked.when(InetUtils::getSelfIP).thenReturn("127.0.0.1");
    }
    
    @AfterEach
    void after() {
        envUtilMocked.close();
        inetUtilsMocked.close();
        configExecutorMocked.close();
    }
    
    @Test
    void testSyncConfigChangeCallback() {
        long timeStamp = System.currentTimeMillis();
        Member member1 = new Member();
        member1.setIp("testip1" + timeStamp);
        member1.setState(NodeState.UP);
        AsyncNotifyService asyncNotifyService = new AsyncNotifyService(serverMemberManager);
        ReflectionTestUtils.setField(asyncNotifyService, "configClusterRpcClientProxy", configClusterRpcClientProxy);
        String dataId = "testDataId" + timeStamp;
        String group = "testGroup";
        AsyncNotifyService.NotifySingleRpcTask notifySingleRpcTask = new AsyncNotifyService.NotifySingleRpcTask(dataId,
                group, null, null, 0, member1);
        configExecutorMocked.when(
                () -> ConfigExecutor.scheduleAsyncNotify(any(Runnable.class), anyLong(), any(TimeUnit.class)))
                .thenAnswer(invocation -> null);
        
        notifySingleRpcTask.setTag("test");
        notifySingleRpcTask.setBeta(false);
        AsyncRpcNotifyCallBack asyncRpcNotifyCallBack = new AsyncRpcNotifyCallBack(asyncNotifyService,
                notifySingleRpcTask);
        ConfigChangeClusterSyncResponse successResponse = new ConfigChangeClusterSyncResponse();
        //1. success response
        asyncRpcNotifyCallBack.onResponse(successResponse);
        //2. fail response
        successResponse.setResultCode(500);
        asyncRpcNotifyCallBack.onResponse(successResponse);
        //3. exception
        asyncRpcNotifyCallBack.onException(new NacosException());
        
        // expect schedule twice fail or exception response.
        configExecutorMocked.verify(
                () -> ConfigExecutor.scheduleAsyncNotify(any(AsyncNotifyService.AsyncRpcTask.class), anyLong(),
                        any(TimeUnit.class)), times(2));
    }
    
    /**
     * test HandleConfigDataChangeEvent. expect create a AsyncRpcTask and execute in ConfigExecutor.
     */
    @Test
    void testHandleConfigDataChangeEvent() {
        long timeStamp = System.currentTimeMillis();
        
        List<Member> memberList = new ArrayList<>();
        // member1 success
        Member member1 = new Member();
        member1.setIp("testip1" + timeStamp);
        member1.setState(NodeState.UP);
        memberList.add(member1);
        // member2 exception
        Member member2 = new Member();
        member2.setIp("testip2" + timeStamp);
        member2.setState(NodeState.UP);
        memberList.add(member2);
        // member3 unhealth
        Member member3 = new Member();
        member3.setIp("testip3" + timeStamp);
        member3.setState(NodeState.DOWN);
        memberList.add(member3);
        
        Mockito.when(serverMemberManager.allMembersWithoutSelf()).thenReturn(memberList);
        
        configExecutorMocked.when(
                () -> ConfigExecutor.scheduleAsyncNotify(any(Runnable.class), anyLong(), any(TimeUnit.class)))
                .thenAnswer(invocation -> null);
        String dataId = "testDataId" + timeStamp;
        String group = "testGroup";
        AsyncNotifyService asyncNotifyService = new AsyncNotifyService(serverMemberManager);
        asyncNotifyService.handleConfigDataChangeEvent(
                new ConfigDataChangeEvent(dataId, group, null, System.currentTimeMillis()));
        
        // expect schedule twice fail or exception response.
        configExecutorMocked.verify(() -> ConfigExecutor.executeAsyncNotify(any(AsyncNotifyService.AsyncRpcTask.class)),
                times(1));
        
    }
    
    @Test
    void testExecuteAsyncRpcTask() throws Exception {
        long timeStamp = System.currentTimeMillis();
        String dataId = "testDataId" + timeStamp;
        String group = "testGroup";
        
        List<Member> memberList = new ArrayList<>();
        // member1 success
        Member member1 = new Member();
        member1.setIp("testip1" + timeStamp);
        member1.setState(NodeState.UP);
        memberList.add(member1);
        // member2 exception
        Member member2 = new Member();
        member2.setIp("testip2" + timeStamp);
        member2.setState(NodeState.UP);
        memberList.add(member2);
        // member3 unhealth
        Member member3 = new Member();
        member3.setIp("testip3" + timeStamp);
        member3.setState(NodeState.DOWN);
        memberList.add(member3);
        Queue<AsyncNotifyService.NotifySingleRpcTask> rpcQueue = new LinkedList<>();
        
        for (Member member : memberList) {
            // grpc report data change only
            rpcQueue.add(
                    new AsyncNotifyService.NotifySingleRpcTask(dataId, group, null, null, System.currentTimeMillis(),
                            member));
        }
        
        AsyncNotifyService asyncNotifyService = new AsyncNotifyService(serverMemberManager);
        
        ReflectionTestUtils.setField(asyncNotifyService, "configClusterRpcClientProxy", configClusterRpcClientProxy);
        Mockito.when(serverMemberManager.allMembersWithoutSelf()).thenReturn(memberList);
        Mockito.when(serverMemberManager.hasMember(eq(member1.getAddress()))).thenReturn(true);
        Mockito.when(serverMemberManager.hasMember(eq(member2.getAddress()))).thenReturn(true);
        Mockito.when(serverMemberManager.hasMember(eq(member3.getAddress()))).thenReturn(true);
        Mockito.when(serverMemberManager.stateCheck(eq(member1.getAddress()), eq(HEALTHY_CHECK_STATUS)))
                .thenReturn(true);
        Mockito.when(serverMemberManager.stateCheck(eq(member2.getAddress()), eq(HEALTHY_CHECK_STATUS)))
                .thenReturn(true);
        // mock stateCheck fail before notify member3
        Mockito.when(serverMemberManager.stateCheck(eq(member3.getAddress()), eq(HEALTHY_CHECK_STATUS)))
                .thenReturn(false);
        //mock syncConfigChange exception when notify member2
        Mockito.doThrow(new NacosException()).when(configClusterRpcClientProxy)
                .syncConfigChange(eq(member2), any(ConfigChangeClusterSyncRequest.class), any(RequestCallBack.class));
        configExecutorMocked.when(
                () -> ConfigExecutor.scheduleAsyncNotify(any(Runnable.class), anyLong(), any(TimeUnit.class)))
                .thenAnswer(invocation -> null);
        
        asyncNotifyService.executeAsyncRpcTask(rpcQueue);
        
        Mockito.verify(configClusterRpcClientProxy, times(1))
                .syncConfigChange(eq(member1), any(ConfigChangeClusterSyncRequest.class), any(RequestCallBack.class));
        Mockito.verify(configClusterRpcClientProxy, times(1))
                .syncConfigChange(eq(member2), any(ConfigChangeClusterSyncRequest.class), any(RequestCallBack.class));
        Mockito.verify(configClusterRpcClientProxy, times(0))
                .syncConfigChange(eq(member3), any(ConfigChangeClusterSyncRequest.class), any(RequestCallBack.class));
        
        //verify scheduleAsyncNotify member2 & member3 in task when syncConfigChange fail
        configExecutorMocked.verify(
                () -> ConfigExecutor.scheduleAsyncNotify(any(AsyncNotifyService.AsyncRpcTask.class), anyLong(),
                        any(TimeUnit.class)), times(2));
        
    }
}
