Service tests

Author: Peter554

pyproject.toml

@@ -26,6 +26,7 @@ build-backend = "uv_build"
 dev = [
     "deptry>=0.24.0",
     "pytest>=9.0.2",
+    "pytest-asyncio>=1.3.0",
     "ruff>=0.14.14",
     "time-machine>=3.2.0",
     "ty>=0.0.14",

src/circle/api.py

@@ -76,7 +76,9 @@ async def get_pipeline_by_number(
                 )
             return api_types.Pipeline.model_validate(response.json())
 
-    async def get_workflows(self, pipeline_id: str) -> list[api_types.Workflow]:
+    async def get_pipeline_workflows(
+        self, pipeline_id: str
+    ) -> list[api_types.Workflow]:
         """
         GET /pipeline/{pipeline-id}/workflow
         """
@@ -98,7 +100,7 @@ async def get_workflow(self, workflow_id: str) -> api_types.Workflow:
                 )
             return api_types.Workflow.model_validate(response.json())
 
-    async def get_jobs(self, workflow_id: str) -> list[api_types.Job]:
+    async def get_workflow_jobs(self, workflow_id: str) -> list[api_types.Job]:
         """
         GET /workflow/{id}/job
         """

src/circle/service.py

@@ -114,7 +114,7 @@ async def get_pipeline_workflows(
 
         workflows = self.cache_manager.get_pipeline_workflows(pipeline_id)
         if workflows is None:
-            workflows = await self.api_client.get_workflows(pipeline_id)
+            workflows = await self.api_client.get_pipeline_workflows(pipeline_id)
             self.cache_manager.set_pipeline_workflows(pipeline_id, workflows)
 
         return workflows
@@ -297,7 +297,7 @@ async def _get_workflow_jobs(
     ) -> list[api_types.Job]:
         jobs = self.cache_manager.get_workflow_jobs(workflow.id)
         if jobs is None:
-            jobs = await self.api_client.get_jobs(workflow.id)
+            jobs = await self.api_client.get_workflow_jobs(workflow.id)
             self.cache_manager.set_workflow_jobs(workflow.id, workflow.status, jobs)
         return jobs
 

tests/conftest.py

@@ -5,6 +5,19 @@
 from circle import api_types
 
 
+class FakeCache:
+    def __init__(self):
+        self.store: dict[str, object] = {}
+        self.ttls: dict[str, int | None] = {}
+
+    def get(self, k: str):
+        return self.store.get(k)
+
+    def set(self, k: str, v: object, ttl: int | None):
+        self.store[k] = v
+        self.ttls[k] = ttl
+
+
 def _deep_merge(base: dict, overrides: dict) -> dict:
     result = base.copy()
     for k, v in overrides.items():