| 1 | import unittest |
| 2 | from gateway import normalize, normalize_anthropic, restore |
| 3 | |
| 4 | |
| 5 | class ResponsesCompatibility(unittest.TestCase): |
| 6 | def test_anthropic_effort_matches_the_qwen_template(self): |
| 7 | for requested, expected in [("low", "low"), ("medium", "medium"), ("high", "xhigh"), ("max", "xhigh"), ("xhigh", "xhigh")]: |
| 8 | with self.subTest(effort=requested): |
| 9 | data = normalize_anthropic({"output_config": {"effort": requested, "format": {"type": "json_schema"}}}) |
| 10 | self.assertEqual(data["output_config"], {"effort": expected, "format": {"type": "json_schema"}}) |
| 11 | |
| 12 | def test_claude_normal_configuration_system_blocks(self): |
| 13 | request = {"system": [{"type": "text", "text": "instructions", "cache_control": {"type": "ephemeral"}}, {"type": "text", "text": "workspace context"}], "messages": [{"role": "user", "content": "task"}, {"role": "system", "content": "late system message"}]} |
| 14 | data = normalize_anthropic(request) |
| 15 | self.assertEqual(data["system"], "instructions\n\nworkspace context\n\nlate system message") |
| 16 | self.assertEqual(data["messages"], [{"role": "user", "content": "task"}]) |
| 17 | |
| 18 | def test_developer_messages_after_user_are_preserved_at_start(self): |
| 19 | request = {"instructions": "first", "input": [ |
| 20 | {"role": "developer", "content": [{"type": "input_text", "text": "second"}]}, |
| 21 | {"role": "user", "content": "task"}, |
| 22 | {"role": "developer", "content": "third"}, |
| 23 | ]} |
| 24 | data, _ = normalize(request) |
| 25 | self.assertEqual(data["input"], [ |
| 26 | {"role": "system", "content": "first\n\nsecond\n\nthird"}, |
| 27 | {"role": "user", "content": "task"}, |
| 28 | ]) |
| 29 | |
| 30 | def test_namespaced_tool_roundtrip_including_history(self): |
| 31 | request = {"tools": [{"type": "namespace", "name": "functions", "tools": [ |
| 32 | {"type": "function", "name": "exec_command", "parameters": {"type": "object"}}, |
| 33 | ]}], "input": [{"type": "function_call", "namespace": "functions", "name": "exec_command", "arguments": "{}", "call_id": "call_1"}]} |
| 34 | data, identities = normalize(request) |
| 35 | self.assertEqual(data["tools"][0]["name"], data["input"][0]["name"]) |
| 36 | self.assertEqual(restore(data["input"], identities), [{"type": "function_call", "namespace": "functions", "name": "exec_command", "arguments": "{}", "call_id": "call_1"}]) |
| 37 | |
| 38 | def test_raw_patch_text_survives_custom_tool_roundtrip(self): |
| 39 | text = "*** Begin Patch\n*** Add File: example.txt\n+hello\n*** End Patch" |
| 40 | request = {"tools": [{"type": "namespace", "name": "functions", "tools": [ |
| 41 | {"type": "custom", "name": "apply_patch", "format": {"type": "grammar", "syntax": "lark", "definition": "ignored by JSON adapter"}}, |
| 42 | ]}], "input": [{"type": "custom_tool_call", "namespace": "functions", "name": "apply_patch", "input": text, "call_id": "call_2"}, {"type": "custom_tool_call_output", "call_id": "call_2", "output": "ok"}]} |
| 43 | data, identities = normalize(request) |
| 44 | self.assertEqual(restore(data["input"][0], identities), {"type": "custom_tool_call", "namespace": "functions", "name": "apply_patch", "input": text, "call_id": "call_2"}) |
| 45 | self.assertEqual(data["input"][1]["type"], "function_call_output") |
| 46 | |
| 47 | def test_unsupported_server_tools_fail_explicitly(self): |
| 48 | with self.assertRaisesRegex(ValueError, "Unsupported Responses tool type"): |
| 49 | normalize({"tools": [{"type": "web_search"}], "input": "hi"}) |
| 50 | |
| 51 | |
| 52 | if __name__ == "__main__": |
| 53 | unittest.main() |