chore: Update codecov configuration and add missing unit tests (#49)

- Modified pyproject codecov configuration to skip TYPE_CHECKING
sections from coverage calculation
- Updated codecov configuration to only check coverage for the cozepy
directory
- Added missing unit tests to improve overall test coverage

Author: chyroc

codecov.yml

@@ -5,6 +5,8 @@ coverage:
         target: 70%
         threshold: 70%
         base: auto
+        paths:
+          - cozepy
 comment:                  #this is a top-level key
   layout: "diff, flags, files"
   behavior: default

cozepy/auth/__init__.py

@@ -122,7 +122,7 @@ class WebOAuthApp(OAuthApp):
     Normal OAuth App.
     """
 
-    def __init__(self, client_id: str, client_secret: str, base_url: str = COZE_COM_BASE_URL):
+    def __init__(self, client_id: str, client_secret: str, base_url: str = COZE_COM_BASE_URL, www_base_url: str = ""):
         """
         :param client_id:
         :param client_secret:
@@ -133,7 +133,7 @@ def __init__(self, client_id: str, client_secret: str, base_url: str = COZE_COM_
         self._base_url = base_url
         self._api_endpoint = urlparse(base_url).netloc
         self._token = ""
-        super().__init__(client_id, base_url, www_base_url="")
+        super().__init__(client_id, base_url, www_base_url=www_base_url)
 
     def get_oauth_url(
         self,

cozepy/exception.py

@@ -24,16 +24,17 @@ def __init__(self, code: int = None, msg: str = "", logid: str = None):
             super().__init__(f"msg: {msg}, logid: {logid}")
 
 
-class CozePKCEAuthError(CozeAPIError):
+class CozePKCEAuthError(CozeError):
     """
     base class for all pkce auth errors
     """
 
     def __init__(
         self, error: Literal["authorization_pending", "slow_down", "access_denied", "expired_token"], logid: str = None
     ):
-        super().__init__(code=None, msg=error, logid=logid)
+        super().__init__(f"pkce auth error: {error}")
         self.error = error
+        self.logid = logid
 
 
 class CozeEventError(CozeError):

cozepy/model.py

@@ -20,9 +20,6 @@ def __init__(self, items: List[T], has_more: bool):
         self.items = items
         self.has_more = has_more
 
-    def __repr__(self):
-        return f"PagedBase(items={self.items}, has_more={self.has_more})"
-
 
 class TokenPaged(PagedBase[T]):
     """

pyproject.toml

@@ -27,6 +27,15 @@ line-length = 120
 select = ["E", "F", "I"]
 ignore = ["E501"]
 
+[tool.coverage.report]
+precision = 1
+skip_covered = true
+exclude_lines = [
+    "abc.abstractmethod",
+    "if TYPE_CHECKING.*:",
+    "raise NotImplementedError",
+]
+
 [build-system]
 requires = ["poetry-core"]
 build-backend = "poetry.core.masonry.api"

tests/test_auth.py

@@ -41,6 +41,30 @@ def test_get_oauth_url(self, respx_mock):
             "state=state"
         ) == url
 
+    def test_get_oauth_url_config_www_url(self, respx_mock):
+        app = WebOAuthApp("client id", "client secret", www_base_url="https://example.com")
+
+        url = app.get_oauth_url("https://example.com", "state")
+        assert (
+            "https://example.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_oauth_url_config_custom_api_base_url(self, respx_mock):
+        app = WebOAuthApp("client id", "client secret", base_url="https://api.example.com")
+
+        url = app.get_oauth_url("https://example.com", "state")
+        assert (
+            "https://api.example.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)

tests/test_chat.py

@@ -81,6 +81,36 @@ def test_stream(self, respx_mock):
         )
         assert events[len(events) - 1].event == ChatEventType.CONVERSATION_CHAT_COMPLETED
 
+    def test_stream_error(self, respx_mock):
+        coze = Coze(auth=TokenAuth(token="token"))
+
+        respx_mock.post("/v3/chat").mock(
+            httpx.Response(
+                200,
+                content="""
+event:error
+data:{}
+        """,
+            )
+        )
+        with pytest.raises(Exception, match="error event"):
+            list(coze.chat.stream(bot_id="bot", user_id="user"))
+
+    def test_stream_invalid_event(self, respx_mock):
+        coze = Coze(auth=TokenAuth(token="token"))
+
+        respx_mock.post("/v3/chat").mock(
+            httpx.Response(
+                200,
+                content="""
+event:invalid
+data:{}
+        """,
+            )
+        )
+        with pytest.raises(Exception, match="invalid chat.event: invalid"):
+            list(coze.chat.stream(bot_id="bot", user_id="user"))
+
     def test_retrieve(self, respx_mock):
         coze = Coze(auth=TokenAuth(token="token"))
 

tests/test_model.py

@@ -0,0 +1,36 @@
+from typing import Dict
+
+import pytest
+
+from cozepy import CozeEventError, LastIDPaged, NumberPaged, Stream, TokenPaged
+
+
+def test_page_mode():
+    page = TokenPaged(["a", "b", "c"], "next_page", True)
+    assert f"{page}" == "TokenPaged(items=['a', 'b', 'c'], next_page_token=next_page)"
+
+    page = NumberPaged(["a", "b", "c"], 1, 3, 100)
+    assert f"{page}" == "NumberPaged(items=['a', 'b', 'c'], page_num=1, page_size=3, total=100)"
+
+    page = LastIDPaged(["a", "b", "c"], 1, 3, True)
+    assert f"{page}" == "LastIDPaged(items=['a', 'b', 'c'], first_id=1, last_id=3, has_more=True)"
+
+
+def mock_handler(d: Dict[str, str]):
+    return d
+
+
+def test_stream_invalid_event():
+    items = ["event:x"]
+    s = Stream(iter(items), ["field"], mock_handler)
+
+    with pytest.raises(CozeEventError, match="invalid event, data: event:x"):
+        next(s)
+
+
+def test_stream_invalid_field():
+    items = ["event:x1", "event:x2"]
+    s = Stream(iter(items), ["event", "second"], mock_handler)
+
+    with pytest.raises(CozeEventError, match="invalid event, field: event, data: event:x2"):
+        next(s)

tests/test_request.py

@@ -0,0 +1,68 @@
+import httpx
+import pytest
+
+from cozepy import CozeAPIError, CozePKCEAuthError
+from cozepy.model import CozeModel
+from cozepy.request import Requester
+
+
+class TestModel(CozeModel):
+    id: str
+
+
+class TestDebugModel(CozeModel):
+    debug_url: str
+    data: str
+
+
+@pytest.mark.respx(base_url="https://api.coze.com")
+class TestRequester:
+    def test_code_msg(self, respx_mock):
+        respx_mock.post("/api/test").mock(
+            httpx.Response(200, json={"code": 100, "msg": "request failed"}, headers={"x-tt-logid": "mock-logid"})
+        )
+
+        with pytest.raises(CozeAPIError, match="code: 100, msg: request failed, logid: mock-logid"):
+            Requester().request("post", "https://api.coze.com/api/test", TestModel)
+
+    def test_auth_slow_down(self, respx_mock):
+        respx_mock.post("/api/test").mock(
+            httpx.Response(
+                200,
+                json={
+                    "error_code": "slow_down",
+                },
+                headers={"x-tt-logid": "mock-logid"},
+            )
+        )
+
+        with pytest.raises(CozePKCEAuthError, match="pkce auth error: slow_down"):
+            Requester().request("post", "https://api.coze.com/api/test", TestModel)
+
+    def test_error_message(self, respx_mock):
+        respx_mock.post("/api/test").mock(
+            httpx.Response(
+                200,
+                json={
+                    "error_message": "error_message",
+                },
+                headers={"x-tt-logid": "mock-logid"},
+            )
+        )
+
+        with pytest.raises(CozeAPIError, match="msg: error_message, logid: mock-logid"):
+            Requester().request("post", "https://api.coze.com/api/test", TestModel)
+
+    def test_debug_url(self, respx_mock):
+        respx_mock.post("/api/test").mock(
+            httpx.Response(
+                200,
+                json={
+                    "debug_url": "debug_url",
+                    "data": "data",
+                },
+                headers={"x-tt-logid": "mock-logid"},
+            )
+        )
+
+        Requester().request("post", "https://api.coze.com/api/test", TestDebugModel)

tests/test_workflow.py

@@ -86,3 +86,26 @@ def test_resume(self, respx_mock):
 
         assert events
         assert len(events) == 9
+
+    def test_stream_invalid_event(self, respx_mock):
+        coze = Coze(auth=TokenAuth(token="token"))
+
+        respx_mock.post("/v1/workflow/stream_resume").mock(
+            httpx.Response(
+                200,
+                content="""
+id:0
+event:invalid
+data:{}""",
+            )
+        )
+
+        with pytest.raises(Exception, match="invalid workflows.event: invalid"):
+            list(
+                coze.workflows.runs.resume(
+                    workflow_id="id",
+                    event_id="event_id",
+                    resume_data="resume_data",
+                    interrupt_type=123,
+                )
+            )