Skip to content

Commit

Permalink
chore: Add respx dependency and update unit tests to use mocks
Browse files Browse the repository at this point in the history
  • Loading branch information
chyroc committed Sep 30, 2024
1 parent 19f9840 commit b8d1dbb
Show file tree
Hide file tree
Showing 19 changed files with 588 additions and 289 deletions.
7 changes: 0 additions & 7 deletions .github/workflows/ci.yml
Original file line number Diff line number Diff line change
Expand Up @@ -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:
Expand Down
8 changes: 1 addition & 7 deletions cozepy/auth/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -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):
Expand Down Expand Up @@ -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:
"""
Expand Down
12 changes: 6 additions & 6 deletions cozepy/bots/__init__.py
Original file line number Diff line number Diff line change
@@ -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
Expand Down Expand Up @@ -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):
Expand Down
12 changes: 6 additions & 6 deletions cozepy/chat/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -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":
Expand Down
2 changes: 1 addition & 1 deletion cozepy/files/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -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):
"""
Expand Down
8 changes: 4 additions & 4 deletions cozepy/knowledge/documents/__init__.py
Original file line number Diff line number Diff line change
@@ -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
Expand Down Expand Up @@ -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 时间戳。
Expand Down Expand Up @@ -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
4 changes: 2 additions & 2 deletions cozepy/request.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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:
Expand Down
16 changes: 15 additions & 1 deletion poetry.lock

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

1 change: 1 addition & 0 deletions pyproject.toml
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
34 changes: 6 additions & 28 deletions tests/config.py
Original file line number Diff line number Diff line change
@@ -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
Loading

0 comments on commit b8d1dbb

Please sign in to comment.