chore: Add respx dependency and update unit tests to use mocks

.github/workflows/ci.yml

@@ -38,13 +38,6 @@ jobs:
           poetry build
       - name: Run tests
         run: poetry run pytest --cov --cov-report=xml
-        env:
-          COZE_TOKEN: ${{ secrets.COZE_TOKEN }}
-          SPACE_ID_1: ${{ secrets.SPACE_ID_1 }}
-          COZE_JWT_AUTH_CLIENT_ID: ${{ secrets.COZE_JWT_AUTH_CLIENT_ID }}
-          COZE_JWT_AUTH_PRIVATE_KEY: ${{ secrets.COZE_JWT_AUTH_PRIVATE_KEY }}
-          COZE_JWT_AUTH_KEY_ID: ${{ secrets.COZE_JWT_AUTH_KEY_ID }}
-          COZE_BOT_ID_TRANSLATE: ${{ secrets.COZE_BOT_ID_TRANSLATE }}
       - name: Upload coverage to Codecov
         uses: codecov/codecov-action@v3
         with:

cozepy/auth/__init__.py

@@ -35,7 +35,7 @@ class DeviceAuthCode(CozeModel):
     interval: int = 5
     # The expiration time of the device code
     expires_in: int
-    verification_url: str = None
+    verification_url: Optional[str] = None
 
 
 class ScopeAccountPermission(CozeModel):
