Add code-owners plugin as an extension
Adds a new sibling package `gerrit_mcp_server_code_owners` wrapping the
Gerrit code-owners plugin REST API. `register()` wires up three tools,
all gated via `@requires_plugin("code-owners")`:
- `get_code_owner_status`: per-file ownership status for a change
(APPROVED/PENDING/INSUFFICIENT_REVIEWERS), with pagination via
`limit` and `start`.
- `get_code_owners_for_path`: suggested owners for a specific path in a
change revision, filtered and ranked by the plugin (current reviewers
ranked higher, change owner and service users excluded).
- `check_code_owner`: checks code ownership for a path in a branch,
returning ownership kind (global/default/fallback) and optional
change-level permissions when `change_id` is supplied.
All tools return typed TypedDicts so the MCP SDK auto-generates an
output_schema and emits both structured and unstructured content. Error
paths raise exceptions rather than returning text error payloads.
Includes 23 unit tests and updates docs/available_tools.md.
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
Change-Id: Ibf1ec8a491c18a39f20c0f294abc818b9ad055bf
diff --git a/docs/available_tools.md b/docs/available_tools.md
index dc42429..53fcdd2 100644
--- a/docs/available_tools.md
+++ b/docs/available_tools.md
@@ -81,3 +81,26 @@
- **get_dependents**: Returns changes that explicitly declare `Depends-on:` on
the given change, via the `independson:<change_id>` query operator registered
by the plugin.
+
+### Code-owners plugin (`gerrit_mcp_server_code_owners`)
+
+Requires the
+[`code-owners`](https://gerrit.googlesource.com/plugins/code-owners/) plugin on
+the target Gerrit host. All tools auto-route to a configured host that has the
+plugin installed.
+
+- **get_code_owner_status**: Returns the per-file code owner approval status for
+ a change. Each file entry carries a `status` of `APPROVED`, `PENDING`, or
+ `INSUFFICIENT_REVIEWERS` for both its old and new paths (relevant for
+ renames). Supports `limit` and `start` for pagination when a change touches
+ many files.
+- **get_code_owners_for_path**: Returns the suggested code owners for a specific
+ path in a change revision. The plugin filters out the change owner and service
+ users, and ranks current reviewers higher. Pass `revision_id` to target a
+ specific patch set (defaults to `current`) and `limit` to cap the result
+ count.
+- **check_code_owner**: Checks whether a specific user (by email) is a code
+ owner for a path in a branch. Returns detailed ownership information including
+ whether the user is a global, default, or fallback code owner, and optionally
+ their change-level permissions (read ref, see change, approve) when
+ `change_id` is supplied.
diff --git a/gerrit_mcp_server_code_owners/__init__.py b/gerrit_mcp_server_code_owners/__init__.py
new file mode 100644
index 0000000..e69de29
--- /dev/null
+++ b/gerrit_mcp_server_code_owners/__init__.py
diff --git a/gerrit_mcp_server_code_owners/extension.py b/gerrit_mcp_server_code_owners/extension.py
new file mode 100644
index 0000000..2b9c9fc
--- /dev/null
+++ b/gerrit_mcp_server_code_owners/extension.py
@@ -0,0 +1,259 @@
+"""gerrit-mcp-server extension for the Gerrit code-owners plugin.
+
+Exposes three MCP tools:
+ - get_code_owner_status: per-file code owner approval status for a change
+ - get_code_owners_for_path: code owners for a specific path in a change revision
+ - check_code_owner: check whether a user is a code owner for a path in a branch
+
+All tools require the 'code-owners' Gerrit plugin via @requires_plugin.
+"""
+
+import json
+from typing import Any, Dict, List, Optional, TypedDict
+from urllib.parse import quote
+
+from gerrit_mcp_server.extensions import ExtensionContext, requires_plugin
+
+
+class _AccountInfo(TypedDict):
+ account_id: int
+ name: Optional[str]
+ email: Optional[str]
+ username: Optional[str]
+
+
+class _PathCodeOwnerStatusInfo(TypedDict):
+ path: str
+ status: str
+ reasons: Optional[List[str]]
+
+
+class _FileCodeOwnerStatusInfo(TypedDict):
+ change_type: Optional[str]
+ old_path_status: Optional[_PathCodeOwnerStatusInfo]
+ new_path_status: Optional[_PathCodeOwnerStatusInfo]
+
+
+class _CodeOwnerStatusResult(TypedDict):
+ change_id: str
+ patch_set_number: int
+ file_code_owner_statuses: List[_FileCodeOwnerStatusInfo]
+ more: Optional[bool]
+ accounts: Optional[Dict[str, _AccountInfo]]
+
+
+class _CodeOwnerInfo(TypedDict):
+ account: _AccountInfo
+ scorings: Optional[Dict[str, int]]
+
+
+class _CodeOwnersForPathResult(TypedDict):
+ change_id: str
+ revision_id: str
+ path: str
+ code_owners: List[_CodeOwnerInfo]
+ owned_by_all_users: Optional[bool]
+
+
+class _CheckCodeOwnerResult(TypedDict):
+ project: str
+ branch: str
+ path: str
+ email: str
+ is_code_owner: bool
+ is_resolvable: bool
+ can_read_ref: Optional[bool]
+ can_see_change: Optional[bool]
+ can_approve_change: Optional[bool]
+ is_fallback_code_owner: Optional[bool]
+ is_default_code_owner: Optional[bool]
+ is_global_code_owner: Optional[bool]
+ is_owned_by_all_users: Optional[bool]
+ annotation: Optional[List[str]]
+
+
+def _parse_path_status(
+ raw: Optional[Dict[str, Any]],
+) -> Optional[_PathCodeOwnerStatusInfo]:
+ if raw is None:
+ return None
+ return {
+ "path": raw.get("path", ""),
+ "status": raw.get("status", ""),
+ "reasons": raw.get("reasons") or None,
+ }
+
+
+def register(ctx: ExtensionContext) -> None:
+ """Register code-owners MCP tools."""
+
+ @ctx.mcp.tool()
+ @requires_plugin("code-owners", ctx.plugin_registry)
+ async def get_code_owner_status(
+ change_id: str,
+ limit: Optional[int] = None,
+ start: Optional[int] = None,
+ gerrit_base_url: Optional[str] = None,
+ ) -> _CodeOwnerStatusResult:
+ """Return code owner approval status for each file in a change.
+
+ The status field for each path is one of:
+ APPROVED - a code owner approved, or an override is present
+ PENDING - a code owner is a reviewer but has not approved yet
+ INSUFFICIENT_REVIEWERS - no code owner has been added as a reviewer
+
+ Pass limit and start for pagination when a change touches many files.
+ """
+ base_url = ctx.get_base_url(gerrit_base_url)
+ url = f"{base_url}/changes/{quote(str(change_id))}/code_owners.status"
+ params = []
+ if limit is not None:
+ params.append(f"n={limit}")
+ if start is not None:
+ params.append(f"S={start}")
+ if params:
+ url += "?" + "&".join(params)
+
+ result_str = await ctx.run_curl([url], base_url)
+ try:
+ data = json.loads(result_str)
+ except json.JSONDecodeError as e:
+ raise ValueError(f"Could not parse Gerrit response: {result_str}") from e
+
+ file_statuses: List[_FileCodeOwnerStatusInfo] = []
+ for raw_file in data.get("file_code_owner_statuses", []):
+ entry: _FileCodeOwnerStatusInfo = {
+ "change_type": raw_file.get("change_type") or None,
+ "old_path_status": _parse_path_status(raw_file.get("old_path_status")),
+ "new_path_status": _parse_path_status(raw_file.get("new_path_status")),
+ }
+ file_statuses.append(entry)
+
+ accounts: Optional[Dict[str, _AccountInfo]] = None
+ raw_accounts = data.get("accounts")
+ if raw_accounts:
+ accounts = {}
+ for account_id, raw_account in raw_accounts.items():
+ accounts[account_id] = {
+ "account_id": raw_account.get("_account_id", 0),
+ "name": raw_account.get("name") or None,
+ "email": raw_account.get("email") or None,
+ "username": raw_account.get("username") or None,
+ }
+
+ return {
+ "change_id": change_id,
+ "patch_set_number": data.get("patch_set_number", 0),
+ "file_code_owner_statuses": file_statuses,
+ "more": data.get("more") or None,
+ "accounts": accounts,
+ }
+
+ @ctx.mcp.tool()
+ @requires_plugin("code-owners", ctx.plugin_registry)
+ async def get_code_owners_for_path(
+ change_id: str,
+ path: str,
+ revision_id: str = "current",
+ limit: Optional[int] = None,
+ gerrit_base_url: Optional[str] = None,
+ ) -> _CodeOwnersForPathResult:
+ """Return suggested code owners for a specific path in a change revision.
+
+ The plugin filters out the change owner and service users, and ranks
+ reviewers higher. Use this to find who should review a particular file.
+
+ Pass revision_id to target a specific patch set (defaults to 'current').
+ Pass limit to cap how many owners are returned.
+ """
+ base_url = ctx.get_base_url(gerrit_base_url)
+ encoded_path = quote(path.lstrip("/"), safe="/")
+ url = (
+ f"{base_url}/changes/{quote(str(change_id))}"
+ f"/revisions/{quote(str(revision_id))}"
+ f"/code_owners/{encoded_path}"
+ )
+ if limit is not None:
+ url += f"?n={limit}"
+
+ result_str = await ctx.run_curl([url], base_url)
+ try:
+ data = json.loads(result_str)
+ except json.JSONDecodeError as e:
+ raise ValueError(f"Could not parse Gerrit response: {result_str}") from e
+
+ owners: List[_CodeOwnerInfo] = []
+ for raw_owner in data.get("code_owners", []):
+ raw_account = raw_owner.get("account", {})
+ account: _AccountInfo = {
+ "account_id": raw_account.get("_account_id", 0),
+ "name": raw_account.get("name") or None,
+ "email": raw_account.get("email") or None,
+ "username": raw_account.get("username") or None,
+ }
+ owner: _CodeOwnerInfo = {
+ "account": account,
+ "scorings": raw_owner.get("scorings") or None,
+ }
+ owners.append(owner)
+
+ return {
+ "change_id": change_id,
+ "revision_id": revision_id,
+ "path": path,
+ "code_owners": owners,
+ "owned_by_all_users": data.get("owned_by_all_users") or None,
+ }
+
+ @ctx.mcp.tool()
+ @requires_plugin("code-owners", ctx.plugin_registry)
+ async def check_code_owner(
+ project: str,
+ branch: str,
+ path: str,
+ email: str,
+ change_id: Optional[str] = None,
+ gerrit_base_url: Optional[str] = None,
+ ) -> _CheckCodeOwnerResult:
+ """Check whether a user (by email) is a code owner for a path in a branch.
+
+ Returns detailed ownership information including whether the user can
+ read the ref, see and approve the change (if change_id is provided),
+ and which kind of code owner they are (global, default, fallback).
+
+ Pass change_id to also check change-level permissions.
+ """
+ base_url = ctx.get_base_url(gerrit_base_url)
+ encoded_project = quote(project, safe="")
+ encoded_branch = quote(branch, safe="")
+ url = (
+ f"{base_url}/projects/{encoded_project}"
+ f"/branches/{encoded_branch}"
+ f"/code_owners.check"
+ f"?email={quote(email)}&path={quote(path)}"
+ )
+ if change_id is not None:
+ url += f"&change={quote(str(change_id))}"
+
+ result_str = await ctx.run_curl([url], base_url)
+ try:
+ data = json.loads(result_str)
+ except json.JSONDecodeError as e:
+ raise ValueError(f"Could not parse Gerrit response: {result_str}") from e
+
+ return {
+ "project": project,
+ "branch": branch,
+ "path": path,
+ "email": email,
+ "is_code_owner": data.get("is_code_owner", False),
+ "is_resolvable": data.get("is_resolvable", False),
+ "can_read_ref": data.get("can_read_ref"),
+ "can_see_change": data.get("can_see_change"),
+ "can_approve_change": data.get("can_approve_change"),
+ "is_fallback_code_owner": data.get("is_fallback_code_owner"),
+ "is_default_code_owner": data.get("is_default_code_owner"),
+ "is_global_code_owner": data.get("is_global_code_owner"),
+ "is_owned_by_all_users": data.get("is_owned_by_all_users"),
+ "annotation": data.get("annotation") or None,
+ }
diff --git a/gerrit_mcp_server_code_owners/pyproject.toml b/gerrit_mcp_server_code_owners/pyproject.toml
new file mode 100644
index 0000000..eaf5562
--- /dev/null
+++ b/gerrit_mcp_server_code_owners/pyproject.toml
@@ -0,0 +1,17 @@
+[project]
+name = "gerrit-mcp-server-code-owners"
+version = "0.1.0"
+description = "gerrit-mcp-server extension for the Gerrit code-owners plugin"
+requires-python = ">=3.12"
+dependencies = ["gerrit-mcp-server"]
+
+[project.entry-points."gerrit_mcp_server.extensions"]
+gerrit_mcp_server_code_owners = "gerrit_mcp_server_code_owners.extension:register"
+
+[build-system]
+requires = ["setuptools>=68"]
+build-backend = "setuptools.build_meta"
+
+[tool.setuptools]
+packages = ["gerrit_mcp_server_code_owners"]
+package-dir = {"gerrit_mcp_server_code_owners" = "."}
diff --git a/pyproject.toml b/pyproject.toml
index c22546f..a6fabac 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -8,7 +8,8 @@
"uvicorn",
"websockets",
"gerrit-mcp-server-task",
- "gerrit-mcp-server-depends-on"
+ "gerrit-mcp-server-depends-on",
+ "gerrit-mcp-server-code-owners"
]
[project.optional-dependencies]
@@ -41,12 +42,13 @@
build-backend = "setuptools.build_meta"
[tool.uv.workspace]
-members = ["gerrit_mcp_server_task", "gerrit_mcp_server_depends_on"]
+members = ["gerrit_mcp_server_task", "gerrit_mcp_server_depends_on", "gerrit_mcp_server_code_owners"]
[tool.uv.sources]
gerrit-mcp-server = { workspace = true }
gerrit-mcp-server-task = { workspace = true }
gerrit-mcp-server-depends-on = { workspace = true }
+gerrit-mcp-server-code-owners = { workspace = true }
[tool.setuptools]
packages = ["gerrit_mcp_server"]
diff --git a/tests/integration/test_build_and_run.py b/tests/integration/test_build_and_run.py
index e36f558..cad340a 100644
--- a/tests/integration/test_build_and_run.py
+++ b/tests/integration/test_build_and_run.py
@@ -40,6 +40,7 @@
"gerrit_mcp_server",
"gerrit_mcp_server_task",
"gerrit_mcp_server_depends_on",
+ "gerrit_mcp_server_code_owners",
]
for file_name in self.files_to_copy:
diff --git a/tests/unit/test_code_owners_extension.py b/tests/unit/test_code_owners_extension.py
new file mode 100644
index 0000000..0f5d941
--- /dev/null
+++ b/tests/unit/test_code_owners_extension.py
@@ -0,0 +1,449 @@
+import json
+from pathlib import Path
+from unittest.mock import AsyncMock, MagicMock
+from urllib.parse import quote
+
+import pytest
+
+from gerrit_mcp_server_code_owners.extension import register
+
+# ---------------------------------------------------------------------------
+# Test fixtures and helpers
+# ---------------------------------------------------------------------------
+
+BASE_URL = "https://gerrit.example.com/a"
+
+
+@pytest.fixture
+def ctx(tmp_path: Path):
+ mock_ctx = MagicMock()
+ mock_ctx.get_base_url = MagicMock(return_value=BASE_URL)
+ mock_ctx.normalize_url = lambda u: u
+ mock_ctx.run_curl = AsyncMock()
+ mock_ctx.load_config = MagicMock(return_value={})
+ mock_ctx.log_path = tmp_path / "server.log"
+ mock_ctx.plugin_registry = MagicMock()
+ mock_ctx.plugin_registry.host_has_plugin = AsyncMock(return_value=True)
+ mock_ctx.extension_config = MagicMock(return_value={})
+ return mock_ctx
+
+
+def _capture_tools(ctx) -> dict:
+ """Replace ctx.mcp.tool with a capturing decorator; return registered fns."""
+ registered = {}
+
+ def capture_decorator():
+ def decorator(fn):
+ registered[fn.__name__] = fn
+ return fn
+
+ return decorator
+
+ ctx.mcp.tool = MagicMock(side_effect=lambda: capture_decorator())
+ register(ctx)
+ return registered
+
+
+# ---------------------------------------------------------------------------
+# register()
+# ---------------------------------------------------------------------------
+
+
+class TestRegister:
+ def test_registers_three_tools(self, ctx):
+ register(ctx)
+ assert ctx.mcp.tool.call_count == 3
+
+
+# ---------------------------------------------------------------------------
+# get_code_owner_status
+# ---------------------------------------------------------------------------
+
+
+class TestGetCodeOwnerStatus:
+ def _make_status_response(self, patch_set=1, files=None, more=False):
+ data = {
+ "patch_set_number": patch_set,
+ "file_code_owner_statuses": files or [],
+ }
+ if more:
+ data["more"] = True
+ return json.dumps(data)
+
+ @pytest.mark.asyncio
+ async def test_approved_file(self, ctx):
+ ctx.run_curl = AsyncMock(
+ return_value=self._make_status_response(
+ patch_set=2,
+ files=[
+ {
+ "change_type": "MODIFIED",
+ "new_path_status": {
+ "path": "src/foo.py",
+ "status": "APPROVED",
+ },
+ }
+ ],
+ )
+ )
+ fn = _capture_tools(ctx)["get_code_owner_status"]
+ data = await fn(change_id="100", gerrit_base_url=BASE_URL)
+ assert data["change_id"] == "100"
+ assert data["patch_set_number"] == 2
+ assert len(data["file_code_owner_statuses"]) == 1
+ status = data["file_code_owner_statuses"][0]
+ assert status["change_type"] == "MODIFIED"
+ assert status["new_path_status"]["path"] == "src/foo.py"
+ assert status["new_path_status"]["status"] == "APPROVED"
+
+ @pytest.mark.asyncio
+ async def test_pending_file_with_reasons(self, ctx):
+ ctx.run_curl = AsyncMock(
+ return_value=self._make_status_response(
+ files=[
+ {
+ "new_path_status": {
+ "path": "src/bar.py",
+ "status": "PENDING",
+ "reasons": ["<GERRIT_ACCOUNT_1> is a reviewer"],
+ },
+ }
+ ],
+ )
+ )
+ fn = _capture_tools(ctx)["get_code_owner_status"]
+ data = await fn(change_id="200", gerrit_base_url=BASE_URL)
+ path_status = data["file_code_owner_statuses"][0]["new_path_status"]
+ assert path_status["status"] == "PENDING"
+ assert len(path_status["reasons"]) == 1
+
+ @pytest.mark.asyncio
+ async def test_renamed_file_has_old_and_new_path(self, ctx):
+ ctx.run_curl = AsyncMock(
+ return_value=self._make_status_response(
+ files=[
+ {
+ "change_type": "RENAMED",
+ "old_path_status": {
+ "path": "old/path.py",
+ "status": "APPROVED",
+ },
+ "new_path_status": {
+ "path": "new/path.py",
+ "status": "APPROVED",
+ },
+ }
+ ],
+ )
+ )
+ fn = _capture_tools(ctx)["get_code_owner_status"]
+ data = await fn(change_id="300", gerrit_base_url=BASE_URL)
+ entry = data["file_code_owner_statuses"][0]
+ assert entry["old_path_status"]["path"] == "old/path.py"
+ assert entry["new_path_status"]["path"] == "new/path.py"
+
+ @pytest.mark.asyncio
+ async def test_more_flag_included_when_true(self, ctx):
+ ctx.run_curl = AsyncMock(return_value=self._make_status_response(more=True))
+ fn = _capture_tools(ctx)["get_code_owner_status"]
+ data = await fn(change_id="400", gerrit_base_url=BASE_URL)
+ assert data.get("more") is True
+
+ @pytest.mark.asyncio
+ async def test_more_flag_absent_when_false(self, ctx):
+ ctx.run_curl = AsyncMock(return_value=self._make_status_response(more=False))
+ fn = _capture_tools(ctx)["get_code_owner_status"]
+ data = await fn(change_id="400", gerrit_base_url=BASE_URL)
+ assert data.get("more") is None
+
+ @pytest.mark.asyncio
+ async def test_limit_and_start_in_url(self, ctx):
+ ctx.run_curl = AsyncMock(return_value=self._make_status_response())
+ fn = _capture_tools(ctx)["get_code_owner_status"]
+ await fn(change_id="500", limit=10, start=5, gerrit_base_url=BASE_URL)
+ called_url = ctx.run_curl.call_args[0][0][0]
+ assert "n=10" in called_url
+ assert "S=5" in called_url
+
+ @pytest.mark.asyncio
+ async def test_invalid_json_raises(self, ctx):
+ ctx.run_curl = AsyncMock(return_value="not json")
+ fn = _capture_tools(ctx)["get_code_owner_status"]
+ with pytest.raises(ValueError, match="Could not parse"):
+ await fn(change_id="999", gerrit_base_url=BASE_URL)
+
+ @pytest.mark.asyncio
+ async def test_accounts_populated_when_present(self, ctx):
+ data = {
+ "patch_set_number": 1,
+ "file_code_owner_statuses": [
+ {
+ "new_path_status": {
+ "path": "src/foo.py",
+ "status": "PENDING",
+ "reasons": ["<GERRIT_ACCOUNT_42> is a reviewer"],
+ }
+ }
+ ],
+ "accounts": {
+ "42": {
+ "_account_id": 42,
+ "name": "Alice",
+ "email": "alice@example.com",
+ "username": "alice",
+ }
+ },
+ }
+ ctx.run_curl = AsyncMock(return_value=json.dumps(data))
+ fn = _capture_tools(ctx)["get_code_owner_status"]
+ result = await fn(change_id="600", gerrit_base_url=BASE_URL)
+ assert result["accounts"] is not None
+ assert "42" in result["accounts"]
+ assert result["accounts"]["42"]["account_id"] == 42
+ assert result["accounts"]["42"]["name"] == "Alice"
+
+ @pytest.mark.asyncio
+ async def test_accounts_absent_when_not_in_response(self, ctx):
+ ctx.run_curl = AsyncMock(return_value=self._make_status_response())
+ fn = _capture_tools(ctx)["get_code_owner_status"]
+ result = await fn(change_id="700", gerrit_base_url=BASE_URL)
+ assert result["accounts"] is None
+
+
+# ---------------------------------------------------------------------------
+# get_code_owners_for_path
+# ---------------------------------------------------------------------------
+
+
+class TestGetCodeOwnersForPath:
+ def _make_owners_response(self, owners=None, owned_by_all=False):
+ data = {"code_owners": owners or []}
+ if owned_by_all:
+ data["owned_by_all_users"] = True
+ return json.dumps(data)
+
+ @pytest.mark.asyncio
+ async def test_returns_owners(self, ctx):
+ ctx.run_curl = AsyncMock(
+ return_value=self._make_owners_response(
+ owners=[
+ {
+ "account": {
+ "_account_id": 42,
+ "name": "Alice",
+ "email": "alice@example.com",
+ },
+ "scorings": {"IS_REVIEWER": 1},
+ }
+ ]
+ )
+ )
+ fn = _capture_tools(ctx)["get_code_owners_for_path"]
+ data = await fn(
+ change_id="100",
+ path="src/foo.py",
+ gerrit_base_url=BASE_URL,
+ )
+ assert data["change_id"] == "100"
+ assert data["path"] == "src/foo.py"
+ assert data["revision_id"] == "current"
+ assert len(data["code_owners"]) == 1
+ owner = data["code_owners"][0]
+ assert owner["account"]["account_id"] == 42
+ assert owner["account"]["name"] == "Alice"
+ assert owner["scorings"] == {"IS_REVIEWER": 1}
+
+ @pytest.mark.asyncio
+ async def test_owned_by_all_users(self, ctx):
+ ctx.run_curl = AsyncMock(
+ return_value=self._make_owners_response(owned_by_all=True)
+ )
+ fn = _capture_tools(ctx)["get_code_owners_for_path"]
+ data = await fn(
+ change_id="100",
+ path="OWNERS",
+ gerrit_base_url=BASE_URL,
+ )
+ assert data.get("owned_by_all_users") is True
+
+ @pytest.mark.asyncio
+ async def test_owned_by_all_users_absent_when_false(self, ctx):
+ ctx.run_curl = AsyncMock(return_value=self._make_owners_response())
+ fn = _capture_tools(ctx)["get_code_owners_for_path"]
+ data = await fn(
+ change_id="100",
+ path="src/foo.py",
+ gerrit_base_url=BASE_URL,
+ )
+ assert data.get("owned_by_all_users") is None
+
+ @pytest.mark.asyncio
+ async def test_url_contains_path(self, ctx):
+ ctx.run_curl = AsyncMock(return_value=self._make_owners_response())
+ fn = _capture_tools(ctx)["get_code_owners_for_path"]
+ await fn(
+ change_id="100",
+ path="src/my file.py",
+ revision_id="abc123",
+ gerrit_base_url=BASE_URL,
+ )
+ called_url = ctx.run_curl.call_args[0][0][0]
+ assert "abc123" in called_url
+ assert "src/my%20file.py" in called_url
+
+ @pytest.mark.asyncio
+ async def test_leading_slash_stripped_from_path(self, ctx):
+ ctx.run_curl = AsyncMock(return_value=self._make_owners_response())
+ fn = _capture_tools(ctx)["get_code_owners_for_path"]
+ await fn(
+ change_id="100",
+ path="/src/foo.py",
+ gerrit_base_url=BASE_URL,
+ )
+ called_url = ctx.run_curl.call_args[0][0][0]
+ assert "//src" not in called_url
+
+ @pytest.mark.asyncio
+ async def test_limit_in_url(self, ctx):
+ ctx.run_curl = AsyncMock(return_value=self._make_owners_response())
+ fn = _capture_tools(ctx)["get_code_owners_for_path"]
+ await fn(
+ change_id="100",
+ path="src/foo.py",
+ limit=5,
+ gerrit_base_url=BASE_URL,
+ )
+ called_url = ctx.run_curl.call_args[0][0][0]
+ assert "n=5" in called_url
+
+ @pytest.mark.asyncio
+ async def test_invalid_json_raises(self, ctx):
+ ctx.run_curl = AsyncMock(return_value="not json")
+ fn = _capture_tools(ctx)["get_code_owners_for_path"]
+ with pytest.raises(ValueError, match="Could not parse"):
+ await fn(change_id="100", path="src/foo.py", gerrit_base_url=BASE_URL)
+
+
+# ---------------------------------------------------------------------------
+# check_code_owner
+# ---------------------------------------------------------------------------
+
+
+class TestCheckCodeOwner:
+ def _make_check_response(self, is_code_owner=True, is_resolvable=True, **kwargs):
+ data = {
+ "is_code_owner": is_code_owner,
+ "is_resolvable": is_resolvable,
+ }
+ data.update(kwargs)
+ return json.dumps(data)
+
+ @pytest.mark.asyncio
+ async def test_is_code_owner(self, ctx):
+ ctx.run_curl = AsyncMock(
+ return_value=self._make_check_response(
+ is_code_owner=True,
+ is_resolvable=True,
+ can_read_ref=True,
+ can_approve_change=True,
+ )
+ )
+ fn = _capture_tools(ctx)["check_code_owner"]
+ data = await fn(
+ project="my/project",
+ branch="main",
+ path="src/foo.py",
+ email="alice@example.com",
+ gerrit_base_url=BASE_URL,
+ )
+ assert data["project"] == "my/project"
+ assert data["branch"] == "main"
+ assert data["path"] == "src/foo.py"
+ assert data["email"] == "alice@example.com"
+ assert data["is_code_owner"] is True
+ assert data["is_resolvable"] is True
+ assert data["can_read_ref"] is True
+ assert data["can_approve_change"] is True
+
+ @pytest.mark.asyncio
+ async def test_not_code_owner(self, ctx):
+ ctx.run_curl = AsyncMock(
+ return_value=self._make_check_response(
+ is_code_owner=False,
+ is_resolvable=True,
+ )
+ )
+ fn = _capture_tools(ctx)["check_code_owner"]
+ data = await fn(
+ project="my/project",
+ branch="main",
+ path="src/bar.py",
+ email="bob@example.com",
+ gerrit_base_url=BASE_URL,
+ )
+ assert data["is_code_owner"] is False
+
+ @pytest.mark.asyncio
+ async def test_fallback_owner_fields(self, ctx):
+ ctx.run_curl = AsyncMock(
+ return_value=self._make_check_response(
+ is_fallback_code_owner=True,
+ is_global_code_owner=False,
+ is_default_code_owner=False,
+ annotation=["FALLBACK_CODE_OWNER"],
+ )
+ )
+ fn = _capture_tools(ctx)["check_code_owner"]
+ data = await fn(
+ project="my/project",
+ branch="main",
+ path="src/baz.py",
+ email="charlie@example.com",
+ gerrit_base_url=BASE_URL,
+ )
+ assert data["is_fallback_code_owner"] is True
+ assert data["annotation"] == ["FALLBACK_CODE_OWNER"]
+
+ @pytest.mark.asyncio
+ async def test_url_contains_email_and_path(self, ctx):
+ ctx.run_curl = AsyncMock(return_value=self._make_check_response())
+ fn = _capture_tools(ctx)["check_code_owner"]
+ await fn(
+ project="my/project",
+ branch="main",
+ path="src/foo.py",
+ email="alice@example.com",
+ gerrit_base_url=BASE_URL,
+ )
+ called_url = ctx.run_curl.call_args[0][0][0]
+ assert quote("alice@example.com") in called_url
+ assert quote("src/foo.py") in called_url
+ assert "code_owners.check" in called_url
+
+ @pytest.mark.asyncio
+ async def test_change_id_appended_to_url(self, ctx):
+ ctx.run_curl = AsyncMock(return_value=self._make_check_response())
+ fn = _capture_tools(ctx)["check_code_owner"]
+ await fn(
+ project="my/project",
+ branch="main",
+ path="src/foo.py",
+ email="alice@example.com",
+ change_id="123",
+ gerrit_base_url=BASE_URL,
+ )
+ called_url = ctx.run_curl.call_args[0][0][0]
+ assert "change=123" in called_url
+
+ @pytest.mark.asyncio
+ async def test_invalid_json_raises(self, ctx):
+ ctx.run_curl = AsyncMock(return_value="not json")
+ fn = _capture_tools(ctx)["check_code_owner"]
+ with pytest.raises(ValueError, match="Could not parse"):
+ await fn(
+ project="p",
+ branch="b",
+ path="f",
+ email="e@x.com",
+ gerrit_base_url=BASE_URL,
+ )
diff --git a/uv.lock b/uv.lock
index d6ae7b6..e6efd0c 100644
--- a/uv.lock
+++ b/uv.lock
@@ -11,6 +11,7 @@
[manifest]
members = [
"gerrit-mcp-server",
+ "gerrit-mcp-server-code-owners",
"gerrit-mcp-server-depends-on",
"gerrit-mcp-server-task",
]
@@ -191,6 +192,7 @@
version = "1.0.0"
source = { editable = "." }
dependencies = [
+ { name = "gerrit-mcp-server-code-owners" },
{ name = "gerrit-mcp-server-depends-on" },
{ name = "gerrit-mcp-server-task" },
{ name = "mcp" },
@@ -212,6 +214,7 @@
[package.metadata]
requires-dist = [
+ { name = "gerrit-mcp-server-code-owners", editable = "gerrit_mcp_server_code_owners" },
{ name = "gerrit-mcp-server-depends-on", editable = "gerrit_mcp_server_depends_on" },
{ name = "gerrit-mcp-server-task", editable = "gerrit_mcp_server_task" },
{ name = "mcp", specifier = "<2" },
@@ -229,6 +232,17 @@
provides-extras = ["dev"]
[[package]]
+name = "gerrit-mcp-server-code-owners"
+version = "0.1.0"
+source = { editable = "gerrit_mcp_server_code_owners" }
+dependencies = [
+ { name = "gerrit-mcp-server" },
+]
+
+[package.metadata]
+requires-dist = [{ name = "gerrit-mcp-server", editable = "." }]
+
+[[package]]
name = "gerrit-mcp-server-depends-on"
version = "0.1.0"
source = { editable = "gerrit_mcp_server_depends_on" }
