Unit test part 1

tests/unit/common/test_build_req_inter.py

@@ -0,0 +1,36 @@
+import pytest
+
+from tencentcloud.common import abstract_client
+from tencentcloud.common.exception import TencentCloudSDKException
+
+
+def test_build_req_inter_skip_sign(mocker):
+    options = {'SkipSign': True}
+    client = abstract_client.AbstractClient(None, None, None)
+    mock_method = mocker.patch.object(client, '_build_req_without_signature')
+    client._build_req_inter('action', {'param': 'value'}, 'req_inter', options)
+    assert mock_method.called
+
+
+def test_build_req_inter_tc3_signature(mocker):
+    client = abstract_client.AbstractClient(None, None, None)
+    mock_method = mocker.patch.object(client, '_build_req_with_tc3_signature')
+    client._build_req_inter('action', {'param': 'value'}, 'req_inter')
+    assert mock_method.called
+
+
+def test_build_req_inter_old_signature(mocker):
+    client = abstract_client.AbstractClient(None, None, None)
+    client.profile.signMethod = "HmacSHA1"
+    mock_method = mocker.patch.object(client, '_build_req_with_old_signature')
+    client._build_req_inter('action', {'param': 'value'}, 'req_inter')
+    assert mock_method.called
+
+
+def test_build_req_inter_invalid_signature_method():
+    client = abstract_client.AbstractClient(None, None, None)
+    client.profile.signMethod = "InvalidMethod"
+    with pytest.raises(TencentCloudSDKException) as context:
+        client._build_req_inter('action', {'param': 'value'}, 'req_inter')
+    assert context.value.code == "ClientError"
+    assert context.value.message == "Invalid signature method."

tests/unit/common/test_build_req_with_old_signature_inter.py

@@ -0,0 +1,41 @@
+import os
+
+from tencentcloud.common import abstract_client, credential
+from tencentcloud.common.profile.client_profile import ClientProfile
+from tencentcloud.common.profile.http_profile import HttpProfile
+
+
+class Request:
+    def __init__(self, data, header):
+        self.data = data
+        self.header = header
+
+
+def test_build_req_with_old_signature_inter_correct(mocker):
+    cred = credential.Credential(
+        os.environ.get("TENCENTCLOUD_SECRET_ID"),
+        os.environ.get("TENCENTCLOUD_SECRET_KEY"))
+    httpProfile = HttpProfile()
+    httpProfile.endpoint = "sts.tencentcloudapi.com"
+
+    clientProfile = ClientProfile()
+    clientProfile.httpProfile = httpProfile
+    clientProfile.signMethod = "HmacSHA256"
+
+    client = abstract_client.AbstractClient(cred, None, profile=clientProfile)
+
+    req = Request({
+        "data": {
+            "name": "张三",
+            "age": 30,
+            "email": "[email protected]"
+        }
+    },
+        {
+            "header": {
+                "Content-Type": "application/json"
+            }})
+    with mocker.patch('time.time', return_value=1634567890), mocker.patch('random.randint', return_value=123456):
+        """设置mock对象，固定时间戳和随机数"""
+        client._build_req_with_old_signature('action', {}, req)
+        assert req.header["Content-Type"] == "application/x-www-form-urlencoded"

tests/unit/common/test_build_req_without_signature.py