@@ -315,12 +315,6 @@ def get_device_code(
         """
         Get the pkce flow authorized url.
 
-        :param redirect_uri: The redirect_uri of your app, where authentication responses can be sent and received by
-        your app. It must exactly match one of the redirect URIs you registered in the OAuth Apps.
-        :param state: A value included in the request that is also returned in the token response. It can be a string
-        of any hash value.
-        :param code_verifier:
-        :param code_challenge_method:
         :param workspace_id:
         :return:
         """

cozepy/bots/__init__.py

@@ -1,5 +1,5 @@
 from enum import IntEnum
-from typing import List
+from typing import List, Optional
 
 from cozepy.auth import Auth
 from cozepy.model import CozeModel, NumberPaged
@@ -70,15 +70,15 @@ class Bot(CozeModel):
     # The latest version of the bot.
     version: str = None
     # The prompt configuration for the bot. For more information, see Prompt object.
-    prompt_info: BotPromptInfo = None
+    prompt_info: Optional[BotPromptInfo] = None
     # The onboarding message configuration for the bot. For more information, see Onboarding object.
-    onboarding_info: BotOnboardingInfo = None
+    onboarding_info: Optional[BotOnboardingInfo] = None
     # The mode of the Bot, values: 0: Single Agent mode, 1: Multi Agent mode, 3: Single Agent Workflow mode.
-    bot_mode: BotMode = None
+    bot_mode: Optional[BotMode] = None
     # The plugins configured for the bot. For more information, see  Plugin object.
-    plugin_info_list: List[BotPluginInfo] = None
+    plugin_info_list: Optional[List[BotPluginInfo]] = None
     # The model configured for the bot. For more information, see Model object.
-    model_info: BotModelInfo = None
+    model_info: Optional[BotModelInfo] = None
 
 
 class SimpleBot(CozeModel):

cozepy/chat/__init__.py

@@ -107,12 +107,12 @@ class Message(CozeModel):
     # 自定义键值对，应指定为 Map 对象格式。长度为 16 对键值对，其中键（key）的长度范围为 1～64 个字符，值（value）的长度范围为 1～512 个字符。
     meta_data: Optional[Dict[str, str]] = None
 
-    id: str = None
-    conversation_id: str = None
-    bot_id: str = None
-    chat_id: str = None
-    created_at: int = None
-    updated_at: int = None
+    id: Optional[str] = None
+    conversation_id: Optional[str] = None
+    bot_id: Optional[str] = None
+    chat_id: Optional[str] = None
+    created_at: Optional[int] = None
+    updated_at: Optional[int] = None
 
     @staticmethod
     def user_text_message(content: str, meta_data: Optional[Dict[str, str]] = None) -> "Message":

cozepy/files/__init__.py

@@ -46,7 +46,7 @@ def upload(self, *, file: str) -> File:
         """
         url = f"{self._base_url}/v1/files/upload"
         files = {"file": open(file, "rb")}
-        return self._requester.request("get", url, File, files=files)
+        return self._requester.request("post", url, File, files=files)
 
     def retrieve(self, *, file_id: str):
         """

cozepy/knowledge/documents/__init__.py

@@ -1,5 +1,5 @@
 from enum import IntEnum
-from typing import List
+from typing import List, Optional
 
 from cozepy.auth import Auth
 from cozepy.model import CozeModel, NumberPaged
@@ -124,7 +124,7 @@ class Document(CozeModel):
 
     # The chunking rules. For detailed instructions, refer to the ChunkStrategy object.
     # 分段规则。详细说明可参考 chunk_strategy object。
-    chunk_strategy: DocumentChunkStrategy = None
+    chunk_strategy: Optional[DocumentChunkStrategy] = None
 
     # The upload time of the file, in the format of a 10-digit Unix timestamp.
     # 文件的上传时间，格式为 10 位的 Unixtime 时间戳。
@@ -381,14 +381,14 @@ def list(
             "size": page_size,
         }
         headers = {"Agw-Js-Conv": "str"}
-        res = self._requester.request("post", url, self._PrivateListDocumentsV1Data, body=body, headers=headers)
+        res = self._requester.request("post", url, self._PrivateListDocumentsData, body=body, headers=headers)
         return NumberPaged(
             items=res.document_infos,
             page_num=page_num,
             page_size=page_size,
             total=res.total,
         )
 
-    class _PrivateListDocumentsV1Data(CozeModel):
+    class _PrivateListDocumentsData(CozeModel):
         document_infos: List[Document]
         total: int

cozepy/request.py

@@ -14,7 +14,7 @@
 import httpx
 from httpx import Response
 from pydantic import BaseModel
-from typing_extensions import get_args
+from typing_extensions import get_args, get_origin
 
 from cozepy.config import DEFAULT_CONNECTION_LIMITS, DEFAULT_TIMEOUT
 from cozepy.exception import CozeAPIError, CozePKCEAuthError
@@ -86,7 +86,7 @@ def request(
             if msg in ["authorization_pending", "slow_down", "access_denied", "expired_token"]:
                 raise CozePKCEAuthError(msg, logid)
             raise CozeAPIError(code, msg, logid)
-        if isinstance(model, Iterable):
+        if get_origin(model) is list:
             item_model = get_args(model)[0]
             return [item_model.model_validate(item) for item in data]
         else:

pyproject.toml

@@ -18,6 +18,7 @@ pytest = "^7.0.0"
 pytest-cov = "^4.0.0"
 ruff = "^0.6.0"
 pre-commit = "^2.9.0"
+respx = "^0.21.1"
 
 [tool.ruff]
 line-length = 120

tests/config.py

@@ -1,32 +1,10 @@
 import os
 
-from cozepy import COZE_CN_BASE_URL, JWTAuth, JWTOAuthApp, TokenAuth
 
-COZE_JWT_AUTH_CLIENT_ID = os.getenv("COZE_JWT_AUTH_CLIENT_ID").strip()
-COZE_JWT_AUTH_PRIVATE_KEY = os.getenv("COZE_JWT_AUTH_PRIVATE_KEY").strip()
-COZE_JWT_AUTH_KEY_ID = os.getenv("COZE_JWT_AUTH_KEY_ID").strip()
-if COZE_JWT_AUTH_PRIVATE_KEY == "" and os.getenv("COZE_JWT_AUTH_PRIVATE_KEY_FILE").strip() != "":
-    try:
-        with open(os.getenv("COZE_JWT_AUTH_PRIVATE_KEY_FILE").strip(), "r") as f:
-            COZE_JWT_AUTH_PRIVATE_KEY = f.read()
-    except:  # noqa: E722
-        pass
+def read_file(path: str):
+    current_dir = os.path.dirname(os.path.abspath(__file__))
+    file_path = os.path.join(current_dir, path)
 
-COZE_TOKEN = os.getenv("COZE_TOKEN").strip()
-
-jwt_oauth_app = JWTOAuthApp(
-    COZE_JWT_AUTH_CLIENT_ID,
-    COZE_JWT_AUTH_PRIVATE_KEY,
-    COZE_JWT_AUTH_KEY_ID,
-    base_url=COZE_CN_BASE_URL,
-)
-
-fixed_token_auth = TokenAuth(COZE_TOKEN)
-
-jwt_auth = JWTAuth(
-    COZE_JWT_AUTH_CLIENT_ID,
-    COZE_JWT_AUTH_PRIVATE_KEY,
-    COZE_JWT_AUTH_KEY_ID,
-    30,
-    base_url=COZE_CN_BASE_URL,
-)
+    with open(file_path, "r") as file:
+        content = file.read()
+    return content

tests/test_auth.py

@@ -1,18 +1,154 @@
 import time
 
-from tests.config import (
-    jwt_auth,
-    jwt_oauth_app,
-)
+import httpx
+import pytest
 
+from cozepy import COZE_COM_BASE_URL, DeviceAuthCode, DeviceOAuthApp, JWTOAuthApp, OAuthToken, PKCEOAuthApp, WebOAuthApp
+from cozepy.util import random_hex
+from tests.config import read_file
 
-def test_jwt_oauth_app():
-    token = jwt_oauth_app.get_access_token(30)
-    assert token.access_token != ""
-    assert token.token_type == "Bearer"
-    assert token.expires_in - int(time.time()) <= 31
-    assert token.refresh_token == ""
 
+@pytest.mark.respx(base_url="https://api.coze.com")
+class TestWebOAuthApp:
+    def test_get_oauth_url(self, respx_mock):
+        app = WebOAuthApp("client id", "client secret")
 
-def test_jwt_auth():
-    assert jwt_auth.token != ""
+        url = app.get_oauth_url("https://example.com", "state")
+        assert (
+            "https://www.coze.com/api/permission/oauth2/authorize"
+            "?response_type=code&"
+            "client_id=client+id&"
+            "redirect_uri=https%3A%2F%2Fexample.com&"
+            "state=state"
+        ) == url
+
+    def test_get_access_token(self, respx_mock):
+        app = WebOAuthApp("client id", "client secret")
+        mock_token = random_hex(20)
+
+        respx_mock.post("/api/permission/oauth2/token").mock(
+            httpx.Response(
+                200, content=OAuthToken(access_token=mock_token, expires_in=int(time.time())).model_dump_json()
+            )
+        )
+
+        token = app.get_access_token("https://example.com", "code")
+
+        assert token.access_token == mock_token
+
+    def test_refresh_token(self, respx_mock):
+        app = WebOAuthApp("client id", "client secret")
+        mock_token = random_hex(20)
+        respx_mock.post("/api/permission/oauth2/token").mock(
+            httpx.Response(
+                200, content=OAuthToken(access_token=mock_token, expires_in=int(time.time())).model_dump_json()
+            )
+        )
+
+        token = app.refresh_access_token("refresh token")
+        assert token.access_token == mock_token
+
+
+@pytest.mark.respx(base_url="https://api.coze.com")
+class TestJWTOAuthApp:
+    def test_get_access_token(self, respx_mock):
+        private_key = read_file("testdata/private_key.pem")
+        app = JWTOAuthApp("client id", private_key, "public key id")
+        mock_token = random_hex(20)
+        respx_mock.post("/api/permission/oauth2/token").mock(
+            httpx.Response(
+                200, content=OAuthToken(access_token=mock_token, expires_in=int(time.time())).model_dump_json()
+            )
+        )
+
+        token = app.get_access_token(100)
+        assert token.access_token == mock_token
+
+
+@pytest.mark.respx(base_url="https://api.coze.com")
+class TestPKCEOAuthApp:
+    def test_get_oauth_url(self, respx_mock):
+        app = PKCEOAuthApp("client id")
+
+        url = app.get_oauth_url("https://example.com", "state", "code_verifier", "S256")
+        assert (
+            "https://www.coze.com/api/permission/oauth2/authorize?"
+            "response_type=code&client_id=client+id&"
+            "redirect_uri=https%3A%2F%2Fexample.com&state=state&"
+            "code_challenge=73oehA2tBul5grZPhXUGQwNAjxh69zNES8bu2bVD0EM&code_challenge_method=S256"
+        ) == url
+
+    def test_get_access_token(self, respx_mock):
+        app = PKCEOAuthApp("client id")
+        mock_token = random_hex(20)
+
+        respx_mock.post("/api/permission/oauth2/token").mock(
+            httpx.Response(
+                200, content=OAuthToken(access_token=mock_token, expires_in=int(time.time())).model_dump_json()
+            )
+        )
+
+        token = app.get_access_token("https://example.com", "code", "code_verifier")
+
+        assert token.access_token == mock_token
+
+    def test_refresh_token(self, respx_mock):
+        app = PKCEOAuthApp("client id")
+        mock_token = random_hex(20)
+        respx_mock.post("/api/permission/oauth2/token").mock(
+            httpx.Response(
+                200, content=OAuthToken(access_token=mock_token, expires_in=int(time.time())).model_dump_json()
+            )
+        )
+
+        token = app.refresh_access_token("refresh token")
+        assert token.access_token == mock_token
+
+
+@pytest.mark.respx(base_url="https://api.coze.com")
+class TestDeviceOAuthApp:
+    def test_get_device_code(self, respx_mock):
+        app = DeviceOAuthApp("client id")
+        mock_code = random_hex(20)
+
+        respx_mock.post("/api/permission/oauth2/device/code").mock(
+            httpx.Response(
+                200,
+                content=DeviceAuthCode(
+                    device_code=mock_code,
+                    user_code="user_code",
+                    verification_uri=COZE_COM_BASE_URL,
+                    interval=5,
+                    expires_in=int(time.time()),
+                ).model_dump_json(),
+            )
+        )
+
+        device_code = app.get_device_code()
+        assert device_code.device_code == mock_code
+
+    def test_get_access_token(self, respx_mock):
+        app = DeviceOAuthApp("client id")
+        mock_token = random_hex(20)
+
+        respx_mock.post("/api/permission/oauth2/token").mock(
+            httpx.Response(
+                200, content=OAuthToken(access_token=mock_token, expires_in=int(time.time())).model_dump_json()
+            )
+        )
+
+        token = app.get_access_token("https://example.com", False)
+
+        assert token.access_token == mock_token
+
+    def test_refresh_token(self, respx_mock):
+        app = DeviceOAuthApp("client id")
+        mock_token = random_hex(20)
+        respx_mock.post("/api/permission/oauth2/token").mock(
+            httpx.Response(
+                200, content=OAuthToken(access_token=mock_token, expires_in=int(time.time())).model_dump_json()
+            )
+        )
+
+        token = app.refresh_access_token("refresh token")
+        assert token.access_token == mock_token