refactor(auth): simplify URL resolution logic and enhance test coverage

The _get_openapi_base_url method was refactored to remove complex FC region
logic and simplify URL selection. The implementation now prioritizes
intranet URLs when available, falling back to internet URLs. This change
improves code readability and maintainability while maintaining the same
functional behavior.

Additionally, comprehensive test suites were added for RAM signature
helper functions, ControlAPI client methods, and exception handling to
ensure robust authentication and error management.

测试套件已增加以验证 RAM 签名辅助函数、ControlAPI 客户端方法和异常处理，
确保身份验证和错误管理的健壮性。

Change-Id: Ia9d8ff6d2bfd37ec858413f13686fcee8fd6d912
Signed-off-by: OhYee <oyohyee@oyohyee.com>

Author: OhYee

agentrun/toolset/__toolset_async_template.py

@@ -94,20 +94,13 @@ def _get_openapi_auth_defaults(
         return headers, query
 
     def _get_openapi_base_url(self) -> Optional[str]:
-        import os
-
-        fc_region = os.getenv("FC_REGION")
-        arn = pydash.get(self, "status.outputs.function_arn", "")
-
-        if fc_region and arn and pydash.get(arn.split(":"), "[2]"):
-            # 在同一个 region，则使用内网地址
-            return pydash.get(
-                self,
-                "status.outputs.urls.intranet_url",
-                None,
-            )
+        intranet_url: Optional[str] = pydash.get(
+            self, "status.outputs.urls.intranet_url", None
+        )
+        if intranet_url:
+            return intranet_url
 
-        return None
+        return pydash.get(self, "status.outputs.urls.internet_url", None)
 
     async def get_async(self, config: Optional[Config] = None):
         if self.name is None:

agentrun/toolset/toolset.py

@@ -109,20 +109,13 @@ def _get_openapi_auth_defaults(
         return headers, query
 
     def _get_openapi_base_url(self) -> Optional[str]:
-        import os
-
-        fc_region = os.getenv("FC_REGION")
-        arn = pydash.get(self, "status.outputs.function_arn", "")
-
-        if fc_region and arn and pydash.get(arn.split(":"), "[2]"):
-            # 在同一个 region，则使用内网地址
-            return pydash.get(
-                self,
-                "status.outputs.urls.intranet_url",
-                None,
-            )
+        intranet_url: Optional[str] = pydash.get(
+            self, "status.outputs.urls.intranet_url", None
+        )
+        if intranet_url:
+            return intranet_url
 
-        return None
+        return pydash.get(self, "status.outputs.urls.internet_url", None)
 
     async def get_async(self, config: Optional[Config] = None):
         if self.name is None:

tests/unittests/ram_signature/test_signer.py

@@ -5,7 +5,13 @@
 
 import pytest
 
-from agentrun.ram_signature.python.signer import get_agentrun_signed_headers
+from agentrun.utils.ram_signature.signer import (
+    _canonical_headers,
+    _canonical_uri,
+    _percent_encode,
+    get_agentrun_signed_headers,
+    get_agentrun_signed_headers_with_debug,
+)
 
 
 class TestRamSignatureStandalone:
@@ -290,3 +296,159 @@ def test_scenario_3_post_query_empty_body_content_type_json(self):
         )
         if ref is not None:
             assert sig == ref, "SDK 与官方包(ref) 签名应一致"
+
+
+class TestSignerHelperFunctions:
+    """测试签名辅助函数的边界情况"""
+
+    def test_percent_encode_none(self):
+        """测试 _percent_encode(None) 返回空字符串"""
+        assert _percent_encode(None) == ""
+
+    def test_percent_encode_tilde(self):
+        """测试 _percent_encode 正确处理 ~ 字符"""
+        assert "~" in _percent_encode("a~b")
+
+    def test_canonical_uri_empty(self):
+        """测试 _canonical_uri 空字符串返回 /"""
+        assert _canonical_uri("") == "/"
+
+    def test_canonical_uri_none(self):
+        """测试 _canonical_uri None 返回 /"""
+        assert _canonical_uri(None) == "/"
+
+    def test_canonical_uri_normal(self):
+        """测试 _canonical_uri 正常路径"""
+        assert _canonical_uri("/path/to/resource") == "/path/to/resource"
+
+    def test_canonical_headers_skips_none_values(self):
+        """测试 _canonical_headers 跳过 value 为 None 的 header"""
+        headers = {
+            "host": "example.com",
+            "x-acs-date": "2026-01-01T00:00:00Z",
+            "x-acs-skip": None,
+        }
+        canon, signed = _canonical_headers(headers)
+        assert "x-acs-skip" not in signed
+        assert "host" in signed
+
+
+class TestSignerNaiveDatetime:
+    """测试 naive datetime（无时区信息）的处理"""
+
+    def test_naive_datetime_gets_utc(self):
+        """测试 naive datetime 被自动设置为 UTC"""
+        naive_time = datetime(2026, 1, 1, 12, 0, 0)
+        headers = get_agentrun_signed_headers(
+            url="https://x.agentrun-data.cn-hangzhou.aliyuncs.com/path",
+            access_key_id="ak",
+            access_key_secret="sk",
+            sign_time=naive_time,
+        )
+        assert headers["x-acs-date"] == "2026-01-01T12:00:00Z"
+
+
+class TestSignerWithDebug:
+    """测试 get_agentrun_signed_headers_with_debug 函数"""
+
+    def test_returns_headers_and_debug(self):
+        """测试返回 headers 和 debug 信息"""
+        t = datetime(2026, 1, 1, 12, 0, 0, tzinfo=timezone.utc)
+        headers, debug = get_agentrun_signed_headers_with_debug(
+            url="https://x.agentrun-data.cn-hangzhou.aliyuncs.com/path",
+            access_key_id="ak",
+            access_key_secret="sk",
+            sign_time=t,
+        )
+        assert "Agentrun-Authorization" in headers
+        assert "x-acs-date" in headers
+        assert "canonical_request" in debug
+        assert "string_to_sign" in debug
+        assert "signing_key_hex" in debug
+        assert "signature" in debug
+
+    def test_debug_signature_matches_headers(self):
+        """测试 debug 中的 signature 与 headers 中的一致"""
+        t = datetime(2026, 1, 1, 12, 0, 0, tzinfo=timezone.utc)
+        headers, debug = get_agentrun_signed_headers_with_debug(
+            url="https://x.agentrun-data.cn-hangzhou.aliyuncs.com/path",
+            access_key_id="ak",
+            access_key_secret="sk",
+            sign_time=t,
+        )
+        auth = headers["Agentrun-Authorization"]
+        sig_in_auth = auth.split("Signature=")[-1]
+        assert sig_in_auth == debug["signature"]
+
+    def test_debug_matches_non_debug_version(self):
+        """测试 debug 版本与非 debug 版本签名一致"""
+        t = datetime(2026, 1, 1, 12, 0, 0, tzinfo=timezone.utc)
+        opts = dict(
+            url="https://x.agentrun-data.cn-hangzhou.aliyuncs.com/path?a=1",
+            method="POST",
+            access_key_id="ak",
+            access_key_secret="sk",
+            sign_time=t,
+        )
+        headers_normal = get_agentrun_signed_headers(**opts)
+        headers_debug, _ = get_agentrun_signed_headers_with_debug(**opts)
+        assert (
+            headers_normal["Agentrun-Authorization"]
+            == headers_debug["Agentrun-Authorization"]
+        )
+
+    def test_debug_requires_ak_sk(self):
+        """测试 debug 版本也要求 ak/sk"""
+        with pytest.raises(ValueError, match="Access Key ID and Secret"):
+            get_agentrun_signed_headers_with_debug(
+                url="https://x.agentrun-data.cn-hangzhou.aliyuncs.com/",
+                access_key_id="",
+                access_key_secret="sk",
+            )
+
+    def test_debug_with_security_token(self):
+        """测试 debug 版本带 security_token"""
+        t = datetime(2026, 1, 1, 12, 0, 0, tzinfo=timezone.utc)
+        headers, debug = get_agentrun_signed_headers_with_debug(
+            url="https://x.agentrun-data.cn-hangzhou.aliyuncs.com/path",
+            access_key_id="ak",
+            access_key_secret="sk",
+            security_token="sts-token",
+            sign_time=t,
+        )
+        assert "x-acs-security-token" in headers
+        assert "x-acs-security-token" in headers["Agentrun-Authorization"]
+
+    def test_debug_with_content_type(self):
+        """测试 debug 版本带 content_type"""
+        t = datetime(2026, 1, 1, 12, 0, 0, tzinfo=timezone.utc)
+        headers, debug = get_agentrun_signed_headers_with_debug(
+            url="https://x.agentrun-data.cn-hangzhou.aliyuncs.com/path",
+            access_key_id="ak",
+            access_key_secret="sk",
+            content_type="application/json",
+            sign_time=t,
+        )
+        assert "content-type" in headers["Agentrun-Authorization"]
+
+    def test_debug_with_query_params(self):
+        """测试 debug 版本带 query 参数"""
+        t = datetime(2026, 1, 1, 12, 0, 0, tzinfo=timezone.utc)
+        headers, debug = get_agentrun_signed_headers_with_debug(
+            url="https://x.agentrun-data.cn-hangzhou.aliyuncs.com/path?foo=bar&zoo=",
+            access_key_id="ak",
+            access_key_secret="sk",
+            sign_time=t,
+        )
+        assert "foo=bar" in debug["canonical_request"]
+
+    def test_debug_naive_datetime(self):
+        """测试 debug 版本处理 naive datetime"""
+        naive_time = datetime(2026, 1, 1, 12, 0, 0)
+        headers, debug = get_agentrun_signed_headers_with_debug(
+            url="https://x.agentrun-data.cn-hangzhou.aliyuncs.com/path",
+            access_key_id="ak",
+            access_key_secret="sk",
+            sign_time=naive_time,
+        )
+        assert headers["x-acs-date"] == "2026-01-01T12:00:00Z"

tests/unittests/utils/test_control_api.py

@@ -277,3 +277,139 @@ def test_get_devs_client_with_read_timeout(self, mock_client_class):
         config_arg = call_args[0][0]
         assert config_arg.connect_timeout == 300
         assert config_arg.read_timeout == 60000
+
+
+class TestControlAPIGetBailianClient:
+    """测试 ControlAPI._get_bailian_client"""
+
+    @patch("agentrun.utils.control_api.BailianClient")
+    def test_get_bailian_client_basic(self, mock_client_class):
+        """测试获取基本百炼客户端"""
+        config = Config(
+            access_key_id="ak",
+            access_key_secret="sk",
+            region_id="cn-hangzhou",
+        )
+        api = ControlAPI(config=config)
+
+        mock_client = MagicMock()
+        mock_client_class.return_value = mock_client
+
+        result = api._get_bailian_client()
+
+        assert mock_client_class.called
+        call_args = mock_client_class.call_args
+        config_arg = call_args[0][0]
+        assert config_arg.access_key_id == "ak"
+        assert config_arg.access_key_secret == "sk"
+        assert config_arg.region_id == "cn-hangzhou"
+
+    @patch("agentrun.utils.control_api.BailianClient")
+    def test_get_bailian_client_strips_https_prefix(self, mock_client_class):
+        """测试获取百炼客户端时去除 https:// 前缀"""
+        config = Config(
+            access_key_id="ak",
+            access_key_secret="sk",
+            bailian_endpoint="https://bailian.cn-hangzhou.aliyuncs.com",
+        )
+        api = ControlAPI(config=config)
+
+        mock_client = MagicMock()
+        mock_client_class.return_value = mock_client
+
+        api._get_bailian_client()
+
+        call_args = mock_client_class.call_args
+        config_arg = call_args[0][0]
+        assert config_arg.endpoint == "bailian.cn-hangzhou.aliyuncs.com"
+
+    @patch("agentrun.utils.control_api.BailianClient")
+    def test_get_bailian_client_strips_http_prefix(self, mock_client_class):
+        """测试获取百炼客户端时去除 http:// 前缀"""
+        config = Config(
+            access_key_id="ak",
+            access_key_secret="sk",
+            bailian_endpoint="http://bailian.custom.com",
+        )
+        api = ControlAPI(config=config)
+
+        mock_client = MagicMock()
+        mock_client_class.return_value = mock_client
+
+        api._get_bailian_client()
+
+        call_args = mock_client_class.call_args
+        config_arg = call_args[0][0]
+        assert config_arg.endpoint == "bailian.custom.com"
+
+
+class TestControlAPIGetGPDBClient:
+    """测试 ControlAPI._get_gpdb_client"""
+
+    @patch("agentrun.utils.control_api.GPDBClient")
+    def test_get_gpdb_client_known_region(self, mock_client_class):
+        """测试已知 region 使用通用 endpoint"""
+        config = Config(
+            access_key_id="ak",
+            access_key_secret="sk",
+            region_id="cn-hangzhou",
+        )
+        api = ControlAPI(config=config)
+
+        mock_client = MagicMock()
+        mock_client_class.return_value = mock_client
+
+        api._get_gpdb_client()
+
+        call_args = mock_client_class.call_args
+        config_arg = call_args[0][0]
+        assert config_arg.endpoint == "gpdb.aliyuncs.com"
+
+    @patch("agentrun.utils.control_api.GPDBClient")
+    def test_get_gpdb_client_unknown_region(self, mock_client_class):
+        """测试未知 region 使用区域级别 endpoint"""
+        config = Config(
+            access_key_id="ak",
+            access_key_secret="sk",
+            region_id="us-west-1",
+        )
+        api = ControlAPI(config=config)
+
+        mock_client = MagicMock()
+        mock_client_class.return_value = mock_client
+
+        api._get_gpdb_client()
+
+        call_args = mock_client_class.call_args
+        config_arg = call_args[0][0]
+        assert config_arg.endpoint == "gpdb.us-west-1.aliyuncs.com"
+
+    @patch("agentrun.utils.control_api.GPDBClient")
+    def test_get_gpdb_client_all_known_regions(self, mock_client_class):
+        """测试所有已知 region 使用通用 endpoint"""
+        known_regions = [
+            "cn-beijing",
+            "cn-hangzhou",
+            "cn-shanghai",
+            "cn-shenzhen",
+            "cn-hongkong",
+            "ap-southeast-1",
+        ]
+        for region in known_regions:
+            config = Config(
+                access_key_id="ak",
+                access_key_secret="sk",
+                region_id=region,
+            )
+            api = ControlAPI(config=config)
+
+            mock_client = MagicMock()
+            mock_client_class.return_value = mock_client
+
+            api._get_gpdb_client()
+
+            call_args = mock_client_class.call_args
+            config_arg = call_args[0][0]
+            assert (
+                config_arg.endpoint == "gpdb.aliyuncs.com"
+            ), f"Region {region} should use gpdb.aliyuncs.com"