smashimo/oh-my-pi
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(python/robomp): implemented has_authorized_impl_event in the

Browse Source 
- Implement `has_authorized_impl_event` in the database to retrieve historical authorization state.
- Update `_enforce_impl_authorization` to permit actions if prior events on the issue provided implementation authorization.
- Normalize maintainer logins by stripping `[bot]` suffixes and allow match-regex to ignore them.
This commit is contained in:
can1357
2026-06-21 19:18:37 +02:00
parent 4b2e4085e0
commit 24a0c8428a
10 changed files with 140 additions and 29 deletions
Download Patch File Download Diff File Expand all files Collapse all files
python/robomp/tests/test_host_tools.py
+91 -12
View File
@@ -1181,8 +1181,9 @@ def test_review_mode_rejects_push_and_open_pr_before_repo_commands(db: Database,
assert calls == []
def test_impl_gate_rejects_unauthorized_proposal_before_repo_commands(
db: Database, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
@pytest.mark.parametrize("classification", ["enhancement", "proposal"])
def test_impl_gate_rejects_unauthorized_non_auto_classification_before_repo_commands(
db: Database, tmp_path: Path, monkeypatch: pytest.MonkeyPatch, classification: str
) -> None:
calls: list[list[str] | tuple[str, ...]] = []
@@ -1192,7 +1193,7 @@ def test_impl_gate_rejects_unauthorized_proposal_before_repo_commands(
raise AssertionError("repo command must not run before implementation authorization")
bindings, loop, t = _bindings(db, tmp_path, httpx.MockTransport(lambda _r: httpx.Response(500)))
db.set_issue_classification(bindings.issue_key, "proposal")
db.set_issue_classification(bindings.issue_key, classification)
monkeypatch.setattr(host_tools, "_run_repo_command", record_repo_command)
try:
push = next(x for x in build(bindings) if x.name == "gh_push_branch")
@@ -1205,7 +1206,7 @@ def test_impl_gate_rejects_unauthorized_proposal_before_repo_commands(
_stop_loop(loop, t)
for msg in (str(push_exc.value), str(pr_exc.value)):
assert "classified `proposal`" in msg
assert f"classified `{classification}`" in msg
assert "OWNER or allowlisted maintainer" in msg
assert "gh_post_comment" in msg
assert calls == []
@@ -1213,14 +1214,17 @@ def test_impl_gate_rejects_unauthorized_proposal_before_repo_commands(
"SELECT tool, error FROM tool_calls WHERE tool IN ('gh_push_branch', 'gh_open_pr') ORDER BY id"
).fetchall()
assert [row["tool"] for row in rows] == ["gh_push_branch", "gh_open_pr"]
assert all("classified `proposal`" in row["error"] for row in rows)
assert all(f"classified `{classification}`" in row["error"] for row in rows)
def test_impl_gate_allows_authorized_proposal_to_reach_pr_validation(db: Database, tmp_path: Path) -> None:
@pytest.mark.parametrize("classification", ["enhancement", "proposal"])
def test_impl_gate_allows_authorized_non_auto_classification_to_reach_pr_validation(
db: Database, tmp_path: Path, classification: str
) -> None:
from dataclasses import replace
bindings, loop, t = _bindings(db, tmp_path, httpx.MockTransport(lambda _r: httpx.Response(500)))
db.set_issue_classification(bindings.issue_key, "proposal")
db.set_issue_classification(bindings.issue_key, classification)
bindings = replace(bindings, impl_authorized=True)
try:
tool = next(x for x in build(bindings) if x.name == "gh_open_pr")
@@ -1234,8 +1238,9 @@ def test_impl_gate_allows_authorized_proposal_to_reach_pr_validation(db: Databas
assert "OWNER or allowlisted maintainer" not in msg
def test_impl_gate_allows_authorized_proposal_push_to_reach_repo_commands(
db: Database, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
@pytest.mark.parametrize("classification", ["enhancement", "proposal"])
def test_impl_gate_allows_authorized_non_auto_classification_push_to_reach_repo_commands(
db: Database, tmp_path: Path, monkeypatch: pytest.MonkeyPatch, classification: str
) -> None:
from dataclasses import replace
@@ -1244,15 +1249,15 @@ def test_impl_gate_allows_authorized_proposal_push_to_reach_repo_commands(
def record_repo_command(_bindings: ToolBindings, cmd: list[str] | tuple[str, ...], *, timeout: float | None = None):
del timeout
calls.append(cmd)
raise RuntimeError("authorized proposal reached gh_push_branch repo command")
raise RuntimeError("authorized non-auto issue reached gh_push_branch repo command")
bindings, loop, t = _bindings(db, tmp_path, httpx.MockTransport(lambda _r: httpx.Response(500)))
db.set_issue_classification(bindings.issue_key, "proposal")
db.set_issue_classification(bindings.issue_key, classification)
bindings = replace(bindings, impl_authorized=True)
monkeypatch.setattr(host_tools, "_run_repo_command", record_repo_command)
try:
tool = next(x for x in build(bindings) if x.name == "gh_push_branch")
with pytest.raises(RuntimeError, match="authorized proposal reached gh_push_branch repo command"):
with pytest.raises(RuntimeError, match="authorized non-auto issue reached gh_push_branch repo command"):
tool.execute({}, _ctx())
finally:
_stop_loop(loop, t)
@@ -1260,6 +1265,80 @@ def test_impl_gate_allows_authorized_proposal_push_to_reach_repo_commands(
assert calls
def test_impl_gate_allows_later_authorized_event_to_reach_repo_commands(
db: Database, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
calls: list[list[str] | tuple[str, ...]] = []
def record_repo_command(_bindings: ToolBindings, cmd: list[str] | tuple[str, ...], *, timeout: float | None = None):
del timeout
calls.append(cmd)
raise RuntimeError("later authorized event reached gh_push_branch repo command")
bindings, loop, t = _bindings(db, tmp_path, httpx.MockTransport(lambda _r: httpx.Response(500)))
db.set_issue_classification(bindings.issue_key, "enhancement")
db.record_event(
delivery_id="auth-event",
event_type="issue_comment",
repo=bindings.issue.repo,
issue_key=bindings.issue_key,
payload={
"_robomp_directive": {
"body": "go ahead",
"author": "can1357",
"pragmas": [],
"authorizes_impl": True,
}
},
)
monkeypatch.setattr(host_tools, "_run_repo_command", record_repo_command)
try:
tool = next(x for x in build(bindings) if x.name == "gh_push_branch")
with pytest.raises(RuntimeError, match="later authorized event reached gh_push_branch repo command"):
tool.execute({}, _ctx())
finally:
_stop_loop(loop, t)
assert calls
def test_impl_gate_ignores_skipped_authorized_event(db: Database, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
calls: list[list[str] | tuple[str, ...]] = []
def record_repo_command(_bindings: ToolBindings, cmd: list[str] | tuple[str, ...], *, timeout: float | None = None):
del timeout
calls.append(cmd)
raise AssertionError("skipped authorization must not reach repo command")
bindings, loop, t = _bindings(db, tmp_path, httpx.MockTransport(lambda _r: httpx.Response(500)))
db.set_issue_classification(bindings.issue_key, "enhancement")
db.record_event(
delivery_id="skipped-auth-event",
event_type="issue_comment",
repo=bindings.issue.repo,
issue_key=bindings.issue_key,
payload={
"_robomp_directive": {
"body": "go ahead",
"author": "can1357",
"pragmas": [],
"authorizes_impl": True,
}
},
state="skipped",
)
monkeypatch.setattr(host_tools, "_run_repo_command", record_repo_command)
try:
tool = next(x for x in build(bindings) if x.name == "gh_push_branch")
with pytest.raises(RpcCommandError) as exc:
tool.execute({}, _ctx())
finally:
_stop_loop(loop, t)
assert "OWNER or allowlisted maintainer" in str(exc.value)
assert calls == []
def test_impl_gate_allows_bug_without_directive_to_reach_pr_validation(db: Database, tmp_path: Path) -> None:
bindings, loop, t = _bindings(db, tmp_path, httpx.MockTransport(lambda _r: httpx.Response(500)))
db.set_issue_classification(bindings.issue_key, "bug")