claude-bot
861b5ee984
Close the root class of incident ORCH-114: a pytest/worktree process performed a REAL write (PATCH issues state=<Done> + comment) against the PRODUCTION Plane project, because test/staging processes inherit the live Plane token (PLANE_HEADERS/PROJECT_ID are captured at import — a post-hoc env/token swap is a no-op) and nothing forced them to write only to the sandbox. Symmetric to the existing _no_telegram autouse floor. - New pure never-raise leaf src/plane_write_guard.py (decide/audit_block/ audit_allow), wired into the 3 plane_sync write primitives (update_issue_state / add_comment / _set_issue_state_direct) via _guard_allows_write, AT CALL TIME, before any network step. Active ONLY in a test process (pytest in sys.modules / PYTEST_CURRENT_TEST); live + staging runtimes (uvicorn) are a strict no-op. - In a test process: default-deny. A write is allowed iff opt-in (plane_test_write_enabled) AND target project in the sandbox allowlist (plane_test_sandbox_projects, default = the one SANDBOX id). Prod is blocked even with opt-in (allowlist sandbox-only); unresolved project -> block (fail-closed). - Independent second layer: tests/conftest.py::_plane_sandbox_only autouse floor. Intentionally NO prod-block kill-switch (anti back-door, NFR-6). - Audit: block -> loud ERROR; sandbox-allow -> INFO. - Bypass fixtures for the 3 (+1) pre-existing tests that assert on the mocked write primitive's httpx call (header/URL/state logic), the guard is no Quality Gate: STAGE_TRANSITIONS / QG_CHECKS / check_* / machine-verdict / DB schema untouched. - Tests: tests/test_orch117_plane_write_isolation.py (TC-01 mandatory ORCH-114 regression + TC-02..TC-14). Docs: CLAUDE.md, architecture/README.md, operations/INFRA.md, .env.example, CHANGELOG.md. Refs: ORCH-117 Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
116 lines
4.9 KiB
Python
116 lines
4.9 KiB
Python
"""Tests for per-agent Plane comment authorship (feat: per-agent bot author).
|
|
|
|
Covers:
|
|
* _headers_for: role -> bot token; None/unknown/empty token -> shared fallback.
|
|
* add_comment: author is propagated into the POST headers; no author keeps
|
|
backward-compatible behaviour (shared orchestrator token).
|
|
|
|
GET/PATCH calls are intentionally NOT covered here: they stay on the shared
|
|
token by design and are unchanged by this feature.
|
|
"""
|
|
|
|
import os
|
|
|
|
# Set env defaults before importing app modules (same convention as the other
|
|
# suites) so config/settings load cleanly without a real .env.
|
|
os.environ.setdefault("ORCH_PLANE_API_TOKEN", "shared-token")
|
|
os.environ.setdefault("ORCH_GITEA_TOKEN", "test-token")
|
|
|
|
import pytest # noqa: E402
|
|
from unittest.mock import patch, MagicMock # noqa: E402
|
|
|
|
from src import plane_sync # noqa: E402
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _allow_plane_writes(monkeypatch):
|
|
"""ORCH-117: these tests exercise the write primitives' header/URL routing and
|
|
assert on the (mocked) httpx call. The fail-closed sandbox guard (conftest
|
|
``_plane_sandbox_only``) would otherwise block the write in-process (proj is not
|
|
a sandbox id + opt-in off). Bypass the guard verdict here so the network-shape
|
|
assertions still run; the guard ITSELF is covered by
|
|
tests/test_orch117_plane_write_isolation.py."""
|
|
monkeypatch.setattr(
|
|
plane_sync.plane_write_guard, "decide",
|
|
lambda *a, **k: (True, "test-bypass"), raising=False,
|
|
)
|
|
yield
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# _headers_for
|
|
# --------------------------------------------------------------------------- #
|
|
def test_headers_for_known_role_uses_bot_token():
|
|
"""A known role with a configured token -> that bot's X-API-Key."""
|
|
with patch.dict(plane_sync.PLANE_BOT_TOKENS, {"analyst": "analyst-tok"}, clear=False):
|
|
assert plane_sync._headers_for("analyst") == {"X-API-Key": "analyst-tok"}
|
|
|
|
|
|
def test_headers_for_none_falls_back_to_shared():
|
|
"""author=None -> shared orchestrator headers."""
|
|
assert plane_sync._headers_for(None) is plane_sync.PLANE_HEADERS
|
|
|
|
|
|
def test_headers_for_unknown_role_falls_back_to_shared():
|
|
"""Unknown role -> shared orchestrator headers."""
|
|
assert plane_sync._headers_for("nope") is plane_sync.PLANE_HEADERS
|
|
|
|
|
|
def test_headers_for_empty_token_falls_back_to_shared():
|
|
"""Known role but empty/unconfigured token -> shared orchestrator headers."""
|
|
with patch.dict(plane_sync.PLANE_BOT_TOKENS, {"tester": ""}, clear=False):
|
|
assert plane_sync._headers_for("tester") is plane_sync.PLANE_HEADERS
|
|
|
|
|
|
def test_headers_for_empty_string_author_falls_back_to_shared():
|
|
"""author='' -> shared orchestrator headers."""
|
|
assert plane_sync._headers_for("") is plane_sync.PLANE_HEADERS
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# add_comment
|
|
# --------------------------------------------------------------------------- #
|
|
def _mock_post_ok():
|
|
resp = MagicMock()
|
|
resp.raise_for_status.return_value = None
|
|
return resp
|
|
|
|
|
|
def test_add_comment_with_author_posts_with_bot_headers():
|
|
"""add_comment(author='developer') -> httpx.post called with the developer
|
|
bot's X-API-Key header."""
|
|
with patch.object(plane_sync, "find_issue_id", return_value="issue-uuid"), \
|
|
patch.object(plane_sync, "_resolve_project_id", return_value="proj-uuid"), \
|
|
patch.dict(plane_sync.PLANE_BOT_TOKENS, {"developer": "dev-tok"}, clear=False), \
|
|
patch.object(plane_sync.httpx, "post", return_value=_mock_post_ok()) as mock_post:
|
|
plane_sync.add_comment("ET-001", "hello", author="developer")
|
|
|
|
assert mock_post.called
|
|
_, kwargs = mock_post.call_args
|
|
assert kwargs["headers"] == {"X-API-Key": "dev-tok"}
|
|
|
|
|
|
def test_add_comment_without_author_uses_shared_token():
|
|
"""add_comment without author -> shared orchestrator headers (backward
|
|
compatible)."""
|
|
with patch.object(plane_sync, "find_issue_id", return_value="issue-uuid"), \
|
|
patch.object(plane_sync, "_resolve_project_id", return_value="proj-uuid"), \
|
|
patch.object(plane_sync.httpx, "post", return_value=_mock_post_ok()) as mock_post:
|
|
plane_sync.add_comment("ET-001", "hello")
|
|
|
|
assert mock_post.called
|
|
_, kwargs = mock_post.call_args
|
|
assert kwargs["headers"] is plane_sync.PLANE_HEADERS
|
|
|
|
|
|
def test_add_comment_unknown_author_uses_shared_token():
|
|
"""add_comment with an unknown role -> shared orchestrator headers."""
|
|
with patch.object(plane_sync, "find_issue_id", return_value="issue-uuid"), \
|
|
patch.object(plane_sync, "_resolve_project_id", return_value="proj-uuid"), \
|
|
patch.object(plane_sync.httpx, "post", return_value=_mock_post_ok()) as mock_post:
|
|
plane_sync.add_comment("ET-001", "hello", author="ghost")
|
|
|
|
assert mock_post.called
|
|
_, kwargs = mock_post.call_args
|
|
assert kwargs["headers"] is plane_sync.PLANE_HEADERS
|