Add: Add GitHub API to request workflow information

pontos/github/api.py

@@ -542,9 +542,9 @@ def set_labels(self, repo: str, issue: int, labels: List[str]):
         response = self._request(api, data=data, request=httpx.post)
         response.raise_for_status()
 
-    def _get_artifacts(self, api: str) -> Iterable[JSON]:
+    def _get_paged_items(self, api: str, key: str) -> Iterable[JSON]:
         """
-        Internal method to get the artifacts information from different REST
+        Internal method to get the paged items information from different REST
         URLs.
         """
         page = 1
@@ -553,19 +553,18 @@ def _get_artifacts(self, api: str) -> Iterable[JSON]:
         response = self._request(api, request=httpx.get, params=params)
         json = response.json()
         total = json.get("total_count", 0)
-        artifacts = list(json.get("artifacts", []))
-        downloaded = len(artifacts)
+        items = json[key]
+        downloaded = len(items)
 
-        # downloading all the artifacts uses a different pagination :-/
         while total - downloaded > 0:
             page += 1
             params = {"per_page": per_page, "page": page}
             response = self._request(api, request=httpx.get, params=params)
             json = response.json()
-            artifacts.extend(list(json.get("artifacts", [])))
-            downloaded = len(artifacts)
+            items.extend(json[key])
+            downloaded = len(items)
 
-        return artifacts
+        return items
 
     def get_repository_artifacts(self, repo: str) -> Iterable[JSON]:
         """
@@ -578,7 +577,7 @@ def get_repository_artifacts(self, repo: str) -> Iterable[JSON]:
             Information about the artifacts in the repository as a dict
         """
         api = f"/repos/{repo}/actions/artifacts"
-        return self._get_artifacts(api)
+        return self._get_paged_items(api, "artifacts")
 
     def get_repository_artifact(
         self, repo: str, artifact: str
@@ -645,7 +644,7 @@ def get_workflow_artifacts(
             Information about the artifacts in the workflow as a dict
         """
         api = f"/repos/{repo}/actions/runs/{workflow}/artifacts"
-        return self._get_artifacts(api)
+        return self._get_paged_items(api, "artifacts")
 
     def delete_repository_artifact(self, repo: str, artifact: str):
         """
@@ -662,3 +661,20 @@ def delete_repository_artifact(self, repo: str, artifact: str):
         api = f"/repos/{repo}/actions/artifacts/{artifact}"
         response = self._request(api, request=httpx.delete)
         response.raise_for_status()
+
+    def get_workflows(self, repo: str) -> Iterable[JSON]:
+        """
+        List all workflows of a repository
+
+        Args:
+            repo: GitHub repository (owner/name) to use
+
+        Raises:
+            HTTPStatusError: A httpx.HTTPStatusError is raised if the request
+                failed.
+
+        Returns:
+            Information about the  as a dict
+        """
+        api = f"/repos/{repo}/actions/workflows"
+        return self._get_paged_items(api, "workflows")

tests/github/test_api.py

@@ -1094,3 +1094,94 @@ def test_delete_repository_artifact_failure(self, requests_mock: MagicMock):
             params=None,
             follow_redirects=True,
         )
+
+    @patch("pontos.github.api.httpx.get")
+    def test_get_workflows(self, requests_mock: MagicMock):
+        response = MagicMock()
+        response.links = None
+        response.json.return_value = {
+            "total_count": 1,
+            "workflows": [
+                {
+                    "id": 11,
+                    "name": "Foo",
+                }
+            ],
+        }
+        requests_mock.return_value = response
+        api = GitHubRESTApi("12345")
+        artifacts = api.get_workflows("foo/bar")
+
+        requests_mock.assert_called_once_with(
+            "https://api.github.com/repos/foo/bar/actions/workflows",
+            headers={
+                "Authorization": "token 12345",
+                "Accept": "application/vnd.github.v3+json",
+            },
+            params={"per_page": 100, "page": 1},
+            follow_redirects=True,
+        )
+
+        self.assertEqual(len(artifacts), 1)
+        self.assertEqual(artifacts[0]["name"], "Foo")
+
+    @patch("pontos.github.api.httpx.get")
+    def test_get_workflows_with_pagination(self, requests_mock: MagicMock):
+        response = MagicMock()
+        response.links = None
+        response.json.side_effect = [
+            {
+                "total_count": 120,
+                "workflows": [
+                    {
+                        "id": id,
+                        "name": f"Foo-{id}",
+                    }
+                    for id in range(0, 100)
+                ],
+            },
+            {
+                "total_count": 120,
+                "workflows": [
+                    {
+                        "id": id,
+                        "name": f"Foo-{id}",
+                    }
+                    for id in range(100, 120)
+                ],
+            },
+        ]
+        requests_mock.return_value = response
+        api = GitHubRESTApi("12345")
+        artifacts = api.get_workflows("foo/bar")
+
+        requests_mock.assert_has_calls(
+            [
+                call.__bool__(),
+                call(
+                    "https://api.github.com/repos/foo/bar/actions/workflows",
+                    headers={
+                        "Accept": "application/vnd.github.v3+json",
+                        "Authorization": "token 12345",
+                    },
+                    params={"per_page": 100, "page": 1},
+                    follow_redirects=True,
+                ),
+                call().json(),
+                call.__bool__(),
+                call(
+                    "https://api.github.com/repos/foo/bar/actions/workflows",
+                    headers={
+                        "Accept": "application/vnd.github.v3+json",
+                        "Authorization": "token 12345",
+                    },
+                    params={"per_page": 100, "page": 2},
+                    follow_redirects=True,
+                ),
+                call().json(),
+            ]
+        )
+
+        self.assertEqual(len(artifacts), 120)
+        self.assertEqual(artifacts[0]["name"], "Foo-0")
+        self.assertEqual(artifacts[119]["name"], "Foo-119")