feat(cursor-origin): Read app identity, readme, topics and PR template

src/scm/providers/cursor_origin/provider.py

@@ -1,4 +1,4 @@
-from collections.abc import Callable
+from collections.abc import Callable, Iterator
 from datetime import UTC, date, datetime
 from typing import Any, Literal
 from urllib.parse import quote
@@ -8,6 +8,7 @@
 from scm.errors import (
     PathIsDirectory,
     PathIsNotDirectory,
+    ReadmeNotFound,
     ResourceBadRequest,
     ResourceConflict,
     ResourceNotFound,
@@ -72,6 +73,10 @@
 )
 
 PROVIDER_TYPE: ProviderName = "cursor_origin"
+VALID_README_FILES = {"readme", "readme.md", "readme.txt", "readme.rst"}
+PULL_REQUEST_TEMPLATE_PARENT_DIRS = ("", "docs")
+PULL_REQUEST_TEMPLATE_FILENAME = "pull_request_template.md"
+PULL_REQUEST_TEMPLATE_DIRNAME = "pull_request_template"
 CURSOR_ORIGIN_WEB_BASE_URL = "https://cursor.com/codebase"
 # Origin has no no-reply address for an app.
 CURSOR_ORIGIN_APP_COMMIT_EMAIL = "noreply@sentry.io"
@@ -235,6 +240,10 @@ def patch(self, path: str, data: dict[str, Any]) -> requests.Response:
     def delete(self, path: str) -> requests.Response:
         return self.request("DELETE", path=path)
 
+    def get_authenticated_actor(self) -> ActionResult[Author]:
+        response = self.get("/app", credentials_set="application")
+        return map_action(response, map_authenticated_actor)
+
     def get_app_installation(self) -> ActionResult[AppInstallation]:
         response = self.get(f"/app/installations/{self.installation_id}", credentials_set="application")
         return map_action(response, map_app_installation)
@@ -243,6 +252,12 @@ def get_repository(self) -> ActionResult[GitRepository]:
         response = self.get(f"/repos/{self.repository_path}")
         return map_action(response, map_repository)
 
