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.client.config.impl;
  
  import com.alibaba.nacos.api.PropertyKeyConst;
  import com.alibaba.nacos.api.config.ConfigType;
  import com.alibaba.nacos.api.config.listener.AbstractListener;
  import com.alibaba.nacos.api.config.listener.Listener;
  import com.alibaba.nacos.api.config.remote.request.ClientConfigMetricRequest;
  import com.alibaba.nacos.api.config.remote.request.ConfigBatchListenRequest;
  import com.alibaba.nacos.api.config.remote.request.ConfigChangeNotifyRequest;
  import com.alibaba.nacos.api.config.remote.request.ConfigPublishRequest;
  import com.alibaba.nacos.api.config.remote.request.ConfigQueryRequest;
  import com.alibaba.nacos.api.config.remote.request.ConfigRemoveRequest;
  import com.alibaba.nacos.api.config.remote.response.ClientConfigMetricResponse;
  import com.alibaba.nacos.api.config.remote.response.ConfigChangeBatchListenResponse;
  import com.alibaba.nacos.api.config.remote.response.ConfigPublishResponse;
  import com.alibaba.nacos.api.config.remote.response.ConfigQueryResponse;
  import com.alibaba.nacos.api.config.remote.response.ConfigRemoveResponse;
  import com.alibaba.nacos.api.exception.NacosException;
  import com.alibaba.nacos.client.config.common.GroupKey;
  import com.alibaba.nacos.client.config.filter.impl.ConfigFilterChainManager;
  import com.alibaba.nacos.client.config.filter.impl.ConfigResponse;
  import com.alibaba.nacos.client.env.NacosClientProperties;
  import com.alibaba.nacos.client.monitor.MetricsMonitor;
  import com.alibaba.nacos.common.remote.ConnectionType;
  import com.alibaba.nacos.common.remote.client.RpcClient;
  import com.alibaba.nacos.common.remote.client.RpcClientFactory;
  import com.alibaba.nacos.common.remote.client.grpc.GrpcClientConfig;
  import com.alibaba.nacos.common.utils.JacksonUtils;
  import com.alibaba.nacos.common.utils.MD5Utils;
  import com.fasterxml.jackson.databind.JsonNode;
  import io.prometheus.client.Gauge;
  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 java.io.File;
  import java.lang.reflect.Field;
  import java.lang.reflect.Method;
  import java.util.ArrayList;
  import java.util.Arrays;
  import java.util.Collections;
  import java.util.List;
  import java.util.Map;
  import java.util.Properties;
  import java.util.concurrent.Executor;
  import java.util.concurrent.atomic.AtomicBoolean;
  import java.util.concurrent.atomic.AtomicReference;
  
  import static com.alibaba.nacos.api.annotation.NacosProperties.NAMESPACE;
  import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
  import static org.junit.jupiter.api.Assertions.assertEquals;
  import static org.junit.jupiter.api.Assertions.assertFalse;
  import static org.junit.jupiter.api.Assertions.assertNotNull;
  import static org.junit.jupiter.api.Assertions.assertNull;
  import static org.junit.jupiter.api.Assertions.assertTrue;
  import static org.junit.jupiter.api.Assertions.fail;
  import static org.mockito.ArgumentMatchers.any;
  import static org.mockito.ArgumentMatchers.anyLong;
  import static org.mockito.ArgumentMatchers.anyString;
  import static org.mockito.ArgumentMatchers.eq;
  import static org.mockito.Mockito.doReturn;
  import static org.mockito.Mockito.doThrow;
  import static org.mockito.Mockito.mock;
  import static org.mockito.Mockito.never;
  import static org.mockito.Mockito.times;
  import static org.mockito.Mockito.verify;
  
  @ExtendWith(MockitoExtension.class)
  class ClientWorkerTest {
      
      private static final String TEST_NAMESPACE = "TEST_NAMESPACE";
      
      MockedStatic<RpcClientFactory> rpcClientFactoryMockedStatic;
      
      MockedStatic<LocalConfigInfoProcessor> localConfigInfoProcessorMockedStatic;
      
      @Mock
      RpcClient rpcClient;
      
      private ClientWorker clientWorker;
      
      private ClientWorker clientWorkerSpy;
      
      @BeforeEach
      void before() throws Exception {
          rpcClientFactoryMockedStatic = Mockito.mockStatic(RpcClientFactory.class);
          
          rpcClientFactoryMockedStatic.when(() -> RpcClientFactory.createClient(anyString(), any(ConnectionType.class),
                  any(GrpcClientConfig.class))).thenReturn(rpcClient);
          rpcClientFactoryMockedStatic.when(() -> RpcClientFactory.createClient(anyString(), any(ConnectionType.class),
                  any(GrpcClientConfig.class))).thenReturn(rpcClient);
          localConfigInfoProcessorMockedStatic = Mockito.mockStatic(LocalConfigInfoProcessor.class);
          Properties properties = new Properties();
          properties.put(PropertyKeyConst.NAMESPACE, TEST_NAMESPACE);
          ConfigFilterChainManager filter = new ConfigFilterChainManager(properties);
          ConfigServerListManager serverListManager = mock(ConfigServerListManager.class);
          final NacosClientProperties nacosClientProperties = NacosClientProperties.PROTOTYPE.derive(properties);
          try {
              clientWorker = new ClientWorker(filter, serverListManager, nacosClientProperties);
          } catch (NacosException e) {
              throw new RuntimeException(e);
          }
          clientWorkerSpy = Mockito.spy(clientWorker);
      }
      
      @AfterEach
      void after() {
          rpcClientFactoryMockedStatic.close();
          localConfigInfoProcessorMockedStatic.close();
      }
      
      @Test
      void testConstruct() throws NacosException {
          Properties prop = new Properties();
          ConfigFilterChainManager filter = new ConfigFilterChainManager(new Properties());
          ConfigServerListManager agent = Mockito.mock(ConfigServerListManager.class);
          
          final NacosClientProperties nacosClientProperties = NacosClientProperties.PROTOTYPE.derive(prop);
          ClientWorker clientWorker = new ClientWorker(filter, agent, nacosClientProperties);
          assertNotNull(clientWorker);
      }
      
      @Test
      void testAddListenerWithoutTenant() throws NacosException {
          Properties prop = new Properties();
          ConfigFilterChainManager filter = new ConfigFilterChainManager(new Properties());
          ConfigServerListManager agent = Mockito.mock(ConfigServerListManager.class);
          
          final NacosClientProperties nacosClientProperties = NacosClientProperties.PROTOTYPE.derive(prop);
          ClientWorker clientWorker = new ClientWorker(filter, agent, nacosClientProperties);
          String dataId = "a";
          String group = "b";
          
          Listener listener = new AbstractListener() {
              @Override
              public void receiveConfigInfo(String configInfo) {
              }
          };
          
          clientWorker.addListeners(dataId, group, Collections.singletonList(listener));
          List<Listener> listeners = clientWorker.getCache(dataId, group).getListeners();
          assertEquals(1, listeners.size());
          assertEquals(listener, listeners.get(0));
          
          clientWorker.removeListener(dataId, group, listener);
          listeners = clientWorker.getCache(dataId, group).getListeners();
          assertEquals(0, listeners.size());
          
          CacheData cacheData = clientWorker.addCacheDataIfAbsent(dataId, group);
          assertEquals(cacheData, clientWorker.getCache(dataId, group));
      }
      
      @Test
      void testListenerWithTenant() throws NacosException {
          Properties prop = new Properties();
          ConfigFilterChainManager filter = new ConfigFilterChainManager(new Properties());
          ConfigServerListManager agent = Mockito.mock(ConfigServerListManager.class);
          
          final NacosClientProperties nacosClientProperties = NacosClientProperties.PROTOTYPE.derive(prop);
          ClientWorker clientWorker = new ClientWorker(filter, agent, nacosClientProperties);
          
          Listener listener = new AbstractListener() {
              @Override
              public void receiveConfigInfo(String configInfo) {
              }
          };
          
          String dataId = "a";
          String group = "b";
          
          clientWorker.addTenantListeners(dataId, group, Collections.singletonList(listener));
          List<Listener> listeners = clientWorker.getCache(dataId, group).getListeners();
          assertEquals(1, listeners.size());
          assertEquals(listener, listeners.get(0));
          
          clientWorker.removeTenantListener(dataId, group, listener);
          listeners = clientWorker.getCache(dataId, group).getListeners();
          assertEquals(0, listeners.size());
          
          String content = "d";
          clientWorker.addTenantListenersWithContent(dataId, group, content, null, Collections.singletonList(listener));
          listeners = clientWorker.getCache(dataId, group).getListeners();
          assertEquals(1, listeners.size());
          assertEquals(listener, listeners.get(0));
          
          clientWorker.removeTenantListener(dataId, group, listener);
          listeners = clientWorker.getCache(dataId, group).getListeners();
          assertEquals(0, listeners.size());
          
          String tenant = "c";
          CacheData cacheData = clientWorker.addCacheDataIfAbsent(dataId, group, tenant);
          assertEquals(cacheData, clientWorker.getCache(dataId, group, tenant));
          
          clientWorker.removeCache(dataId, group, tenant);
          assertNull(clientWorker.getCache(dataId, group, tenant));
          
      }
      
      @Test
      void testPublishConfigSuccess() throws NacosException {
          Properties prop = new Properties();
          ConfigFilterChainManager filter = new ConfigFilterChainManager(new Properties());
          ConfigServerListManager agent = Mockito.mock(ConfigServerListManager.class);
          
          final NacosClientProperties nacosClientProperties = NacosClientProperties.PROTOTYPE.derive(prop);
          ClientWorker clientWorker = new ClientWorker(filter, agent, nacosClientProperties);
          
          String dataId = "a";
          String group = "b";
          String tenant = "c";
          String content = "d";
          
          String appName = "app";
          String tag = "tag";
          
          String betaIps = "1.1.1.1";
          String casMd5 = "1111";
          
          String type = "properties";
          Mockito.when(rpcClient.request(any(ConfigPublishRequest.class))).thenReturn(new ConfigPublishResponse());
          boolean b = clientWorker.publishConfig(dataId, group, tenant, appName, tag, betaIps, content, null, casMd5,
                  type);
          assertTrue(b);
      }
      
      @Test
      void testPublishConfigFail() throws NacosException {
          Properties prop = new Properties();
          ConfigFilterChainManager filter = new ConfigFilterChainManager(new Properties());
          ConfigServerListManager agent = Mockito.mock(ConfigServerListManager.class);
          
          final NacosClientProperties nacosClientProperties = NacosClientProperties.PROTOTYPE.derive(prop);
          ClientWorker clientWorker = new ClientWorker(filter, agent, nacosClientProperties);
          
          String dataId = "a";
          String group = "b";
          String tenant = "c";
          String content = "d";
          
          String appName = "app";
          String tag = "tag";
          
          String betaIps = "1.1.1.1";
          String casMd5 = "1111";
          
          String type = "properties";
          Mockito.when(rpcClient.request(any(ConfigPublishRequest.class)))
                  .thenReturn(ConfigPublishResponse.buildFailResponse(503, "over limit"));
          boolean b = clientWorker.publishConfig(dataId, group, tenant, appName, tag, betaIps, content, null, casMd5,
                  type);
          assertFalse(b);
          
      }
      
      @Test
      void testPublishConfigException() throws NacosException {
          Properties prop = new Properties();
          ConfigFilterChainManager filter = new ConfigFilterChainManager(new Properties());
          ConfigServerListManager agent = Mockito.mock(ConfigServerListManager.class);
          
          final NacosClientProperties nacosClientProperties = NacosClientProperties.PROTOTYPE.derive(prop);
          ClientWorker clientWorker = new ClientWorker(filter, agent, nacosClientProperties);
          
          String dataId = "a";
          String group = "b";
          String tenant = "c";
          String content = "d";
          
          String appName = "app";
          String tag = "tag";
          
          String betaIps = "1.1.1.1";
          String casMd5 = "1111";
          
          String type = "properties";
          Mockito.when(rpcClient.request(any(ConfigPublishRequest.class))).thenThrow(new NacosException());
          boolean b = clientWorker.publishConfig(dataId, group, tenant, appName, tag, betaIps, content, null, casMd5,
                  type);
          assertFalse(b);
          
      }
      
      @Test
      void testRemoveConfig() throws NacosException {
          
          Properties prop = new Properties();
          ConfigFilterChainManager filter = new ConfigFilterChainManager(new Properties());
          ConfigServerListManager agent = Mockito.mock(ConfigServerListManager.class);
          
          final NacosClientProperties nacosClientProperties = NacosClientProperties.PROTOTYPE.derive(prop);
          ClientWorker clientWorker = new ClientWorker(filter, agent, nacosClientProperties);
          
          String dataId = "a";
          String group = "b";
          String tenant = "c";
          
          String tag = "tag";
          try {
              Mockito.when(rpcClient.request(any(ConfigRemoveRequest.class)))
                      .thenThrow(new NacosException(503, "overlimit"));
              
              clientWorker.removeConfig(dataId, group, tenant, tag);
              fail();
          } catch (NacosException e) {
              assertEquals("overlimit", e.getErrMsg());
              assertEquals(503, e.getErrCode());
              
          }
      }
      
      @Test
      void testGeConfigConfigSuccess() throws NacosException {
          
          Properties prop = new Properties();
          ConfigServerListManager agent = Mockito.mock(ConfigServerListManager.class);
          final NacosClientProperties nacosClientProperties = NacosClientProperties.PROTOTYPE.derive(prop);
          ClientWorker clientWorker = new ClientWorker(null, agent, nacosClientProperties);
          
          String dataId = "a";
          String group = "b";
          String tenant = "c";
          String content = "content" + System.currentTimeMillis();
          
          Mockito.when(rpcClient.request(any(ConfigQueryRequest.class), anyLong()))
                  .thenReturn(ConfigQueryResponse.buildSuccessResponse(content));
          
          ConfigResponse configResponse = clientWorker.getServerConfig(dataId, group, tenant, 100, true);
          assertEquals(content, configResponse.getContent());
          localConfigInfoProcessorMockedStatic.verify(
                  () -> LocalConfigInfoProcessor.saveSnapshot(eq(clientWorker.getAgentName()), eq(dataId), eq(group),
                          eq(tenant), eq(content)), times(1));
      }
      
      @Test
      void testHandleConfigChangeReqeust() throws Exception {
          
          Properties prop = new Properties();
          String tenant = "c";
          
          prop.put(NAMESPACE, tenant);
          ConfigServerListManager agent = Mockito.mock(ConfigServerListManager.class);
          final NacosClientProperties nacosClientProperties = NacosClientProperties.PROTOTYPE.derive(prop);
          ClientWorker clientWorker = new ClientWorker(null, agent, nacosClientProperties);
          
          AtomicReference<Map<String, CacheData>> cacheMapMocked = Mockito.mock(AtomicReference.class);
          Field cacheMap = ClientWorker.class.getDeclaredField("cacheMap");
          cacheMap.setAccessible(true);
          cacheMap.set(clientWorker, cacheMapMocked);
          Map<String, CacheData> cacheDataMapMocked = Mockito.mock(Map.class);
          Mockito.when(cacheMapMocked.get()).thenReturn(cacheDataMapMocked);
          CacheData cacheDataMocked = Mockito.mock(CacheData.class);
          AtomicBoolean atomicBoolean = Mockito.mock(AtomicBoolean.class);
          Mockito.when(cacheDataMocked.getReceiveNotifyChanged()).thenReturn(atomicBoolean);
          String dataId = "a";
          String group = "b";
          Mockito.when(cacheDataMapMocked.get(GroupKey.getKeyTenant(dataId, group, tenant))).thenReturn(cacheDataMocked);
          ConfigChangeNotifyRequest configChangeNotifyRequest = ConfigChangeNotifyRequest.build(dataId, group, tenant);
          ((ClientWorker.ConfigRpcTransportClient) clientWorker.getAgent()).handleConfigChangeNotifyRequest(
                  configChangeNotifyRequest, "testname");
          Mockito.verify(cacheDataMocked, times(1)).setConsistentWithServer(false);
          Mockito.verify(atomicBoolean, times(1)).set(true);
      }
      
      @Test
      void testHandleClientMetricsReqeust() throws Exception {
          
          Properties prop = new Properties();
          String tenant = "c";
          
          prop.put(NAMESPACE, tenant);
          ConfigServerListManager agent = Mockito.mock(ConfigServerListManager.class);
          final NacosClientProperties nacosClientProperties = NacosClientProperties.PROTOTYPE.derive(prop);
          ClientWorker clientWorker = new ClientWorker(null, agent, nacosClientProperties);
          
          AtomicReference<Map<String, CacheData>> cacheMapMocked = Mockito.mock(AtomicReference.class);
          Field cacheMap = ClientWorker.class.getDeclaredField("cacheMap");
          cacheMap.setAccessible(true);
          cacheMap.set(clientWorker, cacheMapMocked);
          Map<String, CacheData> cacheDataMapMocked = Mockito.mock(Map.class);
          Mockito.when(cacheMapMocked.get()).thenReturn(cacheDataMapMocked);
          CacheData cacheDataMocked = Mockito.mock(CacheData.class);
          String content = "content1324567";
          String md5 = MD5Utils.md5Hex(content, "UTF-8");
          Mockito.when(cacheDataMocked.getContent()).thenReturn(content);
          Mockito.when(cacheDataMocked.getMd5()).thenReturn(md5);
          Field uuid1 = ClientWorker.class.getDeclaredField("uuid");
          uuid1.setAccessible(true);
          String uuid = (String) uuid1.get(clientWorker);
          String dataId = "a23456789";
          String group = "b";
          Mockito.when(cacheDataMapMocked.get(GroupKey.getKeyTenant(dataId, group, tenant))).thenReturn(cacheDataMocked);
          ClientConfigMetricRequest configMetricsRequest = new ClientConfigMetricRequest();
          
          configMetricsRequest.setMetricsKeys(Arrays.asList(
                  ClientConfigMetricRequest.MetricsKey.build(ClientConfigMetricRequest.MetricsKey.CACHE_DATA,
                          GroupKey.getKeyTenant(dataId, group, tenant)),
                  ClientConfigMetricRequest.MetricsKey.build(ClientConfigMetricRequest.MetricsKey.SNAPSHOT_DATA,
                          GroupKey.getKeyTenant(dataId, group, tenant))));
          
          ClientConfigMetricResponse metricResponse = ((ClientWorker.ConfigRpcTransportClient) clientWorker.getAgent()).handleClientMetricsRequest(
                  configMetricsRequest);
          JsonNode jsonNode = JacksonUtils.toObj(metricResponse.getMetrics().get(uuid).toString());
          String metricValues = jsonNode.get("metricValues")
                  .get(ClientConfigMetricRequest.MetricsKey.build(ClientConfigMetricRequest.MetricsKey.CACHE_DATA,
                          GroupKey.getKeyTenant(dataId, group, tenant)).toString()).textValue();
          
          int colonIndex = metricValues.lastIndexOf(":");
          assertEquals(content, metricValues.substring(0, colonIndex));
          assertEquals(md5, metricValues.substring(colonIndex + 1, metricValues.length()));
          
      }
      
      @Test
      void testGeConfigConfigNotFound() throws NacosException {
          
          Properties prop = new Properties();
          ConfigServerListManager agent = Mockito.mock(ConfigServerListManager.class);
          final NacosClientProperties nacosClientProperties = NacosClientProperties.PROTOTYPE.derive(prop);
          ClientWorker clientWorker = new ClientWorker(null, agent, nacosClientProperties);
          
          String dataId = "a";
          String group = "b";
          String tenant = "c";
          ConfigQueryResponse configQueryResponse = new ConfigQueryResponse();
          configQueryResponse.setErrorInfo(ConfigQueryResponse.CONFIG_NOT_FOUND, "config not found");
          Mockito.when(rpcClient.request(any(ConfigQueryRequest.class), anyLong())).thenReturn(configQueryResponse);
          
          ConfigResponse configResponse = clientWorker.getServerConfig(dataId, group, tenant, 100, true);
          assertNull(configResponse.getContent());
          localConfigInfoProcessorMockedStatic.verify(
                  () -> LocalConfigInfoProcessor.saveSnapshot(eq(clientWorker.getAgentName()), eq(dataId), eq(group),
                          eq(tenant), eq(null)), times(1));
          
      }
      
      @Test
      void testGeConfigConfigConflict() throws NacosException {
          
          Properties prop = new Properties();
          ConfigServerListManager agent = Mockito.mock(ConfigServerListManager.class);
          final NacosClientProperties nacosClientProperties = NacosClientProperties.PROTOTYPE.derive(prop);
          ClientWorker clientWorker = new ClientWorker(null, agent, nacosClientProperties);
          
          String dataId = "a";
          String group = "b";
          String tenant = "c";
          ConfigQueryResponse configQueryResponse = new ConfigQueryResponse();
          configQueryResponse.setErrorInfo(ConfigQueryResponse.CONFIG_QUERY_CONFLICT, "config is being modified");
          Mockito.when(rpcClient.request(any(ConfigQueryRequest.class), anyLong())).thenReturn(configQueryResponse);
          
          try {
              clientWorker.getServerConfig(dataId, group, tenant, 100, true);
              fail();
          } catch (NacosException e) {
              assertEquals(NacosException.CONFLICT, e.getErrCode());
          }
      }
      
      @Test
      void testShutdown() throws NacosException, NoSuchFieldException, IllegalAccessException {
          Properties prop = new Properties();
          ConfigFilterChainManager filter = new ConfigFilterChainManager(new Properties());
          ConfigServerListManager agent = Mockito.mock(ConfigServerListManager.class);
          
          final NacosClientProperties nacosClientProperties = NacosClientProperties.PROTOTYPE.derive(prop);
          ClientWorker clientWorker = new ClientWorker(filter, agent, nacosClientProperties);
          clientWorker.shutdown();
          Field agent1 = ClientWorker.class.getDeclaredField("agent");
          agent1.setAccessible(true);
          ConfigTransportClient o = (ConfigTransportClient) agent1.get(clientWorker);
          assertTrue(o.getExecutor().isShutdown());
          agent1.setAccessible(false);
          
          assertNull(clientWorker.getAgentName());
      }
      
      @Test
      void testExecuteConfigListen() throws Exception {
          Properties prop = new Properties();
          ConfigFilterChainManager filter = new ConfigFilterChainManager(new Properties());
          ConfigServerListManager agent = Mockito.mock(ConfigServerListManager.class);
          Mockito.when(agent.getName()).thenReturn("mocktest");
          final NacosClientProperties nacosClientProperties = NacosClientProperties.PROTOTYPE.derive(prop);
          ClientWorker clientWorker = new ClientWorker(filter, agent, nacosClientProperties);
          clientWorker.shutdown();
          
          List<CacheData> cacheDatas = new ArrayList<>();
          String group = "group123";
          String tenant = "tenant122324";
          //mock discards cache
          String dataIdDiscard = "dataIdDiscard" + System.currentTimeMillis();
          
          CacheData cacheDataDiscard = discardCache(filter, agent.getName(), dataIdDiscard, group, tenant);
          cacheDatas.add(cacheDataDiscard);
          //mock use local cache
          String dataIdUseLocalCache = "dataIdUseLocalCache" + System.currentTimeMillis();
          CacheData cacheUseLocalCache = useLocalCache(filter, agent.getName(), dataIdUseLocalCache, group, tenant,
                  "content" + System.currentTimeMillis());
          assertFalse(cacheUseLocalCache.isUseLocalConfigInfo());
          
          cacheDatas.add(cacheUseLocalCache);
          
          //mock normal cache
          String dataIdNormal = "dataIdNormal" + System.currentTimeMillis();
          CacheData cacheNormal = normalNotConsistentCache(filter, agent.getName(), dataIdNormal, group, tenant);
          AtomicReference<String> normalContent = new AtomicReference<>();
          cacheNormal.addListener(new Listener() {
              @Override
              public Executor getExecutor() {
                  return null;
              }
              
              @Override
              public void receiveConfigInfo(String configInfo) {
                  System.out.println(configInfo);
                  normalContent.set(configInfo);
              }
          });
          cacheDatas.add(cacheNormal);
          cacheNormal.setInitializing(false);
          Map<String, CacheData> cacheDataMapMocked = Mockito.mock(Map.class);
          Mockito.when(cacheDataMapMocked.get(GroupKey.getKeyTenant(dataIdNormal, group, tenant)))
                  .thenReturn(cacheNormal);
          Mockito.when(cacheDataMapMocked.containsKey(GroupKey.getKeyTenant(dataIdNormal, group, tenant)))
                  .thenReturn(true);
          
          Mockito.when(cacheDataMapMocked.values()).thenReturn(cacheDatas);
          AtomicReference<Map<String, CacheData>> cacheMapMocked = Mockito.mock(AtomicReference.class);
          Mockito.when(cacheMapMocked.get()).thenReturn(cacheDataMapMocked);
          Field cacheMap = ClientWorker.class.getDeclaredField("cacheMap");
          cacheMap.setAccessible(true);
          cacheMap.set(clientWorker, cacheMapMocked);
          
          //mock request
          ConfigChangeBatchListenResponse.ConfigContext configContext = new ConfigChangeBatchListenResponse.ConfigContext();
          configContext.setDataId(dataIdNormal);
          configContext.setGroup(group);
          configContext.setTenant(tenant);
          ConfigChangeBatchListenResponse response = new ConfigChangeBatchListenResponse();
          response.setChangedConfigs(Collections.singletonList(configContext));
          
          RpcClient rpcClientInner = Mockito.mock(RpcClient.class);
          Mockito.when(rpcClientInner.isWaitInitiated()).thenReturn(true, false);
          rpcClientFactoryMockedStatic.when(() -> RpcClientFactory.createClient(anyString(), any(ConnectionType.class),
                  any(GrpcClientConfig.class))).thenReturn(rpcClientInner);
          // mock listen and remove listen request
          Mockito.when(rpcClientInner.request(any(ConfigBatchListenRequest.class))).thenReturn(response, response);
          // mock query changed config
          ConfigQueryResponse configQueryResponse = new ConfigQueryResponse();
          configQueryResponse.setContent("content" + System.currentTimeMillis());
          configQueryResponse.setContentType(ConfigType.JSON.getType());
          Mockito.when(rpcClientInner.request(any(ConfigQueryRequest.class))).thenReturn(configQueryResponse);
          (clientWorker.getAgent()).executeConfigListen();
          //assert
          //use local cache.
          assertTrue(cacheUseLocalCache.isUseLocalConfigInfo());
          //discard cache to be deleted.
          assertFalse(cacheMapMocked.get().containsKey(GroupKey.getKeyTenant(dataIdDiscard, group, tenant)));
          //normal cache listener be notified.
          assertEquals(configQueryResponse.getContent(), normalContent.get());
          
      }
      
      private CacheData discardCache(ConfigFilterChainManager filter, String envName, String dataId, String group,
              String tenant) {
          CacheData cacheData = new CacheData(filter, envName, dataId, group, tenant);
          cacheData.setDiscard(true);
          cacheData.setConsistentWithServer(false);
          File file = Mockito.mock(File.class);
          Mockito.when(file.exists()).thenReturn(false);
          localConfigInfoProcessorMockedStatic.when(
                  () -> LocalConfigInfoProcessor.getFailoverFile(envName, dataId, group, tenant)).thenReturn(file);
          return cacheData;
      }
      
      private CacheData normalNotConsistentCache(ConfigFilterChainManager filter, String envName, String dataId,
              String group, String tenant) throws NacosException {
          CacheData cacheData = new CacheData(filter, envName, dataId, group, tenant);
          cacheData.setDiscard(false);
          cacheData.setConsistentWithServer(false);
          File file = Mockito.mock(File.class);
          Mockito.when(file.exists()).thenReturn(false);
          localConfigInfoProcessorMockedStatic.when(
                  () -> LocalConfigInfoProcessor.getFailoverFile(envName, dataId, group, tenant)).thenReturn(file);
          return cacheData;
      }
      
      private CacheData useLocalCache(ConfigFilterChainManager filter, String envName, String dataId, String group,
              String tenant, String failOverContent) {
          CacheData cacheData = new CacheData(filter, envName, dataId, group, tenant);
          cacheData.setDiscard(true);
          File file = Mockito.mock(File.class);
          Mockito.when(file.exists()).thenReturn(true);
          localConfigInfoProcessorMockedStatic.when(
                  () -> LocalConfigInfoProcessor.getFailoverFile(envName, dataId, group, tenant)).thenReturn(file);
          localConfigInfoProcessorMockedStatic.when(
                  () -> LocalConfigInfoProcessor.getFailover(envName, dataId, group, tenant)).thenReturn(failOverContent);
          return cacheData;
      }
      
      @Test
      void testIsHealthServer() throws NacosException, NoSuchFieldException, IllegalAccessException {
          Properties prop = new Properties();
          ConfigFilterChainManager filter = new ConfigFilterChainManager(new Properties());
          ConfigServerListManager agent = Mockito.mock(ConfigServerListManager.class);
          
          final NacosClientProperties nacosClientProperties = NacosClientProperties.PROTOTYPE.derive(prop);
          ClientWorker clientWorker = new ClientWorker(filter, agent, nacosClientProperties);
          ClientWorker.ConfigRpcTransportClient client = Mockito.mock(ClientWorker.ConfigRpcTransportClient.class);
          Mockito.when(client.isHealthServer()).thenReturn(Boolean.TRUE);
          
          Field declaredField = ClientWorker.class.getDeclaredField("agent");
          declaredField.setAccessible(true);
          declaredField.set(clientWorker, client);
          
          assertTrue(clientWorker.isHealthServer());
          
          Mockito.when(client.isHealthServer()).thenReturn(Boolean.FALSE);
          assertFalse(clientWorker.isHealthServer());
      }
      
      @Test
      void testPutCache() throws Exception {
          // 反射调用私有方法putCacheIfAbsent
          Method putCacheMethod = ClientWorker.class.getDeclaredMethod("putCache", String.class, CacheData.class);
          putCacheMethod.setAccessible(true);
          Properties prop = new Properties();
          ConfigFilterChainManager filter = new ConfigFilterChainManager(new Properties());
          ConfigServerListManager agent = Mockito.mock(ConfigServerListManager.class);
          final NacosClientProperties nacosClientProperties = NacosClientProperties.PROTOTYPE.derive(prop);
          ClientWorker clientWorker = new ClientWorker(filter, agent, nacosClientProperties);
          String key = "testKey";
          CacheData cacheData = new CacheData(filter, "env", "dataId", "group");
          putCacheMethod.invoke(clientWorker, key, cacheData);
          Field cacheMapField = ClientWorker.class.getDeclaredField("cacheMap");
          cacheMapField.setAccessible(true);
          AtomicReference<Map<String, CacheData>> cacheMapRef = (AtomicReference<Map<String, CacheData>>) cacheMapField.get(
                  clientWorker);
          // 检查cacheMap是否包含特定的key
          assertNotNull(cacheMapRef.get().get(key));
          assertEquals(cacheData, cacheMapRef.get().get(key));
          // 测试再次插入相同的key将覆盖原始的值
          CacheData newCacheData = new CacheData(filter, "newEnv", "newDataId", "newGroup");
          putCacheMethod.invoke(clientWorker, key, newCacheData);
          // 检查key对应的value是否改变为newCacheData
          assertEquals(newCacheData, cacheMapRef.get().get(key));
      }
      
      @Test
      void testAddListenersEnsureCacheDataSafe() throws NacosException, IllegalAccessException, NoSuchFieldException {
          String dataId = "testDataId";
          String group = "testGroup";
          // 将key-cacheData插入到cacheMap中
          CacheData cacheData = new CacheData(null, "env", dataId, group);
          Field cacheMapField = ClientWorker.class.getDeclaredField("cacheMap");
          cacheMapField.setAccessible(true);
          AtomicReference<Map<String, CacheData>> cacheMapRef = (AtomicReference<Map<String, CacheData>>) cacheMapField.get(
                  clientWorker);
          String key = GroupKey.getKey(dataId, group);
          cacheMapRef.get().put(key, cacheData);
          // 当addCacheDataIfAbsent得到的differentCacheData，同cacheMap中该key对应的cacheData不一致
          CacheData differentCacheData = new CacheData(null, "env", dataId, group);
          doReturn(differentCacheData).when(clientWorkerSpy).addCacheDataIfAbsent(anyString(), anyString());
          // 使用addListeners将differentCacheData插入到cacheMap中
          clientWorkerSpy.addListeners(dataId, group, Collections.EMPTY_LIST);
          CacheData cacheDataFromCache1 = clientWorker.getCache(dataId, group);
          assertNotNull(cacheDataFromCache1);
          assertEquals(cacheDataFromCache1, differentCacheData);
          assertFalse(cacheDataFromCache1.isDiscard());
          assertFalse(cacheDataFromCache1.isConsistentWithServer());
          // 再次调用addListeners，此时addCacheDataIfAbsent得到的cacheData同cacheMap中该key对应的cacheData一致，均为differentCacheData
          clientWorkerSpy.addListeners(dataId, group, Collections.EMPTY_LIST);
          CacheData cacheDataFromCache2 = clientWorker.getCache(dataId, group);
          assertNotNull(cacheDataFromCache2);
          assertEquals(cacheDataFromCache2, differentCacheData);
          assertFalse(cacheDataFromCache2.isDiscard());
          assertFalse(cacheDataFromCache2.isConsistentWithServer());
      }
      
      @Test
      void testAddTenantListenersEnsureCacheDataSafe()
              throws NacosException, IllegalAccessException, NoSuchFieldException {
          String dataId = "testDataId";
          String group = "testGroup";
          // 将key-cacheData插入到cacheMap中
          CacheData cacheData = new CacheData(null, "env", dataId, group);
          Field cacheMapField = ClientWorker.class.getDeclaredField("cacheMap");
          cacheMapField.setAccessible(true);
          AtomicReference<Map<String, CacheData>> cacheMapRef = (AtomicReference<Map<String, CacheData>>) cacheMapField.get(
                  clientWorker);
          String key = GroupKey.getKeyTenant(dataId, group, TEST_NAMESPACE);
          cacheMapRef.get().put(key, cacheData);
          // 当addCacheDataIfAbsent得到的differentCacheData，同cacheMap中该key对应的cacheData不一致
          CacheData differentCacheData = new CacheData(null, "env", dataId, group);
          doReturn(differentCacheData).when(clientWorkerSpy)
                  .addCacheDataIfAbsent(anyString(), anyString(), eq(TEST_NAMESPACE));
          // 使用addListeners将differentCacheData插入到cacheMap中
          clientWorkerSpy.addTenantListeners(dataId, group, Collections.EMPTY_LIST);
          CacheData cacheDataFromCache1 = clientWorker.getCache(dataId, group, TEST_NAMESPACE);
          assertNotNull(cacheDataFromCache1);
          assertEquals(cacheDataFromCache1, differentCacheData);
          assertFalse(cacheDataFromCache1.isDiscard());
          assertFalse(cacheDataFromCache1.isConsistentWithServer());
          // 再次调用addListeners，此时addCacheDataIfAbsent得到的cacheData同cacheMap中该key对应的cacheData一致，均为differentCacheData
          clientWorkerSpy.addTenantListeners(dataId, group, Collections.EMPTY_LIST);
          CacheData cacheDataFromCache2 = clientWorker.getCache(dataId, group, TEST_NAMESPACE);
          assertNotNull(cacheDataFromCache2);
          assertEquals(cacheDataFromCache2, differentCacheData);
          assertFalse(cacheDataFromCache2.isDiscard());
          assertFalse(cacheDataFromCache2.isConsistentWithServer());
      }
      
      @Test
      void testAddTenantListenersWithContentEnsureCacheDataSafe()
              throws NacosException, IllegalAccessException, NoSuchFieldException {
          String dataId = "testDataId";
          String group = "testGroup";
          // 将key-cacheData插入到cacheMap中
          CacheData cacheData = new CacheData(null, "env", dataId, group);
          Field cacheMapField = ClientWorker.class.getDeclaredField("cacheMap");
          cacheMapField.setAccessible(true);
          AtomicReference<Map<String, CacheData>> cacheMapRef = (AtomicReference<Map<String, CacheData>>) cacheMapField.get(
                  clientWorker);
          String key = GroupKey.getKeyTenant(dataId, group, TEST_NAMESPACE);
          cacheMapRef.get().put(key, cacheData);
          // 当addCacheDataIfAbsent得到的differentCacheData，同cacheMap中该key对应的cacheData不一致
          CacheData differentCacheData = new CacheData(null, "env", dataId, group);
          doReturn(differentCacheData).when(clientWorkerSpy)
                  .addCacheDataIfAbsent(anyString(), anyString(), eq(TEST_NAMESPACE));
          // 使用addListeners将differentCacheData插入到cacheMap中
          clientWorkerSpy.addTenantListenersWithContent(dataId, group, "", "", Collections.EMPTY_LIST);
          CacheData cacheDataFromCache1 = clientWorker.getCache(dataId, group, TEST_NAMESPACE);
          assertNotNull(cacheDataFromCache1);
          assertEquals(cacheDataFromCache1, differentCacheData);
          assertFalse(cacheDataFromCache1.isDiscard());
          assertFalse(cacheDataFromCache1.isConsistentWithServer());
          // 再次调用addListeners，此时addCacheDataIfAbsent得到的cacheData同cacheMap中该key对应的cacheData一致，均为differentCacheData
          clientWorkerSpy.addTenantListenersWithContent(dataId, group, "", "", Collections.EMPTY_LIST);
          CacheData cacheDataFromCache2 = clientWorker.getCache(dataId, group, TEST_NAMESPACE);
          assertNotNull(cacheDataFromCache2);
          assertEquals(cacheDataFromCache2, differentCacheData);
          assertFalse(cacheDataFromCache2.isDiscard());
          assertFalse(cacheDataFromCache2.isConsistentWithServer());
      }
      
      @Test
      void testResponse403() throws NacosException {
          Properties prop = new Properties();
          ConfigFilterChainManager filter = new ConfigFilterChainManager(new Properties());
          ConfigServerListManager agent = Mockito.mock(ConfigServerListManager.class);
          
          final NacosClientProperties nacosClientProperties = NacosClientProperties.PROTOTYPE.derive(prop);
          final ClientWorker clientWorker = new ClientWorker(filter, agent, nacosClientProperties);
          
          ConfigRemoveResponse response = ConfigRemoveResponse.buildFailResponse("accessToken invalid");
          response.setErrorCode(ConfigQueryResponse.NO_RIGHT);
          Mockito.when(rpcClient.request(any(ConfigRemoveRequest.class))).thenReturn(response);
          boolean result = clientWorker.removeConfig("a", "b", "c", "tag");
          assertFalse(result);
      }
      
      @Test
      void testRemoveCacheWithMetricsEnabled() throws Exception {
          String dataId = "testDataId";
          String group = "testGroup";
          String tenant = "testTenant";
          
          Properties prop = new Properties();
          prop.put("enableClientMetrics", "true");
          ConfigFilterChainManager filter = new ConfigFilterChainManager(new Properties());
          ConfigServerListManager agent = mock(ConfigServerListManager.class);
          
          final NacosClientProperties nacosClientProperties = NacosClientProperties.PROTOTYPE.derive(prop);
          final ClientWorker clientWorker = new ClientWorker(filter, agent, nacosClientProperties);
          
          Gauge.Child mockGaugeChild = mock(Gauge.Child.class);
          try (MockedStatic<MetricsMonitor> mockedMetricsMonitor = Mockito.mockStatic(MetricsMonitor.class)) {
              mockedMetricsMonitor.when(MetricsMonitor::getListenConfigCountMonitor).thenReturn(mockGaugeChild);
              
              clientWorker.removeCache(dataId, group, tenant);
              
              verify(mockGaugeChild, times(1)).set(0);
          }
      }
      
      @Test
      void testRemoveCacheWithMetricsDisabled() throws Exception {
          String dataId = "testDataId";
          String group = "testGroup";
          String tenant = "testTenant";
          
          Properties prop = new Properties();
          prop.put(PropertyKeyConst.ENABLE_CLIENT_METRICS, "false");
          ConfigFilterChainManager filter = new ConfigFilterChainManager(new Properties());
          ConfigServerListManager agent = mock(ConfigServerListManager.class);
          
          final NacosClientProperties nacosClientProperties = NacosClientProperties.PROTOTYPE.derive(prop);
          final ClientWorker clientWorker = new ClientWorker(filter, agent, nacosClientProperties);
          clientWorkerSpy = Mockito.spy(clientWorker);
          
          Gauge.Child mockGaugeChild = mock(Gauge.Child.class);
          try (MockedStatic<MetricsMonitor> mockedMetricsMonitor = Mockito.mockStatic(MetricsMonitor.class)) {
              mockedMetricsMonitor.when(MetricsMonitor::getListenConfigCountMonitor).thenReturn(mockGaugeChild);
              
              clientWorker.removeCache(dataId, group, tenant);
              
              verify(mockGaugeChild, times(0)).set(0);
          }
      }
      
      @Test
      void testRemoveCacheWithDefaultClientMetricsEnabled() throws Exception {
          String dataId = "testDataId";
          String group = "testGroup";
          String tenant = "testTenant";
          
          Properties prop = new Properties();
          ConfigFilterChainManager filter = new ConfigFilterChainManager(new Properties());
          ConfigServerListManager agent = mock(ConfigServerListManager.class);
          
          final NacosClientProperties nacosClientProperties = NacosClientProperties.PROTOTYPE.derive(prop);
          final ClientWorker clientWorker = new ClientWorker(filter, agent, nacosClientProperties);
          
          Gauge.Child mockGaugeChild = mock(Gauge.Child.class);
          try (MockedStatic<MetricsMonitor> mockedMetricsMonitor = Mockito.mockStatic(MetricsMonitor.class)) {
              mockedMetricsMonitor.when(MetricsMonitor::getListenConfigCountMonitor).thenReturn(mockGaugeChild);
              
              clientWorker.removeCache(dataId, group, tenant);
              
              verify(mockGaugeChild, times(1)).set(0);
          }
      }
      
      @Test
      void testMetricsMonitorSetThrowsException() throws NacosException {
          String dataId = "testDataId";
          String group = "testGroup";
          String tenant = "testTenant";
          
          Properties prop = new Properties();
          prop.put(PropertyKeyConst.ENABLE_CLIENT_METRICS, "true");
          ConfigFilterChainManager filter = new ConfigFilterChainManager(new Properties());
          ConfigServerListManager agent = mock(ConfigServerListManager.class);
          
          final NacosClientProperties nacosClientProperties = NacosClientProperties.PROTOTYPE.derive(prop);
          final ClientWorker clientWorker = new ClientWorker(filter, agent, nacosClientProperties);
          clientWorkerSpy = Mockito.spy(clientWorker);
          
          Gauge.Child mockGaugeChild = mock(Gauge.Child.class);
          try (MockedStatic<MetricsMonitor> mockedMetricsMonitor = Mockito.mockStatic(MetricsMonitor.class)) {
              mockedMetricsMonitor.when(MetricsMonitor::getListenConfigCountMonitor).thenReturn(mockGaugeChild);
              
              RuntimeException exception = new RuntimeException("Mocked exception");
              doThrow(exception).when(mockGaugeChild).set(0);
              
              assertDoesNotThrow(() -> clientWorker.removeCache(dataId, group, tenant));
          }
      }
      
      @Test
      public void testAddCacheDataIfAbsentEnableClientMetricsTrue() throws NacosException {
          String dataId = "testDataId";
          String group = "testGroup";
          String tenant = "testTenant";
          
          Properties prop = new Properties();
          prop.put(PropertyKeyConst.ENABLE_CLIENT_METRICS, "true");
          ConfigFilterChainManager filter = new ConfigFilterChainManager(new Properties());
          ConfigServerListManager agent = mock(ConfigServerListManager.class);
          
          NacosClientProperties nacosClientProperties = NacosClientProperties.PROTOTYPE.derive(prop);
          ClientWorker clientWorker = new ClientWorker(filter, agent, nacosClientProperties);
          
          Gauge.Child mockGaugeChild = mock(Gauge.Child.class);
          try (MockedStatic<MetricsMonitor> mockedMetricsMonitor = Mockito.mockStatic(MetricsMonitor.class)) {
              mockedMetricsMonitor.when(MetricsMonitor::getListenConfigCountMonitor).thenReturn(mockGaugeChild);
              
              clientWorker.addCacheDataIfAbsent(dataId, group, tenant);
              
              verify(mockGaugeChild, times(1)).set(1);
          }
      }
      
      @Test
      public void testAddCacheDataIfAbsentEnableClientMetricsFalse() throws NacosException {
          
          String dataId = "testDataId";
          String group = "testGroup";
          String tenant = "testTenant";
          
          Properties prop = new Properties();
          prop.put(PropertyKeyConst.ENABLE_CLIENT_METRICS, "false");
          ConfigFilterChainManager filter = new ConfigFilterChainManager(new Properties());
          ConfigServerListManager agent = mock(ConfigServerListManager.class);
          
          NacosClientProperties nacosClientProperties = NacosClientProperties.PROTOTYPE.derive(prop);
          ClientWorker clientWorker = new ClientWorker(filter, agent, nacosClientProperties);
          
          try (MockedStatic<MetricsMonitor> mockedMetricsMonitor = Mockito.mockStatic(MetricsMonitor.class)) {
              clientWorker.addCacheDataIfAbsent(dataId, group, tenant);
              
              mockedMetricsMonitor.verify(MetricsMonitor::getListenConfigCountMonitor, never());
          }
      }
      
      @Test
      public void testAddCacheDataIfAbsentEnableClientMetricsNotSet() throws NacosException {
          String dataId = "testDataId";
          String group = "testGroup";
          String tenant = "testTenant";
          
          Properties prop = new Properties();
          ConfigFilterChainManager filter = new ConfigFilterChainManager(new Properties());
          ConfigServerListManager agent = mock(ConfigServerListManager.class);
          
          NacosClientProperties nacosClientProperties = NacosClientProperties.PROTOTYPE.derive(prop);
          ClientWorker clientWorker = new ClientWorker(filter, agent, nacosClientProperties);
          
          Gauge.Child mockGaugeChild = mock(Gauge.Child.class);
          try (MockedStatic<MetricsMonitor> mockedMetricsMonitor = Mockito.mockStatic(MetricsMonitor.class)) {
              mockedMetricsMonitor.when(MetricsMonitor::getListenConfigCountMonitor).thenReturn(mockGaugeChild);
              
              clientWorker.addCacheDataIfAbsent(dataId, group, tenant);
              
              verify(mockGaugeChild, times(1)).set(1);
          }
      }
  }
  