Method test_connectivity

backend/utils/nacos_client.py:516–577

Test connectivity to the Nacos server. Args: namespace: Nacos namespace ID to test connectivity with. Returns: Dict containing: - success: Whether the connection was successful - message: Human-readable message about the result

(
        self,
        namespace: str = "public"
    )

            return False

    async def test_connectivity(
        self,
        namespace: str = "public"
    ) -> Dict[str, Any]:
        """Test connectivity to the Nacos server.

        Args:
            namespace: Nacos namespace ID to test connectivity with.

        Returns:
            Dict containing:
            - success: Whether the connection was successful
            - message: Human-readable message about the result
        """
        try:
            session = await self._get_session()

            access_token = None
            if self.username and self.password:
                access_token = await self._get_access_token(session)
                if not access_token:
                    return {
                        "success": False,
                        "message": "Authentication failed. Please check username and password."
                    }

            url = f"{self.nacos_addr}/nacos/v3/admin/ns/ops/metrics"
            headers = {}
            if access_token:
                headers["AccessToken"] = access_token

            async with session.get(url, headers=headers) as response:
                if response.status == 200:
                    data = await response.json()
                    if data.get("code") == 0:
                        return {
                            "success": True,
                            "message": "Successfully connected to Nacos server"
                        }
                    else:
                        return {
                            "success": False,
                            "message": f"Nacos API error: {data.get('message', 'unknown')}"
                        }
                elif response.status == 403:
                    return {
                        "success": False,
                        "message": "Authentication failed. Please check username and password."
                    }
                else:
                    text = await response.text()
                    return {
                        "success": False,
                        "message": f"Nacos server returned status {response.status}: {text}"
                    }

        except aiohttp.ClientError as e:
            logger.error(f"Failed to connect to Nacos at {self.nacos_addr}: {e}")