+    def get_repository_topics(
+        self,
+        request_options: RequestOptions | None = None,
+    ) -> ActionResult[list[str]]:
+        return {"data": [], "type": PROVIDER_TYPE, "raw": {"data": [], "headers": {}}, "meta": {}}
+
     def get_pull_request(
         self,
         pull_request_id: str,
@@ -354,6 +369,45 @@ def get_file_content(
             raise PathIsDirectory(detail=path)
         return map_action(response, map_file_content, raw)
 
+    def get_readme(
+        self,
+        ref: str,
+        pagination: PaginationParams | None = None,
+        request_options: RequestOptions | None = None,
+    ) -> ActionResult[FileContent]:
+        for entry in self._directory("", ref, request_options):
+            if entry["type"] == "file" and entry["path"].lower() in VALID_README_FILES:
+                return self.get_file_content(entry["path"], ref=ref, request_options=request_options)
+        raise ReadmeNotFound()
+
+    def get_pull_request_template(
+        self,
+        ref: str,
+        pagination: PaginationParams | None = None,
+        request_options: RequestOptions | None = None,
+    ) -> Iterator[ActionResult[FileContent]]:
+        for parent in PULL_REQUEST_TEMPLATE_PARENT_DIRS:
+            for path in self._template_paths(parent, ref, request_options):
+                yield self.get_file_content(path, ref=ref, request_options=request_options)
+
+    def _template_paths(self, parent: str, ref: str, request_options: RequestOptions | None) -> Iterator[str]:
+        for entry in self._directory(parent, ref, request_options):
+            basename = entry["path"].rsplit("/", 1)[-1].lower()
+            if entry["type"] == "file" and basename == PULL_REQUEST_TEMPLATE_FILENAME:
+                yield entry["path"]
+            elif entry["type"] == "directory" and basename == PULL_REQUEST_TEMPLATE_DIRNAME:
+                for child in self._directory(entry["path"], ref, request_options):
+                    if child["type"] == "file" and child["path"].lower().endswith(".md"):
+                        yield child["path"]
+
+    def _directory(self, path: str, ref: str, request_options: RequestOptions | None) -> list[FileContent]:
+        try:
+            return self.get_directory_contents(path, ref=ref, request_options=request_options)["data"]
+        except SCMCodedError as e:
+            if e.code in ("resource_not_found", "path_is_not_directory"):
+                return []
+            raise
+
     def get_directory_contents(
         self,
         path: str,
@@ -988,6 +1042,10 @@ def _all_comments(
             page = {"per_page": MAX_PAGE_SIZE, "cursor": raw["nextPageToken"]}
 
 
+def map_authenticated_actor(raw: dict[str, Any]) -> Author:
+    return Author(id=raw["id"], username=raw["displayName"])
+
+
 def map_app_installation(raw: dict[str, Any]) -> AppInstallation:
     """A write scope also grants its read scope."""
     scopes = set(raw["scopes"])

tests/unit/provider/test_cursor_origin.py

@@ -7,10 +7,12 @@
 from scm.errors import (
     PathIsDirectory,
     PathIsNotDirectory,
+    ReadmeNotFound,
     ResourceBadRequest,
     ResourceConflict,
     ResourceGatewayTimeout,
     ResourceNotFound,
+    ResourceServerError,
     StaleBranchHead,
     UnexpectedResponseFormat,
 )
@@ -349,6 +351,102 @@ def test_only_tarballs_are_offered(self, provider: CursorOriginProvider) -> None
             provider.download_archive("abc123", archive_format="zip")
 
 
+def _entry(path: str, entry_type: str = "file") -> dict[str, Any]:
+    return {"type": entry_type, "name": path.rsplit("/", 1)[-1], "path": path, "sha": "s", "size": "1"}
+
+
+class TestAuthenticatedActor:
+    def test_the_app_is_read_with_its_own_credentials(
+        self, provider: CursorOriginProvider, client: unittest.mock.MagicMock
+    ) -> None:
+        client.request.return_value = _response({"id": "app_01example", "displayName": "Sentry"})
+
+        result = provider.get_authenticated_actor()
+
+        assert client.request.call_args.kwargs["path"] == "/app"
+        assert client.request.call_args.kwargs["credentials_set"] == "application"
+        assert result["data"] == {"id": "app_01example", "username": "Sentry"}
+
+
+class TestPullRequestTemplate:
+    def test_a_template_in_the_root_is_read(
+        self, provider: CursorOriginProvider, client: unittest.mock.MagicMock
+    ) -> None:
+        client.request.side_effect = [
+            _response({"entries": [_entry("PULL_REQUEST_TEMPLATE.md"), _entry("README.md")]}),
+            _response({**FILE_RAW, "path": "PULL_REQUEST_TEMPLATE.md"}),
+            _response({"entries": []}),
+        ]
+
+        templates = list(provider.get_pull_request_template("main"))
+
+        assert [template["data"]["path"] for template in templates] == ["PULL_REQUEST_TEMPLATE.md"]
+        assert client.request.call_args_list[0].kwargs["params"] == {"path": "", "ref": "main"}
+        assert client.request.call_args_list[2].kwargs["params"] == {"path": "docs", "ref": "main"}
+
+    def test_every_template_in_a_template_directory_is_read(
+        self, provider: CursorOriginProvider, client: unittest.mock.MagicMock
+    ) -> None:
+        client.request.side_effect = [
+            _response({"entries": []}),
+            _response({"entries": [_entry("docs/PULL_REQUEST_TEMPLATE", "dir")]}),
+            _response(
+                {
+                    "entries": [
+                        _entry("docs/PULL_REQUEST_TEMPLATE/bug.md"),
+                        _entry("docs/PULL_REQUEST_TEMPLATE/notes.txt"),
+                    ]
+                }
+            ),
+            _response({**FILE_RAW, "path": "docs/PULL_REQUEST_TEMPLATE/bug.md"}),
+        ]
+
+        templates = list(provider.get_pull_request_template("main"))
+
+        assert [template["data"]["path"] for template in templates] == ["docs/PULL_REQUEST_TEMPLATE/bug.md"]
+
+    def test_a_repository_with_no_template_yields_nothing(
+        self, provider: CursorOriginProvider, client: unittest.mock.MagicMock
+    ) -> None:
+        client.request.return_value = _response({"message": "not found"}, status_code=404)
+
+        assert list(provider.get_pull_request_template("main")) == []
+
+    def test_a_failure_that_is_not_a_missing_path_is_raised(
+        self, provider: CursorOriginProvider, client: unittest.mock.MagicMock
+    ) -> None:
+        client.request.return_value = _response({"message": "boom"}, status_code=500)
+
+        with pytest.raises(ResourceServerError):
+            list(provider.get_pull_request_template("main"))
+
+
+class TestReadme:
+    def test_the_readme_is_found_in_the_root(
+        self, provider: CursorOriginProvider, client: unittest.mock.MagicMock
+    ) -> None:
+        client.request.side_effect = [
+            _response({"entries": [_entry("src", "dir"), _entry("README.md")]}),
+            _response({**FILE_RAW, "path": "README.md"}),
+        ]
+
+        result = provider.get_readme("main")
+
+        assert client.request.call_args_list[0].kwargs["params"] == {"path": "", "ref": "main"}
+        assert result["data"]["path"] == "README.md"
+
+    def test_a_repository_without_one_is_refused(
+        self, provider: CursorOriginProvider, client: unittest.mock.MagicMock
+    ) -> None:
+        client.request.return_value = _response({"entries": [_entry("src/app.py")]})
+
+        with pytest.raises(ReadmeNotFound):
+            provider.get_readme("main")
+
+    def test_a_repository_has_no_topics(self, provider: CursorOriginProvider) -> None:
+        assert provider.get_repository_topics()["data"] == []
+
+
 def _check_run_raw(**overrides: Any) -> dict[str, Any]:
     return {
         "id": "cr_01example",