chore: Add more test (#48)

- Increase coverage for existing modules

cozepy/conversations/__init__.py

@@ -34,7 +34,7 @@ def create(self, *, messages: List[Message] = None, meta_data: Dict[str, str] =
         """
         url = f"{self._base_url}/v1/conversation/create"
         body = {
-            "messages": [i.model_dump() for i in messages] if len(messages) > 0 else [],
+            "messages": [i.model_dump() for i in messages] if messages and len(messages) > 0 else [],
             "meta_data": meta_data,
         }
         return self._requester.request("post", url, Conversation, body=body)

cozepy/request.py

@@ -94,12 +94,6 @@ def request(
                 return None
             return model.model_validate(data)
 
-    async def arequest(self, method: str, path: str, **kwargs) -> dict:
-        """
-        Send a request to the server with asyncio.
-        """
-        pass
-
     def _make_request(
         self,
         method: str,

tests/test_auth.py

@@ -7,6 +7,7 @@
     COZE_COM_BASE_URL,
     DeviceAuthCode,
     DeviceOAuthApp,
+    JWTAuth,
     JWTOAuthApp,
     OAuthToken,
     PKCEOAuthApp,
@@ -69,6 +70,21 @@ def test_refresh_token(self, respx_mock):
 
 @pytest.mark.respx(base_url="https://api.coze.com")
 class TestJWTOAuthApp:
+    def test_jwt_auth(self, respx_mock):
+        private_key = read_file("testdata/private_key.pem")
+        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()) + 100).model_dump_json()
+            )
+        )
+
+        auth = JWTAuth("client id", private_key, "public key id")
+
+        assert "Bearer" == auth.token_type
+        assert mock_token == auth.token
+        assert mock_token == auth.token  # get from cache
+
     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")
@@ -187,6 +203,21 @@ def test_get_access_token(self, respx_mock):
 
         assert token.access_token == mock_token
 
+    def test_get_access_token_poll(self, respx_mock):
+        app = DeviceOAuthApp("client id")
+        mock_token = random_hex(20)
+
+        respx_mock.post("/api/permission/oauth2/token").mock(
+            httpx.Response(200, json={"error_code": "authorization_pending"})
+        ).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", True)
+
+        assert token.access_token == mock_token
+
     def test_refresh_token(self, respx_mock):
         app = DeviceOAuthApp("client id")
         mock_token = random_hex(20)

tests/test_chat.py

@@ -0,0 +1,138 @@
+import httpx
+import pytest
+
+from cozepy import Chat, ChatEvent, ChatEventType, ChatStatus, Coze, TokenAuth
+
+chat_testdata = Chat(
+    id="id",
+    conversation_id="conversation_id",
+    bot_id="bot_id",
+    created_at=123,
+    completed_at=123,
+    failed_at=123,
+    meta_data={},
+    status=ChatStatus.FAILED,
+)
+
+chat_stream_testdata = """
+event:conversation.chat.created
+data:{"id":"7382159487131697202","conversation_id":"7381473525342978089","bot_id":"7379462189365198898","completed_at":1718792949,"last_error":{"code":0,"msg":""},"status":"created","usage":{"token_count":0,"output_count":0,"input_count":0}}
+
+event:conversation.chat.in_progress
+data:{"id":"7382159487131697202","conversation_id":"7381473525342978089","bot_id":"7379462189365198898","completed_at":1718792949,"last_error":{"code":0,"msg":""},"status":"in_progress","usage":{"token_count":0,"output_count":0,"input_count":0}}
+
+event:conversation.message.delta
+data:{"id":"7382159494123470858","conversation_id":"7381473525342978089","bot_id":"7379462189365198898","role":"assistant","type":"answer","content":"2","content_type":"text","chat_id":"7382159487131697202"}
+
+event:conversation.message.delta
+data:{"id":"7382159494123470858","conversation_id":"7381473525342978089","bot_id":"7379462189365198898","role":"assistant","type":"answer","content":"0","content_type":"text","chat_id":"7382159487131697202"}
+
+event:conversation.message.delta
+data:{"id":"7382159494123470858","conversation_id":"7381473525342978089","bot_id":"7379462189365198898","role":"assistant","type":"answer","content":"星期三","content_type":"text","chat_id":"7382159487131697202"}
+
+event:conversation.message.delta
+data:{"id":"7382159494123470858","conversation_id":"7381473525342978089","bot_id":"7379462189365198898","role":"assistant","type":"answer","content":"。","content_type":"text","chat_id":"7382159487131697202"}
+
+event:conversation.message.completed
+data:{"id":"7382159494123470858","conversation_id":"7381473525342978089","bot_id":"7379462189365198898","role":"assistant","type":"answer","content":"2024 年 10 月 1 日是星期三。","content_type":"text","chat_id":"7382159487131697202"}
+
+event:conversation.message.completed
+data:{"id":"7382159494123552778","conversation_id":"7381473525342978089","bot_id":"7379462189365198898","role":"assistant","type":"verbose","content":"{\\"msg_type\\":\\"generate_answer_finish\\",\\"data\\":\\"\\",\\"from_module\\":null,\\"from_unit\\":null}","content_type":"text","chat_id":"7382159487131697202"}
+
+event:conversation.chat.completed
+data:{"id":"7382159487131697202","conversation_id":"7381473525342978089","bot_id":"7379462189365198898","completed_at":1718792949,"last_error":{"code":0,"msg":""},"status":"completed","usage":{"token_count":633,"output_count":19,"input_count":614}}
+
+event:done
+data:"[DONE]"
+        """
+
+
+@pytest.mark.respx(base_url="https://api.coze.com")
+class TestConversationMessage:
+    def test_create(self, respx_mock):
+        coze = Coze(auth=TokenAuth(token="token"))
+
+        respx_mock.post("/v3/chat").mock(httpx.Response(200, json={"data": chat_testdata.model_dump()}))
+        res = coze.chat.create(bot_id="bot", user_id="user")
+
+        assert res
+        assert res.conversation_id == chat_testdata.conversation_id
+
+    def test_stream(self, respx_mock):
+        coze = Coze(auth=TokenAuth(token="token"))
+
+        respx_mock.post("/v3/chat").mock(httpx.Response(200, content=chat_stream_testdata))
+        events = list(coze.chat.stream(bot_id="bot", user_id="user"))
+
+        assert events
+        assert len(events) == 9
+        assert events[0] == ChatEvent(
+            event=ChatEventType.CONVERSATION_CHAT_CREATED,
+            chat=Chat(
+                id="7382159487131697202",
+                conversation_id="7381473525342978089",
+                bot_id="7379462189365198898",
+                created_at=None,
+                completed_at=1718792949,
+                failed_at=None,
+                meta_data=None,
+                status=ChatStatus.CREATED,
+            ),
+        )
+        assert events[len(events) - 1].event == ChatEventType.CONVERSATION_CHAT_COMPLETED
+
+    def test_retrieve(self, respx_mock):
+        coze = Coze(auth=TokenAuth(token="token"))
+
+        respx_mock.post("/v3/chat/retrieve").mock(httpx.Response(200, json={"data": chat_testdata.model_dump()}))
+        res = coze.chat.retrieve(conversation_id="conversation", chat_id="chat")
+
+        assert res
+        assert res.conversation_id == chat_testdata.conversation_id
+
+    def test_submit_tool_outputs_not_stream(self, respx_mock):
+        coze = Coze(auth=TokenAuth(token="token"))
+
+        respx_mock.post("/v3/chat/submit_tool_outputs").mock(
+            httpx.Response(200, json={"data": chat_testdata.model_dump()})
+        )
+        res = coze.chat.submit_tool_outputs(
+            conversation_id="conversation", chat_id="chat", tool_outputs=[], stream=False
+        )
+
+        assert res
+        assert res.conversation_id == chat_testdata.conversation_id
+
+    def test_submit_tool_outputs_stream(self, respx_mock):
+        coze = Coze(auth=TokenAuth(token="token"))
+
+        respx_mock.post("/v3/chat/submit_tool_outputs").mock(httpx.Response(200, content=chat_stream_testdata))
+        events = list(
+            coze.chat.submit_tool_outputs(conversation_id="conversation", chat_id="chat", tool_outputs=[], stream=True)
+        )
+
+        assert events
+        assert len(events) == 9
+        assert events[0] == ChatEvent(
+            event=ChatEventType.CONVERSATION_CHAT_CREATED,
+            chat=Chat(
+                id="7382159487131697202",
+                conversation_id="7381473525342978089",
+                bot_id="7379462189365198898",
+                created_at=None,
+                completed_at=1718792949,
+                failed_at=None,
+                meta_data=None,
+                status=ChatStatus.CREATED,
+            ),
+        )
+        assert events[len(events) - 1].event == ChatEventType.CONVERSATION_CHAT_COMPLETED
+
+    def test_cancel(self, respx_mock):
+        coze = Coze(auth=TokenAuth(token="token"))
+
+        respx_mock.post("/v3/chat/cancel").mock(httpx.Response(200, json={"data": chat_testdata.model_dump()}))
+        res = coze.chat.cancel(conversation_id="conversation", chat_id="chat")
+
+        assert res
+        assert res.conversation_id == chat_testdata.conversation_id

tests/test_chat_message.py

@@ -0,0 +1,23 @@
+import httpx
+import pytest
+
+from cozepy import Coze, Message, TokenAuth
+
+
+@pytest.mark.respx(base_url="https://api.coze.com")
+class TestChatMessage:
+    def test_create(self, respx_mock):
+        coze = Coze(auth=TokenAuth(token="token"))
+
+        msg = Message.user_text_message("hi")
+        msg2 = Message.user_text_message("hey")
+        respx_mock.post("/v3/chat/message/list").mock(
+            httpx.Response(
+                200,
+                json={"data": [msg.model_dump(), msg2.model_dump()]},
+            )
+        )
+
+        message_list = coze.chat.messages.list(conversation_id="conversation id", chat_id="chat id")
+        assert message_list
+        assert message_list[0].content == msg.content

tests/test_conversation.py

@@ -0,0 +1,27 @@
+import httpx
+import pytest
+
+from cozepy import Conversation, Coze, TokenAuth
+
+
+@pytest.mark.respx(base_url="https://api.coze.com")
+class TestConversation:
+    def test_create(self, respx_mock):
+        coze = Coze(auth=TokenAuth(token="token"))
+
+        conversation = Conversation(id="id", created_at=1, meta_data={})
+        respx_mock.post("/v1/conversation/create").mock(httpx.Response(200, json={"data": conversation.model_dump()}))
+
+        res = coze.conversations.create()
+        assert res
+        assert res.id == conversation.id
+
+    def test_retrieve(self, respx_mock):
+        coze = Coze(auth=TokenAuth(token="token"))
+
+        conversation = Conversation(id="id", created_at=1, meta_data={})
+        respx_mock.get("/v1/conversation/retrieve").mock(httpx.Response(200, json={"data": conversation.model_dump()}))
+
+        res = coze.conversations.retrieve(conversation_id="id")
+        assert res
+        assert res.id == conversation.id

tests/test_conversation_message.py

@@ -9,7 +9,7 @@ class TestConversationMessage:
     def test_create(self, respx_mock):
         coze = Coze(auth=TokenAuth(token="token"))
 
-        msg = Message.user_text_message("hi")
+        msg = Message.assistant_text_message("hi")
         respx_mock.post("/v1/conversation/message/create").mock(httpx.Response(200, json={"data": msg.model_dump()}))
 
         message = coze.conversations.messages.create(

tests/test_exception.py

@@ -1,7 +1,7 @@
-from cozepy import CozeAPIError
+from cozepy import CozeAPIError, CozeEventError, CozePKCEAuthError
 
 
-def test_coze_api_error():
+def test_coze_error():
     err = CozeAPIError(1, "msg", "logid")
     assert err.code == 1
     assert err.msg == "msg"
@@ -13,3 +13,12 @@ def test_coze_api_error():
     assert err.msg == "msg"
     assert err.logid == "logid"
     assert str(err) == "msg: msg, logid: logid"
+
+    err = CozePKCEAuthError("authorization_pending")
+    assert err.error == "authorization_pending"
+
+    err = CozeEventError("event", "xxx")
+    assert str(err) == "invalid event, field: event, data: xxx"
+
+    err = CozeEventError("", "xxx")
+    assert str(err) == "invalid event, data: xxx"

tests/test_knowledge_documents.py

@@ -15,37 +15,35 @@
     TokenAuth,
 )
 
+document_testdata = Document.model_validate(
+    {
+        "document_id": "str",
+        "char_count": 1,
+        "create_time": 1,
+        "update_time": 1,
+        "format_type": DocumentFormatType.DOCUMENT,
+        "hit_count": 1,
+        "name": "str",
+        "size": 1,
+        "slice_count": 1,
+        "source_type": DocumentSourceType.LOCAL_FILE,
+        "status": DocumentStatus.FAILED,
+        "type": "str",
+        "update_interval": 1,
+        "update_type": DocumentUpdateType.AUTO_UPDATE,
+    }
+)
+
 
 @pytest.mark.respx(base_url="https://api.coze.com")
 class TestKnowledgeDocuments:
-    def test_create(self, respx_mock):
+    def test_create_web_auto_update(self, respx_mock):
         coze = Coze(auth=TokenAuth(token="token"))
 
         respx_mock.post("/open_api/knowledge/document/create").mock(
             httpx.Response(
                 200,
-                json={
-                    "document_infos": [
-                        Document.model_validate(
-                            {
-                                "document_id": "str",
-                                "char_count": 1,
-                                "create_time": 1,
-                                "update_time": 1,
-                                "format_type": DocumentFormatType.DOCUMENT,
-                                "hit_count": 1,
-                                "name": "str",
-                                "size": 1,
-                                "slice_count": 1,
-                                "source_type": DocumentSourceType.LOCAL_FILE,
-                                "status": DocumentStatus.FAILED,
-                                "type": "str",
-                                "update_interval": 1,
-                                "update_type": DocumentUpdateType.AUTO_UPDATE,
-                            }
-                        ).model_dump()
-                    ]
-                },
+                json={"document_infos": [document_testdata.model_dump()]},
             )
         )
 
@@ -63,6 +61,30 @@ def test_create(self, respx_mock):
         assert documents
         assert len(documents) == 1
 
+    def test_create_local_custom(self, respx_mock):
+        coze = Coze(auth=TokenAuth(token="token"))
+
+        respx_mock.post("/open_api/knowledge/document/create").mock(
+            httpx.Response(
+                200,
+                json={"document_infos": [document_testdata.model_dump()]},
+            )
+        )
+
+        documents = coze.knowledge.documents.create(
+            dataset_id="dataset_id",
+            document_bases=[
+                DocumentBase(
+                    name="name",
+                    source_info=DocumentSourceInfo.from_local_file("content"),
+                    update_rule=DocumentUpdateRule.no_auto_update(),
+                ),
+            ],
+            chunk_strategy=DocumentChunkStrategy.custom(1, ",", False, True),
+        )
+        assert documents
+        assert len(documents) == 1
+
     def test_update(self, respx_mock):
         coze = Coze(auth=TokenAuth(token="token"))
 

tests/test_log.py

@@ -0,0 +1,12 @@
+import logging
+
+import pytest
+
+from cozepy import setup_logging
+
+
+def test_log():
+    with pytest.raises(ValueError):
+        setup_logging(123)
+
+    setup_logging(logging.DEBUG)