fix: name the resource at the API layer so a forbidden MR lookup points at Merge requests

Author: lesnik512

release_scope/_errors.py

@@ -16,8 +16,8 @@ class AuthError(ReleaseScopeError):
 class GitLabError(ReleaseScopeError):
     exit_code: typing.ClassVar[int] = 4
 
-    def __init__(self, message: str, *, path: str, status: int | None = None, reason: str = "") -> None:
+    def __init__(self, message: str, *, resource: str, status: int | None = None, reason: str = "") -> None:
         super().__init__(message)
-        self.path = path
+        self.resource = resource
         self.status = status
         self.reason = reason

release_scope/_gitlab.py

@@ -135,14 +135,19 @@ class _Bridges(pydantic.RootModel[list[Bridge]]):
     pass
 
 
-def _translate(exc: httpware.ClientError, *, url: str) -> Exception:
+Resource: typing.TypeAlias = typing.Literal[
+    "group", "project", "deployments", "pipelines", "repository", "merge_requests"
+]
+
+
+def _translate(exc: httpware.ClientError, *, url: str, resource: Resource) -> Exception:
     if isinstance(exc, httpware.UnauthorizedError):
         return AuthError("GitLab rejected the token (401). Check that it is valid and not expired.")
     if isinstance(exc, httpware.StatusError):
         status: typing.Final = exc.response.status_code
-        return GitLabError(f"GitLab returned {status} for {unquote(url)}.", path=url, status=status)
+        return GitLabError(f"GitLab returned {status} for {unquote(url)}.", resource=resource, status=status)
     reason: typing.Final = type(exc).__name__
-    return GitLabError(f"GitLab request {unquote(url)} failed: {reason}.", path=url, reason=reason)
+    return GitLabError(f"GitLab request {unquote(url)} failed: {reason}.", resource=resource, reason=reason)
 
 
 def _quote(value: str) -> str:
@@ -153,18 +158,19 @@ def _quote(value: str) -> str:
 class GitLabApi:
     http: httpware.Client
 
-    def _get(self, url: str, params: dict[str, typing.Any], model: type[_ModelT]) -> _ModelT:
+    def _get(self, url: str, params: dict[str, typing.Any], model: type[_ModelT], *, resource: Resource) -> _ModelT:
         try:
             return self.http.get(url, params=params, response_model=model)
         except httpware.ClientError as exc:
-            raise _translate(exc, url=url) from exc
+            raise _translate(exc, url=url, resource=resource) from exc
 
     def _pages(
         self,
         url: str,
         params: dict[str, typing.Any],
         model: type[pydantic.RootModel[list[_ModelT]]],
         *,
+        resource: Resource,
         max_items: int | None = None,
     ) -> tuple[list[_ModelT], bool]:
         max_pages: typing.Final = _MAX_PAGES if max_items is None else math.ceil(max_items / _PER_PAGE)
@@ -175,7 +181,7 @@ def _pages(
                     url, params={**params, "per_page": _PER_PAGE, "page": page}, response_model=model
                 )
             except httpware.ClientError as exc:
-                raise _translate(exc, url=url) from exc
+                raise _translate(exc, url=url, resource=resource) from exc
             items.extend(batch.root)
             if not response.headers.get("x-next-page"):
                 break
@@ -186,13 +192,14 @@ def _pages(
         return items, False
 
     def get_project(self, path: str) -> Project:
-        return self._get(f"{_API}/projects/{_quote(path)}", {}, Project)
+        return self._get(f"{_API}/projects/{_quote(path)}", {}, Project, resource="project")
 
     def list_group_projects(self, group: str, *, include_subgroups: bool) -> list[Project]:
         projects, _ = self._pages(
             f"{_API}/groups/{_quote(group)}/projects",
             {"archived": "false", "with_shared": "false", "include_subgroups": str(include_subgroups).lower()},
             _Projects,
+            resource="group",
         )
         return projects
 
@@ -201,11 +208,12 @@ def latest_deployment(self, project_id: int, environment: str) -> Deployment | N
             f"{_API}/projects/{project_id}/deployments",
             {"environment": environment, "status": "success", "order_by": "id", "sort": "desc", "per_page": 1},
             _Deployments,
+            resource="deployments",
         )
         return deployments.root[0] if deployments.root else None
 
     def list_tags(self, project_id: int) -> tuple[list[Tag], bool]:
