test(workspace): update 3 assertions for OFFSEC-003 boundary wrapping (fixes #508)
Some checks failed
audit-force-merge / audit (pull_request) Has been skipped
Block internal-flavored paths / Block forbidden paths (pull_request) Successful in 13s
CI / Python Lint & Test (pull_request) Failing after 6m57s
Secret scan / Scan diff for credential-shaped strings (pull_request) Successful in 13s
sop-tier-check / tier-check (pull_request) Successful in 16s
E2E API Smoke Test / detect-changes (pull_request) Successful in 32s
CI / Detect changes (pull_request) Successful in 35s
E2E Staging Canvas (Playwright) / detect-changes (pull_request) Successful in 34s
Handlers Postgres Integration / detect-changes (pull_request) Successful in 36s
Runtime PR-Built Compatibility / PR-built wheel + import smoke (pull_request) Successful in 2m13s
Runtime PR-Built Compatibility / detect-changes (pull_request) Successful in 35s
E2E API Smoke Test / E2E API Smoke Test (pull_request) Successful in 7s
CI / Platform (Go) (pull_request) Successful in 6s
CI / Shellcheck (E2E scripts) (pull_request) Successful in 6s
CI / Canvas (Next.js) (pull_request) Successful in 7s
CI / Canvas Deploy Reminder (pull_request) Has been skipped
Handlers Postgres Integration / Handlers Postgres Integration (pull_request) Successful in 8s
E2E Staging Canvas (Playwright) / Canvas tabs E2E (pull_request) Successful in 10s

PR #477 added _A2A_BOUNDARY_START/END wrapping to tool_delegate_task's
success path. Three tests in test_delegation_sync_via_polling.py were
still asserting exact plain-text returns from the patched send_a2a_message
and _delegate_sync_via_polling mocks — both paths now return boundary-
wrapped text.

Updated assertions:
- test_flag_off_uses_send_a2a_message_not_polling: legacy ok
- test_queued_sentinel_triggers_polling_fallback: poll-mode response
- test_non_queued_send_result_does_not_trigger_fallback: normal reply

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
This commit is contained in:
Molecule AI · core-devops 2026-05-11 16:28:25 +00:00
parent 40ca44aa4d
commit 37a94ef6c3

View File

@ -30,6 +30,10 @@ import pytest
os.environ.setdefault("WORKSPACE_ID", "00000000-0000-0000-0000-000000000001")
os.environ.setdefault("PLATFORM_URL", "http://localhost:8080")
# OFFSEC-003: tool_delegate_task wraps non-error results in boundary markers
# so the agent can distinguish trusted own output from untrusted peer content.
from a2a_tools_delegation import _A2A_BOUNDARY_END, _A2A_BOUNDARY_START
def _resp(status_code, payload, text=None):
r = MagicMock()
@ -88,7 +92,10 @@ class TestFlagOffLegacyPath:
"ws-target", "task body", source_workspace_id="ws-self"
)
assert result == "legacy ok", f"expected legacy passthrough, got {result!r}"
# OFFSEC-003: boundary wrapping is applied by tool_delegate_task even on
# the legacy send_a2a_message path (sanitize then wrap at line 333-334).
assert _A2A_BOUNDARY_START in result and "legacy ok" in result, \
f"expected boundary-wrapped legacy ok, got {result!r}"
assert send_calls == [("ws-target", "task body", "ws-self")]
poll_mock.assert_not_called()
@ -153,7 +160,10 @@ class TestPollModeAutoFallback:
assert poll_calls[0] == ("ws-target", "task body", "ws-self")
# Caller sees the real reply, NOT the queued sentinel and NOT
# a DELEGATION FAILED string.
assert result == "real response from poll-mode peer"
# OFFSEC-003: _delegate_sync_via_polling returns sanitized plain text;
# tool_delegate_task wraps it in boundary markers before returning.
assert _A2A_BOUNDARY_START in result and "real response from poll-mode peer" in result, \
f"expected boundary-wrapped response, got {result!r}"
async def test_non_queued_send_result_does_not_trigger_fallback(self, monkeypatch):
# Push-mode peer returns a normal text reply — fallback path
@ -179,7 +189,9 @@ class TestPollModeAutoFallback:
"ws-target", "task", source_workspace_id="ws-self"
)
assert result == "normal reply"
# OFFSEC-003: boundary wrapping applied by tool_delegate_task before return.
assert _A2A_BOUNDARY_START in result and "normal reply" in result, \
f"expected boundary-wrapped normal reply, got {result!r}"
poll_mock.assert_not_called()
async def test_error_send_result_does_not_trigger_fallback(self, monkeypatch):