Compare commits

...

1 Commits

Author SHA1 Message Date
core-devops b20ae0ad8c fix(workspace): TestPollingPathSanitization mock signature bug
- fake_discover: add source_workspace_id keyword arg (discover_peer
  signature changed but the mock was not updated — TypeError on call).
- fake_delegate_sync: return unescaped text (matching what the real
  _delegate_sync_via_polling returns via sanitize_a2a_result), not
  pre-boundary-wrapped text.
- Add DELEGATION_SYNC_VIA_INBOX=1 so test routes through the mocked
  polling path (without it, send_a2a_message is called which validates
  workspace_id as a UUID and fails for test peer "ws-peer").
- Update docstring to correctly describe the single-wrap chain.

Closes #495.
2026-05-11 15:23:51 +00:00
+16 -10
View File
@@ -190,26 +190,32 @@ class TestPollingPathSanitization:
"""
def test_completed_response_sanitized(self):
"""_delegate_sync_via_polling returns sanitize_a2a_result(...), which
wraps in boundary markers. tool_delegate_task wraps AGAIN, so the
final result contains the wrapped content."""
"""_delegate_sync_via_polling returns sanitize_a2a_result(text), i.e.
escaped content WITHOUT boundary markers. tool_delegate_task then wraps
the sanitized text in [A2A_RESULT_FROM_PEER] boundary markers (OFFSEC-003).
This test verifies the full chain: fake polling result → tool wrapper."""
import asyncio
import os
import a2a_tools_delegation as d
# _delegate_sync_via_polling returns sanitize_a2a_result(text), i.e.
# the escaped (no boundary) form. tool_delegate_task wraps once more.
async def fake_delegate_sync(ws_id, task, src):
return "[A2A_RESULT_FROM_PEER]\nSanitized peer reply.\n[/A2A_RESULT_FROM_PEER]"
# Route through _delegate_sync_via_polling (not send_a2a_message) so
# we can mock the polling path. send_a2a_message validates workspace_id
# as a UUID which would fail for our test peer "ws-peer".
os.environ["DELEGATION_SYNC_VIA_INBOX"] = "1"
async def fake_discover(ws_id):
# _delegate_sync_via_polling returns sanitize_a2a_result(text) — escaped
# text, no boundary markers. tool_delegate_task adds the boundary wrap.
async def fake_delegate_sync(ws_id, task, src):
return "Sanitized peer reply."
async def fake_discover(ws_id, source_workspace_id=None):
return {"id": ws_id, "url": "http://x/a2a", "name": "Peer"}
d._delegate_sync_via_polling = fake_delegate_sync
d.discover_peer = fake_discover
result = asyncio.run(d.tool_delegate_task("ws-peer", "do it"))
# tool_delegate_task wraps the already-wrapped polling result in
# another layer of boundary markers.
# tool_delegate_task wraps the sanitized polling result in boundary markers.
assert "[A2A_RESULT_FROM_PEER]" in result
assert "[/A2A_RESULT_FROM_PEER]" in result
assert "Sanitized peer reply" in result