-        return self._pages(f"{_API}/projects/{project_id}/repository/tags", {}, _Tags)
+        return self._pages(f"{_API}/projects/{project_id}/repository/tags", {}, _Tags, resource="repository")
 
     def list_first_parent_commits(
         self, project_id: int, ref_range: str, *, max_items: int
@@ -214,6 +222,7 @@ def list_first_parent_commits(
             f"{_API}/projects/{project_id}/repository/commits",
             {"ref_name": ref_range, "first_parent": "true"},
             _Commits,
+            resource="repository",
             max_items=max_items,
         )
 
@@ -224,12 +233,16 @@ def list_merged_merge_requests(
             f"{_API}/projects/{project_id}/merge_requests",
             {"state": "merged", "target_branch": target_branch, "updated_after": updated_after.isoformat()},
             _MergeRequests,
+            resource="merge_requests",
         )
         return merge_requests
 
     def commit_merge_requests(self, project_id: int, sha: str) -> list[MergeRequest]:
         merge_requests, _ = self._pages(
-            f"{_API}/projects/{project_id}/repository/commits/{sha}/merge_requests", {}, _MergeRequests
+            f"{_API}/projects/{project_id}/repository/commits/{sha}/merge_requests",
+            {},
+            _MergeRequests,
+            resource="merge_requests",
         )
         return merge_requests
 
@@ -238,6 +251,7 @@ def list_push_pipelines(self, project_id: int, *, ref: str, updated_after: datet
             f"{_API}/projects/{project_id}/pipelines",
             {"ref": ref, "source": "push", "updated_after": updated_after.isoformat()},
             _Pipelines,
+            resource="pipelines",
         )
         return pipelines
 
@@ -246,18 +260,25 @@ def latest_pipeline(self, project_id: int, *, ref: str) -> Pipeline | None:
             f"{_API}/projects/{project_id}/pipelines",
             {"ref": ref, "order_by": "id", "sort": "desc", "per_page": 1},
             _Pipelines,
+            resource="pipelines",
         )
         return pipelines.root[0] if pipelines.root else None
 
     def failed_jobs(self, project_id: int, pipeline_id: int) -> list[Job]:
         jobs, _ = self._pages(
-            f"{_API}/projects/{project_id}/pipelines/{pipeline_id}/jobs", {"scope[]": "failed"}, _Jobs
+            f"{_API}/projects/{project_id}/pipelines/{pipeline_id}/jobs",
+            {"scope[]": "failed"},
+            _Jobs,
+            resource="pipelines",
         )
         return jobs
 
     def failed_bridges(self, project_id: int, pipeline_id: int) -> list[Bridge]:
         # `trigger_jobs` replaces this route only from GitLab 19.2; older instances have `bridges` alone.
         bridges, _ = self._pages(
-            f"{_API}/projects/{project_id}/pipelines/{pipeline_id}/bridges", {"scope[]": "failed"}, _Bridges
+            f"{_API}/projects/{project_id}/pipelines/{pipeline_id}/bridges",
+            {"scope[]": "failed"},
+            _Bridges,
+            resource="pipelines",
         )
         return bridges

release_scope/_messages.py

@@ -1,12 +1,10 @@
-import re
+import http
 import typing
 
 from release_scope._errors import GitLabError
 from release_scope._gitlab import Project
 
 
