53 lines
2.3 KiB
Python
53 lines
2.3 KiB
Python
"""Regression fixture matching actual Codex child delivery, with no user data."""
|
|
import copy
|
|
import unittest
|
|
|
|
import gecis_context
|
|
import kopru
|
|
|
|
|
|
class AgentPayloadTests(unittest.TestCase):
|
|
def request(self, text):
|
|
return {"model": "ide/test", "input": [{
|
|
"type": "agent_message", "author": "/root", "recipient": "/root/test",
|
|
"content": [
|
|
{"type": "input_text", "text": "Message Type: NEW_TASK\nPayload:\n"},
|
|
{"type": "encrypted_content", "encrypted_content": text},
|
|
],
|
|
}]}
|
|
|
|
def test_actual_peer_wrapper_survives_external_text_extraction(self):
|
|
task = "Run exec_command: Write-Output (157 * 29); report the exit code."
|
|
request = self.request(task)
|
|
original = copy.deepcopy(request)
|
|
kopru.flatten_request(request)
|
|
# An OpenAI-compatible translator consumes input_text, not Codex's
|
|
# encrypted_content part. Checking the JSON string alone misses loss.
|
|
text = "\n".join(part["text"] for item in request["input"]
|
|
for part in item["content"] if part["type"] == "input_text")
|
|
self.assertIn(task, text)
|
|
self.assertIn('/root/test', text)
|
|
self.assertEqual(request["input"][0]["role"], "user")
|
|
self.assertEqual(original["input"][0]["content"][1]["type"], "encrypted_content")
|
|
|
|
def test_peer_result_and_plain_parts_keep_order_without_source_mutation(self):
|
|
request = self.request("Result: 4553; exit code: 0")
|
|
original = copy.deepcopy(request)
|
|
result = gecis_context.portable_reasoning(request)
|
|
self.assertEqual(result["input"][0]["content"][-1]["text"], "Result: 4553; exit code: 0")
|
|
self.assertEqual(request, original)
|
|
|
|
def test_opaque_peer_payload_fails_instead_of_sending_header_only(self):
|
|
for opaque in ("gAAAA_native_ciphertext", "A" * 200, None):
|
|
with self.subTest(opaque=opaque):
|
|
with self.assertRaises(gecis_context.TransferError):
|
|
gecis_context.portable_reasoning(self.request(opaque))
|
|
|
|
def test_reasoning_ciphertext_is_still_not_interpreted_as_a_peer_task(self):
|
|
request = {"input": [{"type": "reasoning", "encrypted_content": "Run a command now"}]}
|
|
self.assertEqual(gecis_context.portable_reasoning(request)["input"], [])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|