diff --git a/src/scm/providers/cursor_origin/provider.py b/src/scm/providers/cursor_origin/provider.py index b9a15a6..62f4c74 100644 --- a/src/scm/providers/cursor_origin/provider.py +++ b/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"]) diff --git a/tests/unit/provider/test_cursor_origin.py b/tests/unit/provider/test_cursor_origin.py index d74a326..055900e 100644 --- a/tests/unit/provider/test_cursor_origin.py +++ b/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",