@@ -0,0 +1,122 @@
+import copy
+import json
+import os
+from unittest.mock import MagicMock, patch
+from tencentcloud.common import credential, abstract_client
+from urllib.parse import urlencode
+from tencentcloud.common.profile.client_profile import ClientProfile
+from tencentcloud.common.profile.http_profile import HttpProfile
+
+_default_content_type = "application/json"
+_form_urlencoded_content = "application/x-www-form-urlencoded"
+_json_content = "application/json"
+_multipart_content = "multipart/form-data"
+_octet_stream = "application/octet-stream"
+
+
+def get_client():
+    cred = credential.Credential(
+        os.environ.get("TENCENTCLOUD_SECRET_ID"),
+        os.environ.get("TENCENTCLOUD_SECRET_KEY"))
+    httpProfile = HttpProfile()
+    httpProfile.endpoint = "sts.tencentcloudapi.com"
+
+    clientProfile = ClientProfile()
+    clientProfile.httpProfile = httpProfile
+    clientProfile.signMethod = "HmacSHA256"
+
+    client = abstract_client.AbstractClient(cred, "ap-shanghai", profile=clientProfile)
+    return client
+
+
+class Request:
+    def __init__(self, data, header, method):
+        self.data = data
+        self.header = header
+        self.method = method
+
+
+def test_build_req_without_signature_get_method():
+    """测试 GET 请求"""
+    client = get_client()
+    req = Request({}, {}, "GET")
+    action = 'DescribeServices'
+    params = {'service': 'cvm'}
+    options = None
+    client._build_req_without_signature(action, params, req, options)
+    print(req.header)
+    assert req.header["Content-Type"] == _form_urlencoded_content
+    assert req.data == urlencode(copy.deepcopy(client._fix_params(params)))
+
+
+def test_build_req_without_signature_post_method():
+    """测试 POST 请求"""
+    client = get_client()
+    req = Request({}, {}, "GET")
+    action = 'DescribeServices'
+    params = {'service': 'cvm'}
+    options = None
+    req.method = 'POST'
+    client._build_req_without_signature(action, params, req, options)
+    assert req.header["Content-Type"] == _json_content
+    assert req.data == json.dumps(params)
+
+
+def test_build_req_without_signature_multipart_method(mocker):
+    """测试 multipart格式"""
+    client = get_client()
+    req = Request({}, {}, "GET")
+    action = 'DescribeServices'
+    params = {'service': 'cvm'}
+    options = {"IsMultipart": True}
+    req.method = 'POST'
+    with mocker.patch('uuid.uuid4', return_value=MagicMock(hex="123456")):
+        client._build_req_without_signature(action, params, req, options)
+        expected_content_type = f"{_multipart_content}; boundary=123456"
+        assert req.header["Content-Type"] == expected_content_type
+
+
+def test_build_req_without_signature_octet_stream_method():
+    """测试 octet-stream"""
+    client = get_client()
+    req = Request({}, {}, "POST")
+    action = 'DescribeServices'
+    params = {'service': 'cvm'}
+    options = {"IsOctetStream": True}
+    client._build_req_without_signature(action, params, req, options)
+    assert req.header["Content-Type"] == _octet_stream
+
+
+def test_build_req_without_signature_unsigned_payload():
+    """测试无签名"""
+    client = get_client()
+    req = Request({}, {}, "POST")
+    action = 'DescribeServices'
+    params = {'service': 'cvm'}
+    options = None
+    client.profile.unsignedPayload = True
+    client._build_req_without_signature(action, params, req, options)
+    assert req.header["X-TC-Content-SHA256"] == "UNSIGNED-PAYLOAD"
+
+
+def test_build_req_without_signature_region():
+    """测试 region"""
+    client = get_client()
+    req = Request({}, {}, "POST")
+    action = 'DescribeServices'
+    params = {'service': 'cvm'}
+    options = None
+    client._build_req_without_signature(action, params, req, options)
+    assert req.header['X-TC-Region'] == client.region
+
+
+def test_build_req_without_signature_language():
+    """测试所选语言"""
+    client = get_client()
+    req = Request({}, {}, "POST")
+    action = 'DescribeServices'
+    params = {'service': 'cvm'}
+    options = None
+    client.profile.language = 'zh-CN'
+    client._build_req_without_signature(action, params, req, options)
+    assert req.header['X-TC-Language'] == client.profile.language

tests/unit/common/test_call_with_region_breaker.py

@@ -0,0 +1,49 @@
+import os
+
+
+from tencentcloud.common import abstract_client, credential
+from tencentcloud.common.profile.client_profile import ClientProfile
+from tencentcloud.common.profile.http_profile import HttpProfile
+
+
+def test_normal_request(mocker):
+    cred = credential.Credential(
+        os.environ.get("TENCENTCLOUD_SECRET_ID"),
+        os.environ.get("TENCENTCLOUD_SECRET_KEY"))
+    httpProfile = HttpProfile()
+    httpProfile.endpoint = "sts.tencentcloudapi.com"
+
+    clientProfile = ClientProfile()
+    clientProfile.httpProfile = httpProfile
+    clientProfile.signMethod = "HmacSHA256"
+    clientProfile.disable_region_breaker = False
+
+    client = abstract_client.AbstractClient(cred, None, profile=clientProfile)
+    action = "DescribeServices"
+    params = {"service": "cvm"}
+    mock_request = mocker.patch.object(client, 'request')
+    client._call_with_region_breaker(action, params)
+    mock_request.send_request.assert_called_once()
+
+
+def test_circuit_breaker_open(mocker):
+    cred = credential.Credential(
+        os.environ.get("TENCENTCLOUD_SECRET_ID"),
+        os.environ.get("TENCENTCLOUD_SECRET_KEY"))
+    httpProfile = HttpProfile()
+    httpProfile.endpoint = "sts.tencentcloudapi.com"
+
+    clientProfile = ClientProfile()
+    clientProfile.httpProfile = httpProfile
+    clientProfile.signMethod = "HmacSHA256"
+    clientProfile.disable_region_breaker = False
+
+    client = abstract_client.AbstractClient(cred, None, profile=clientProfile)
+
+    action = "DescribeServices"
+    params = {"service": "cvm"}
+    mocker_circuit_breaker = mocker.patch.object(client, 'circuit_breaker')
+    mocker_circuit_breaker.before_requests.return_value = 1, True
+    client._call_with_region_breaker(action, params)
+    assert ".tencentcloudapi.com" in client._get_endpoint()
+