-_FORBIDDEN: typing.Final = 403
-_PROJECT_RESOURCE: typing.Final = re.compile(r"/projects/[^/]+/([^/?]+)")
 _RESOURCES: typing.Final[dict[str, tuple[str, tuple[str, ...]]]] = {
     "deployments": ("deployments", ("Environments", "CI/CD")),
     "pipelines": ("pipelines", ("CI/CD",)),
@@ -20,17 +18,12 @@ def _feature_setting(project: Project, feature: str) -> str:
     return f"{project.web_url}/edit#js-shared-permissions → Visibility, project features, permissions → {feature}"
 
 
-def _resource(path: str) -> tuple[str, tuple[str, ...]]:
-    match: typing.Final = _PROJECT_RESOURCE.search(path)
-    return _RESOURCES.get(match.group(1), ("project data", ())) if match else ("project data", ())
-
-
 def explain_failure(project: Project, error: GitLabError) -> str:
     name: typing.Final = project.path_with_namespace
-    resource, features = _resource(error.path)
+    resource, features = _RESOURCES.get(error.resource, ("project data", ()))
     if error.status is None:
         return f"{name}: GitLab request for {resource} failed ({error.reason})."
-    if error.status != _FORBIDDEN:
+    if error.status != http.HTTPStatus.FORBIDDEN:
         return f"{name}: GitLab returned {error.status} for {resource}."
     checks: typing.Final = [
         f"- {feature} {'are' if feature in _PLURAL_FEATURES else 'is'} enabled: {_feature_setting(project, feature)}"

release_scope/_use_case.py

@@ -1,6 +1,7 @@
 import collections.abc
 import dataclasses
 import datetime
+import http
 import typing
 from urllib.parse import quote
 
@@ -25,12 +26,11 @@
 from release_scope._settings import Settings
 
 
-_FORBIDDEN: typing.Final = 403
 _SETTLED_PIPELINE_STATUSES: typing.Final = frozenset({"success", "failed", "canceled", "skipped"})
 
 
 def _resolution_error(error: GitLabError, target: str) -> Exception:
-    if error.status == _FORBIDDEN:
+    if error.status == http.HTTPStatus.FORBIDDEN:
         return AuthError(
             f"GitLab denied access to {target} (403). "
             "Check that the token has the 'read_api' scope and that its user can see it."

tests/test_cli.py

@@ -81,6 +81,28 @@ def test_failed_service_is_reported_and_exits_non_zero(gitlab: respx.Router, tmp
     assert [item["error"] is None for item in json.loads(output.read_text())["services"]] == [False, True]
 
 
+@pytest.mark.usefixtures("cli_env")
+def test_forbidden_service_fails_alone_and_exits_non_zero(gitlab: respx.Router, tmp_path: pathlib.Path) -> None:
+    gitlab["group"].respond(json=[SERVICE, project(2, "team/nodeploy")])
+    gitlab.get(f"{API}/projects/2/deployments").respond(403)
+
+    result: typing.Final = _invoke("collect", "-g", "team", "-o", str(tmp_path / "report.json"))
+
+    assert result.exit_code == 1
+    assert "Error: team/nodeploy: GitLab denied access to deployments (403). Check that:" in result.output
+    assert "1 failed" in result.output
+
+
+@pytest.mark.usefixtures("cli_env")
+def test_rejected_token_exits_with_auth_code(httpx2_mock: respx.Router, tmp_path: pathlib.Path) -> None:
+    httpx2_mock.get(f"{API}/groups/team/projects").respond(401)
+
+    result: typing.Final = _invoke("collect", "-g", "team", "-o", str(tmp_path / "report.json"))
+
+    assert result.exit_code == 3
+    assert "Error: GitLab rejected the token (401)." in result.output
+
+
 @pytest.mark.usefixtures("cli_env")
 def test_authentication_failure_exits_with_auth_code(httpx2_mock: respx.Router, tmp_path: pathlib.Path) -> None:
     httpx2_mock.get(f"{API}/groups/team/projects").respond(403)

tests/test_use_case.py

@@ -204,6 +204,23 @@ def test_forbidden_resource_lists_the_feature_that_guards_it(
     )
 
 
+@pytest.mark.httpx2(assert_all_called=False)
+def test_forbidden_commit_lookup_points_at_merge_requests(gitlab: respx.Router) -> None:
+    gitlab["commit_mrs:head"].respond(403)
+
+    error: typing.Final = _only_service(_collect()).error
+
+    assert error is not None
+    assert error.startswith("team/svc: GitLab denied access to merge requests (403). Check that:\n- Merge requests are")
+
+
+def test_forbidden_project_stops_the_run(httpx2_mock: respx.Router) -> None:
+    httpx2_mock.get(f"{API}/projects/team%2Fsecret").respond(403)
+
+    with pytest.raises(AuthError, match=r"GitLab denied access to project 'team/secret' \(403\)"):
+        _collect(groups=(), projects=("team/secret",))
+
+
 @pytest.mark.httpx2(assert_all_called=False)
 def test_network_failure_names_the_project(gitlab: respx.Router) -> None:
     gitlab["tags"].mock(side_effect=httpcore2.ConnectError("down"))