can1357
4b2e4085e0
- Implemented case-insensitive normalization for bot logins to handle mention handles and `[bot]` suffixes consistently. - Added support for `ROBOMP_MAINTAINER_LOGINS` to allow authorized non-owner users to execute implementations. - Refined authorization logic to distinguish between personal repository owners and organizational accounts. - Updated documentation and added comprehensive tests to verify authorization handling across tasks, workers, and directive processing.
267 lines
8.4 KiB
Python
267 lines
8.4 KiB
Python
"""Verify pragmas survive the payload round-trip from server → durable queue → tasks."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from types import SimpleNamespace
|
|
|
|
import pytest
|
|
from robomp import tasks
|
|
from robomp.github_client import IssueInfo, RepoInfo
|
|
from robomp.tasks import _attach_thread, _directive_from_payload
|
|
from robomp.worker import DirectiveInfo
|
|
|
|
|
|
def test_directive_from_payload_parses_pragmas() -> None:
|
|
directive = _directive_from_payload(
|
|
{
|
|
"_robomp_directive": {
|
|
"body": "do the thing",
|
|
"author": "can1357",
|
|
"pragmas": [["model", "gpt"], ["thinking", "low"]],
|
|
}
|
|
}
|
|
)
|
|
assert directive is not None
|
|
assert directive.body == "do the thing"
|
|
assert directive.author == "can1357"
|
|
assert directive.pragmas == (("model", "gpt"), ("thinking", "low"))
|
|
assert directive.authorizes_impl is False
|
|
|
|
|
|
def test_directive_from_payload_missing_pragmas_is_empty_tuple() -> None:
|
|
directive = _directive_from_payload({"_robomp_directive": {"body": "x", "author": "can1357"}})
|
|
assert directive is not None
|
|
assert directive.pragmas == ()
|
|
assert directive.authorizes_impl is False
|
|
|
|
|
|
def test_directive_from_payload_drops_malformed_pragma_entries() -> None:
|
|
directive = _directive_from_payload(
|
|
{
|
|
"_robomp_directive": {
|
|
"body": "x",
|
|
"author": "can1357",
|
|
"pragmas": [
|
|
["model", "gpt"],
|
|
["bad"], # wrong arity
|
|
[1, "v"], # non-string key
|
|
"string-instead-of-pair",
|
|
],
|
|
}
|
|
}
|
|
)
|
|
assert directive is not None
|
|
assert directive.pragmas == (("model", "gpt"),)
|
|
|
|
|
|
def test_directive_from_payload_parses_implementation_authorization() -> None:
|
|
directive = _directive_from_payload(
|
|
{
|
|
"_robomp_directive": {
|
|
"body": "do the thing",
|
|
"author": "can1357",
|
|
"authorizes_impl": True,
|
|
}
|
|
}
|
|
)
|
|
assert directive is not None
|
|
assert directive.authorizes_impl is True
|
|
|
|
|
|
def test_directive_from_payload_returns_none_for_missing_directive() -> None:
|
|
assert _directive_from_payload({}) is None
|
|
assert _directive_from_payload({"_robomp_directive": "not-a-mapping"}) is None
|
|
|
|
|
|
async def test_attach_thread_preserves_authorizes_impl(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
async def fake_fetch_thread(*args, **kwargs):
|
|
return ()
|
|
|
|
monkeypatch.setattr("robomp.tasks._fetch_thread", fake_fetch_thread)
|
|
|
|
directive = DirectiveInfo(
|
|
body="test body",
|
|
author="test_author",
|
|
authorizes_impl=True,
|
|
)
|
|
hydrated = await _attach_thread(None, directive, "owner/repo", 42, is_pr=False)
|
|
assert hydrated is not None
|
|
assert hydrated.body == "test body"
|
|
assert hydrated.author == "test_author"
|
|
assert hydrated.authorizes_impl is True
|
|
|
|
|
|
|
|
def _payload_with_directive(*, issue_number: int, body: str = "@robomp-bot ship it") -> dict[str, object]:
|
|
return {
|
|
"repository": {
|
|
"full_name": "octo/widget",
|
|
"default_branch": "main",
|
|
"clone_url": "https://x/octo/widget.git",
|
|
"private": False,
|
|
},
|
|
"issue": {
|
|
"number": issue_number,
|
|
"title": "proposal",
|
|
"body": "issue body",
|
|
"state": "open",
|
|
"user": {"login": "alice"},
|
|
"labels": [{"name": "proposal"}],
|
|
},
|
|
"comment": {
|
|
"id": 99,
|
|
"body": body,
|
|
"created_at": "2026-01-01T00:00:00Z",
|
|
"user": {"login": "owner"},
|
|
},
|
|
"_robomp_directive": {
|
|
"body": body,
|
|
"author": "owner",
|
|
"authorizes_impl": True,
|
|
},
|
|
}
|
|
|
|
|
|
def _workspace(tmp_path):
|
|
repo_dir = tmp_path / "repo"
|
|
session_dir = tmp_path / "session"
|
|
repo_dir.mkdir(exist_ok=True)
|
|
session_dir.mkdir(exist_ok=True)
|
|
return SimpleNamespace(root=tmp_path, repo_dir=repo_dir, session_dir=session_dir, branch="robomp/issue-42")
|
|
|
|
|
|
async def test_handle_comment_preserves_authorizes_impl_to_run_task(
|
|
db, tmp_path, settings, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
workspace = _workspace(tmp_path)
|
|
captured: dict[str, object] = {}
|
|
|
|
async def fake_attach_thread(
|
|
_github: object,
|
|
directive: DirectiveInfo | None,
|
|
_repo: str,
|
|
_number: int,
|
|
*,
|
|
is_pr: bool,
|
|
) -> DirectiveInfo | None:
|
|
assert directive is not None
|
|
assert is_pr is False
|
|
captured["attached_authorizes_impl"] = directive.authorizes_impl
|
|
return directive
|
|
|
|
async def fake_run_task(
|
|
*,
|
|
task_kind: str,
|
|
inputs: object,
|
|
directive: DirectiveInfo | None = None,
|
|
**_kwargs: object,
|
|
) -> None:
|
|
del inputs
|
|
captured["task_kind"] = task_kind
|
|
captured["run_task_authorizes_impl"] = directive.authorizes_impl if directive is not None else None
|
|
|
|
monkeypatch.setattr(tasks, "_attach_thread", fake_attach_thread)
|
|
monkeypatch.setattr(tasks, "run_task", fake_run_task)
|
|
|
|
await tasks.handle_comment(
|
|
settings=settings,
|
|
db=db,
|
|
github=SimpleNamespace(),
|
|
sandbox=SimpleNamespace(natives_cache=None, ensure_workspace=lambda **_kwargs: workspace),
|
|
git_transport=SimpleNamespace(),
|
|
payload=_payload_with_directive(issue_number=42),
|
|
delivery_id="d-comment",
|
|
)
|
|
|
|
assert captured == {
|
|
"attached_authorizes_impl": True,
|
|
"task_kind": "triage_issue",
|
|
"run_task_authorizes_impl": True,
|
|
}
|
|
|
|
|
|
async def test_handle_pr_conversation_preserves_authorizes_impl_to_run_task(
|
|
db, tmp_path, settings, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
workspace = _workspace(tmp_path)
|
|
db.upsert_issue(
|
|
key="octo/widget#42",
|
|
repo="octo/widget",
|
|
number=42,
|
|
state="opened",
|
|
branch=workspace.branch,
|
|
session_dir=str(workspace.session_dir),
|
|
pr_number=7,
|
|
)
|
|
captured: dict[str, object] = {}
|
|
|
|
class FakeGitHub:
|
|
async def get_repo(self, repo: str) -> RepoInfo:
|
|
assert repo == "octo/widget"
|
|
return RepoInfo(full_name=repo, default_branch="main", clone_url="https://x/octo/widget.git", private=False)
|
|
|
|
async def get_issue(self, repo: str, number: int) -> IssueInfo:
|
|
assert repo == "octo/widget"
|
|
assert number == 42
|
|
return IssueInfo(
|
|
repo=repo,
|
|
number=number,
|
|
title="proposal",
|
|
body="issue body",
|
|
state="open",
|
|
author="alice",
|
|
labels=("proposal",),
|
|
is_pull_request=False,
|
|
)
|
|
|
|
async def fake_attach_thread(
|
|
_github: object,
|
|
directive: DirectiveInfo | None,
|
|
repo: str,
|
|
number: int,
|
|
*,
|
|
is_pr: bool,
|
|
) -> DirectiveInfo | None:
|
|
assert directive is not None
|
|
assert repo == "octo/widget"
|
|
assert number == 7
|
|
assert is_pr is True
|
|
captured["attached_authorizes_impl"] = directive.authorizes_impl
|
|
return directive
|
|
|
|
async def fake_run_task(
|
|
*,
|
|
task_kind: str,
|
|
inputs: object,
|
|
pr_number: int | None = None,
|
|
directive: DirectiveInfo | None = None,
|
|
**_kwargs: object,
|
|
) -> None:
|
|
del inputs
|
|
captured["task_kind"] = task_kind
|
|
captured["pr_number"] = pr_number
|
|
captured["run_task_authorizes_impl"] = directive.authorizes_impl if directive is not None else None
|
|
|
|
monkeypatch.setattr(tasks, "_attach_thread", fake_attach_thread)
|
|
monkeypatch.setattr(tasks, "run_task", fake_run_task)
|
|
|
|
payload = _payload_with_directive(issue_number=7)
|
|
issue_payload = payload["issue"]
|
|
assert isinstance(issue_payload, dict)
|
|
issue_payload["pull_request"] = {"url": "https://api.github.com/repos/octo/widget/pulls/7"}
|
|
await tasks.handle_pr_conversation(
|
|
settings=settings,
|
|
db=db,
|
|
github=FakeGitHub(),
|
|
sandbox=SimpleNamespace(natives_cache=None, ensure_workspace=lambda **_kwargs: workspace),
|
|
git_transport=SimpleNamespace(),
|
|
payload=payload,
|
|
delivery_id="d-pr-comment",
|
|
)
|
|
|
|
assert captured == {
|
|
"attached_authorizes_impl": True,
|
|
"task_kind": "handle_comment",
|
|
"pr_number": 7,
|
|
"run_task_authorizes_impl": True,
|
|
} |