content: /*
   * 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;
  
  import com.alibaba.nacos.common.notify.NotifyCenter;
  import com.alibaba.nacos.common.utils.MD5Utils;
  import com.alibaba.nacos.config.server.model.ConfigListenState;
  import com.alibaba.nacos.config.server.model.SampleResult;
  import com.alibaba.nacos.config.server.model.event.LocalDataChangeEvent;
  import com.alibaba.nacos.config.server.utils.ConfigExecutor;
  import com.alibaba.nacos.config.server.utils.GroupKey;
  import com.alibaba.nacos.config.server.utils.MD5Util;
  import com.alibaba.nacos.plugin.control.ControlManagerCenter;
  import com.alibaba.nacos.plugin.control.connection.ConnectionControlManager;
  import com.alibaba.nacos.plugin.control.connection.response.ConnectionCheckResponse;
  import com.alibaba.nacos.sys.env.EnvUtil;
  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.mockito.junit.jupiter.MockitoExtension;
  import org.mockito.junit.jupiter.MockitoSettings;
  import org.mockito.quality.Strictness;
  
  import jakarta.servlet.AsyncContext;
  import jakarta.servlet.http.HttpServletRequest;
  import jakarta.servlet.http.HttpServletResponse;
  import java.io.IOException;
  import java.io.PrintWriter;
  import java.util.Collections;
  import java.util.HashMap;
  import java.util.Map;
  
  import static org.junit.jupiter.api.Assertions.assertEquals;
  import static org.junit.jupiter.api.Assertions.assertFalse;
  import static org.mockito.ArgumentMatchers.any;
  import static org.mockito.ArgumentMatchers.anyInt;
  import static org.mockito.ArgumentMatchers.anyString;
  import static org.mockito.ArgumentMatchers.eq;
  import static org.mockito.Mockito.times;
  
  @ExtendWith(MockitoExtension.class)
  // todo remove this
  @MockitoSettings(strictness = Strictness.LENIENT)
  class LongPollingServiceTest {
      
      LongPollingService longPollingService;
      
      MockedStatic<ConfigCacheService> configCacheServiceMockedStatic;
      
      MockedStatic<ConfigExecutor> configExecutorMocked;
      
      MockedStatic<ControlManagerCenter> connectionControlManagerMockedStatic;
      
      @Mock
      ControlManagerCenter controlManagerCenter;
      
      @Mock
      ConnectionControlManager connectionControlManager;
      
      MockedStatic<SwitchService> switchServiceMockedStatic;
      
      MockedStatic<EnvUtil> envUtilMockedStatic;
      
      @BeforeEach
      void before() {
          longPollingService = new LongPollingService();
          switchServiceMockedStatic = Mockito.mockStatic(SwitchService.class);
          configCacheServiceMockedStatic = Mockito.mockStatic(ConfigCacheService.class);
          configExecutorMocked = Mockito.mockStatic(ConfigExecutor.class);
          connectionControlManagerMockedStatic = Mockito.mockStatic(ControlManagerCenter.class);
          connectionControlManagerMockedStatic.when(() -> ControlManagerCenter.getInstance()).thenReturn(controlManagerCenter);
          envUtilMockedStatic = Mockito.mockStatic(EnvUtil.class);
          envUtilMockedStatic.when(() -> EnvUtil.getProperty("nacos.config.cache.type", "nacos")).thenReturn("nacos");
          Mockito.when(controlManagerCenter.getConnectionControlManager()).thenReturn(connectionControlManager);
      }
      
      @AfterEach
      void after() {
          configCacheServiceMockedStatic.close();
          if (!configExecutorMocked.isClosed()) {
              configExecutorMocked.close();
          }
          connectionControlManagerMockedStatic.close();
          switchServiceMockedStatic.close();
          envUtilMockedStatic.close();
      }
      
      @Test
      void testAddLongPollingClientHasNotEqualsMd5() throws IOException {
          
          Map<String, ConfigListenState> clientMd5Map = new HashMap<>();
          String group = "group";
          String tenant = "tenat";
          String dataIdEquals = "dataIdEquals0";
          String groupKeyEquals = GroupKey.getKeyTenant(dataIdEquals, group, tenant);
          String md5Equals0 = MD5Utils.md5Hex("countEquals0", "UTF-8");
          ConfigListenState configListenState1 = new ConfigListenState(md5Equals0);
          clientMd5Map.put(groupKeyEquals, configListenState1);
          String md5NotEquals1 = MD5Utils.md5Hex("countNotEquals", "UTF-8");
          ConfigListenState configListenState2 = new ConfigListenState(md5NotEquals1);
          String dataIdNotEquals = "dataIdNotEquals0";
          String groupKeyNotEquals = GroupKey.getKeyTenant(dataIdNotEquals, group, tenant);
          clientMd5Map.put(groupKeyNotEquals, configListenState2);
          MockedStatic<MD5Util> md5UtilMockedStatic = Mockito.mockStatic(MD5Util.class);
          md5UtilMockedStatic.when(() -> MD5Util.compareMd5(any(), any(), any()))
                  .thenReturn(Collections.singletonMap(groupKeyNotEquals, configListenState2));
          
          HttpServletRequest httpServletRequest = Mockito.mock(HttpServletRequest.class);
          Mockito.when(httpServletRequest.getHeader(eq(LongPollingService.LONG_POLLING_NO_HANG_UP_HEADER))).thenReturn(null);
          String clientIp = "192.168.0.1";
          Mockito.when(httpServletRequest.getHeader(eq("X-Forwarded-For"))).thenReturn(clientIp);
          
          configCacheServiceMockedStatic.when(
                  () -> ConfigCacheService.isUptodate(eq(groupKeyNotEquals), eq(md5NotEquals1), eq(clientIp), eq(null))).thenReturn(false);
          configCacheServiceMockedStatic.when(() -> ConfigCacheService.isUptodate(eq(groupKeyEquals), eq(md5Equals0), eq(clientIp), eq(null)))
                  .thenReturn(true);
          HttpServletResponse httpServletResponse = Mockito.mock(HttpServletResponse.class);
          PrintWriter printWriter = Mockito.mock(PrintWriter.class);
          Mockito.when(httpServletResponse.getWriter()).thenReturn(printWriter);
          int propSize = 3;
          longPollingService.addLongPollingClient(httpServletRequest, httpServletResponse, clientMd5Map, propSize);
          
          String responseString = MD5Util.compareMd5ResultString(Collections.singletonMap(groupKeyNotEquals, configListenState2));
          //expect print not equals group
          Mockito.verify(printWriter, times(1)).println(eq(responseString));
          Mockito.verify(httpServletResponse, times(1)).setStatus(eq(HttpServletResponse.SC_OK));
          md5UtilMockedStatic.close();
      }
      
      @Test
      void testRejectByConnectionLimit() throws Exception {
          //mock connection no limit
          ConnectionCheckResponse connectionCheckResponse = new ConnectionCheckResponse();
          connectionCheckResponse.setSuccess(false);
          Mockito.when(connectionControlManager.check(any())).thenReturn(connectionCheckResponse);
          HttpServletResponse httpServletResponse = Mockito.mock(HttpServletResponse.class);
          PrintWriter printWriter = Mockito.mock(PrintWriter.class);
          Mockito.when(httpServletResponse.getWriter()).thenReturn(printWriter);
          
          HttpServletRequest httpServletRequest = Mockito.mock(HttpServletRequest.class);
          Mockito.when(httpServletRequest.getHeader(eq(LongPollingService.LONG_POLLING_NO_HANG_UP_HEADER))).thenReturn(null);
          String clientIp = "192.168.0.1";
          Mockito.when(httpServletRequest.getHeader(eq("X-Forwarded-For"))).thenReturn(clientIp);
          Mockito.when(httpServletRequest.startAsync()).thenReturn(Mockito.mock(AsyncContext.class));
          int propSize = 3;
          Map<String, ConfigListenState> clientMd5Map = new HashMap<>();
          longPollingService.addLongPollingClient(httpServletRequest, httpServletResponse, clientMd5Map, propSize);
          Thread.sleep(3000L);
          //expect response not returned
          Mockito.verify(httpServletResponse, times(1)).setStatus(eq(503));
          
      }
      
      @Test
      void testAddLongPollingClientAllEqualsMd5() throws IOException {
          //mock connection no limit
          ConnectionCheckResponse connectionCheckResponse = new ConnectionCheckResponse();
          connectionCheckResponse.setSuccess(true);
          Mockito.when(connectionControlManager.check(any())).thenReturn(connectionCheckResponse);
          
          Map<String, ConfigListenState> clientMd5Map = new HashMap<>();
          String group = "group";
          String tenant = "tenat";
          String dataIdEquals = "dataIdEquals01";
          String groupKeyEquals = GroupKey.getKeyTenant(dataIdEquals, group, tenant);
          
          String md5Equals0 = MD5Utils.md5Hex("countEquals01", "UTF-8");
          ConfigListenState configListenState1 = new ConfigListenState(md5Equals0);
          clientMd5Map.put(groupKeyEquals, configListenState1);
          String md5NotEquals1 = MD5Utils.md5Hex("countNotEquals1", "UTF-8");
          ConfigListenState configListenState2 = new ConfigListenState(md5NotEquals1);
          String dataIdNotEquals = "dataIdNotEquals01";
          String groupKeyNotEquals = GroupKey.getKeyTenant(dataIdNotEquals, group, tenant);
          clientMd5Map.put(groupKeyNotEquals, configListenState2);
          HttpServletRequest httpServletRequest = Mockito.mock(HttpServletRequest.class);
          
          Mockito.when(httpServletRequest.getHeader(eq(LongPollingService.LONG_POLLING_HEADER))).thenReturn("5000");
          Mockito.when(httpServletRequest.getHeader(eq(LongPollingService.LONG_POLLING_NO_HANG_UP_HEADER))).thenReturn(null);
          String clientIp = "192.168.0.1";
          Mockito.when(httpServletRequest.getHeader(eq("X-Forwarded-For"))).thenReturn(clientIp);
          Mockito.when(httpServletRequest.startAsync()).thenReturn(Mockito.mock(AsyncContext.class));
          configCacheServiceMockedStatic.when(
                  () -> ConfigCacheService.isUptodate(eq(groupKeyNotEquals), eq(md5NotEquals1), eq(clientIp), eq(null))).thenReturn(true);
          configCacheServiceMockedStatic.when(() -> ConfigCacheService.isUptodate(eq(groupKeyEquals), eq(md5Equals0), eq(clientIp), eq(null)))
                  .thenReturn(true);
          int propSize = 3;
          HttpServletResponse httpServletResponse = Mockito.mock(HttpServletResponse.class);
          longPollingService.addLongPollingClient(httpServletRequest, httpServletResponse, clientMd5Map, propSize);
          
          //expect response not returned
          Mockito.verify(httpServletResponse, times(0)).setStatus(anyInt());
          //expect to schedule a task
          configExecutorMocked.verify(() -> ConfigExecutor.executeLongPolling(any(LongPollingService.ClientLongPolling.class)), times(1));
          
      }
      
      @Test
      void testReceiveDataChangeEventAndNotify() throws Exception {
          configExecutorMocked.close();
          
          //mock connection no limit
          ConnectionCheckResponse connectionCheckResponse = new ConnectionCheckResponse();
          connectionCheckResponse.setSuccess(true);
          Mockito.when(connectionControlManager.check(any())).thenReturn(connectionCheckResponse);
          
          String dataIdChanged = "dataIdChanged";
          String group = "group";
          String tenant = "tenant";
          String groupKeyChanged = GroupKey.getKeyTenant(dataIdChanged, group, tenant);
          Map<String, ConfigListenState> clientMd5Map = new HashMap<>();
          ConfigListenState configListenState = new ConfigListenState("mockMd5");
          clientMd5Map.put(groupKeyChanged, configListenState);
          HttpServletRequest httpServletRequest = Mockito.mock(HttpServletRequest.class);
          HttpServletResponse httpServletResponse = Mockito.mock(HttpServletResponse.class);
          PrintWriter printWriter = Mockito.mock(PrintWriter.class);
          Mockito.when(httpServletResponse.getWriter()).thenReturn(printWriter);
          
          Mockito.when(httpServletRequest.getHeader(eq(LongPollingService.LONG_POLLING_HEADER))).thenReturn("5000");
          Mockito.when(httpServletRequest.getHeader(eq(LongPollingService.LONG_POLLING_NO_HANG_UP_HEADER))).thenReturn(null);
          String clientIp = "192.168.0.1";
          Mockito.when(httpServletRequest.getHeader(eq("X-Forwarded-For"))).thenReturn(clientIp);
          AsyncContext asyncContext = Mockito.mock(AsyncContext.class);
          Mockito.when(httpServletRequest.startAsync()).thenReturn(asyncContext);
          Mockito.when(asyncContext.getRequest()).thenReturn(httpServletRequest);
          Mockito.when(asyncContext.getResponse()).thenReturn(httpServletResponse);
          
          configCacheServiceMockedStatic.when(() -> ConfigCacheService.isUptodate(anyString(), anyString(), anyString(), eq(null)))
                  .thenReturn(true);
          
          longPollingService.addLongPollingClient(httpServletRequest, httpServletResponse, clientMd5Map, 3);
          
          //test getSubscribleInfo by groupKey
          SampleResult subscribleInfo = longPollingService.getCollectSubscribleInfo(dataIdChanged, group, tenant);
          Map<String, String> lisentersGroupkeyStatus = subscribleInfo.getLisentersGroupkeyStatus();
          assertFalse(lisentersGroupkeyStatus.isEmpty());
          assertEquals("mockMd5", lisentersGroupkeyStatus.get(clientIp));
          SampleResult collectSubscribleInfoByIp = longPollingService.getCollectSubscribleInfoByIp(clientIp);
          Map<String, String> lisentersGroupkeyStatus1 = collectSubscribleInfoByIp.getLisentersGroupkeyStatus();
          assertFalse(lisentersGroupkeyStatus1.isEmpty());
          assertEquals("mockMd5", lisentersGroupkeyStatus1.get(groupKeyChanged));
          
          //test receive config change event
          LocalDataChangeEvent localDataChangeEvent = new LocalDataChangeEvent(groupKeyChanged);
          
          NotifyCenter.publishEvent(localDataChangeEvent);
          Thread.sleep(1100L);
          String responseString = MD5Util.compareMd5ResultString(Collections.singletonMap(groupKeyChanged, configListenState));
          //expect print not equals group
          Mockito.verify(printWriter, times(1)).println(eq(responseString));
          Mockito.verify(asyncContext, times(1)).complete();
          
      }
      
      @Test
      void testLongPollingTimeout() throws Exception {
          configExecutorMocked.close();
          String dataIdChanged = "dataIdChanged";
          String group = "group";
          String tenant = "tenant";
          String groupKeyChanged = GroupKey.getKeyTenant(dataIdChanged, group, tenant);
          
          //mock connection no limit
          ConnectionCheckResponse connectionCheckResponse = new ConnectionCheckResponse();
          connectionCheckResponse.setSuccess(true);
          Mockito.when(connectionControlManager.check(any())).thenReturn(connectionCheckResponse);
          
          Map<String, ConfigListenState> clientMd5Map = new HashMap<>();
          ConfigListenState configListenState = new ConfigListenState("md5");
          clientMd5Map.put(groupKeyChanged, configListenState);
          switchServiceMockedStatic.when(() -> SwitchService.getSwitchInteger(eq("MIN_LONG_POOLING_TIMEOUT"), eq(10000))).thenReturn(1000);
          HttpServletRequest httpServletRequest = Mockito.mock(HttpServletRequest.class);
          
          Mockito.when(httpServletRequest.getHeader(eq(LongPollingService.LONG_POLLING_HEADER))).thenReturn("1000");
          Mockito.when(httpServletRequest.getHeader(eq(LongPollingService.LONG_POLLING_NO_HANG_UP_HEADER))).thenReturn(null);
          String clientIp = "192.168.0.1";
          Mockito.when(httpServletRequest.getHeader(eq("X-Forwarded-For"))).thenReturn(clientIp);
          AsyncContext asyncContext = Mockito.mock(AsyncContext.class);
          Mockito.when(httpServletRequest.startAsync()).thenReturn(asyncContext);
          Mockito.when(asyncContext.getRequest()).thenReturn(httpServletRequest);
          
          configCacheServiceMockedStatic.when(() -> ConfigCacheService.isUptodate(anyString(), anyString(), anyString(), eq(null)))
                  .thenReturn(true);
          HttpServletResponse httpServletResponse = Mockito.mock(HttpServletResponse.class);
          longPollingService.addLongPollingClient(httpServletRequest, httpServletResponse, clientMd5Map, 3);
          
          //wait time out condition arrived.
          Thread.sleep(1200L);
          //expect print not equals group
          Mockito.verify(asyncContext, times(1)).complete();
          
      }
  }
  